Multi-Agent Systems

Agent = (Beliefs, Desires, Intentions)

Beliefs:    What the agent knows about the world
Desires:    What the agent wants to achieve (goals)
Intentions: What the agent has committed to doing (plans)

# Classical agent: behavior defined in code
class ClassicalAgent:
    def decide(self, percepts: dict) -> str:
        if percepts["temperature"] > 30:
            return "turn_on_cooling"
        elif percepts["temperature"] < 18:
            return "turn_on_heating"
        return "do_nothing"

# LLM agent: behavior defined by prompt
LLM_AGENT_PROMPT = """You are a building climate control agent.
Given sensor readings, decide which action to take.
Consider energy efficiency, occupant comfort, and weather forecasts.
Explain your reasoning before stating your action."""

from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class Message:
    """A message between agents."""

    sender: str
    receiver: str
    content: str
    message_type: str = "inform"  # inform, request, propose, accept, reject
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    metadata: dict = field(default_factory=dict)


class MessageBus:
    """A simple message bus for agent communication."""

    def __init__(self):
        self.messages: list[Message] = []
        self.subscribers: dict[str, list] = {}  # agent_name -> callback list

    def send(self, message: Message) -> None:
        """Send a message to a specific agent."""
        self.messages.append(message)
        if message.receiver in self.subscribers:
            for callback in self.subscribers[message.receiver]:
                callback(message)

    def subscribe(self, agent_name: str, callback) -> None:
        """Subscribe an agent to receive messages."""
        if agent_name not in self.subscribers:
            self.subscribers[agent_name] = []
        self.subscribers[agent_name].append(callback)

    def get_history(
        self, agent_name: str | None = None, limit: int = 50
    ) -> list[Message]:
        """Get message history, optionally filtered by agent."""
        if agent_name:
            msgs = [
                m for m in self.messages
                if m.sender == agent_name or m.receiver == agent_name
            ]
        else:
            msgs = self.messages
        return msgs[-limit:]

import threading
from typing import Any


class SharedState:
    """Thread-safe shared state for multi-agent systems.

    All agents can read and write to the shared state, providing
    a common ground truth for coordination.
    """

    def __init__(self):
        self._state: dict[str, Any] = {}
        self._lock = threading.Lock()
        self._history: list[dict] = []

    def get(self, key: str, default: Any = None) -> Any:
        """Read a value from shared state."""
        with self._lock:
            return self._state.get(key, default)

    def set(self, key: str, value: Any, agent_name: str = "unknown") -> None:
        """Write a value to shared state."""
        with self._lock:
            old_value = self._state.get(key)
            self._state[key] = value
            self._history.append({
                "agent": agent_name,
                "key": key,
                "old_value": old_value,
                "new_value": value,
                "timestamp": datetime.now().isoformat(),
            })

    def get_all(self) -> dict[str, Any]:
        """Get a snapshot of the entire shared state."""
        with self._lock:
            return dict(self._state)

    def get_changes_by(self, agent_name: str) -> list[dict]:
        """Get all state changes made by a specific agent."""
        return [h for h in self._history if h["agent"] == agent_name]

class Blackboard:
    """Blackboard architecture for multi-agent collaboration.

    The blackboard is a shared workspace where agents post their
    contributions. A controller decides which agent should act next.
    """

    def __init__(self):
        self.entries: list[dict] = []
        self.agents: dict[str, dict] = {}  # name -> agent config

    def register_agent(self, name: str, expertise: list[str], agent_fn) -> None:
        """Register an agent with its areas of expertise."""
        self.agents[name] = {
            "expertise": expertise,
            "agent_fn": agent_fn,
        }

    def post(self, agent_name: str, content: str, entry_type: str = "contribution") -> None:
        """Post a contribution to the blackboard."""
        self.entries.append({
            "agent": agent_name,
            "content": content,
            "type": entry_type,
            "timestamp": datetime.now().isoformat(),
        })

    def get_entries(self, entry_type: str | None = None) -> list[dict]:
        """Get blackboard entries, optionally filtered by type."""
        if entry_type:
            return [e for e in self.entries if e["type"] == entry_type]
        return list(self.entries)

    def get_current_state(self) -> str:
        """Format the blackboard contents for an agent to read."""
        if not self.entries:
            return "The blackboard is empty."
        lines = ["Current blackboard state:"]
        for entry in self.entries:
            lines.append(f"  [{entry['agent']}] ({entry['type']}): {entry['content']}")
        return "\n".join(lines)

    def select_next_agent(self, task_description: str) -> str | None:
        """Select the most appropriate agent for the current task.

        Simple implementation: match task keywords to agent expertise.
        More sophisticated implementations would use an LLM.
        """
        task_lower = task_description.lower()
        best_match = None
        best_score = 0
        for name, config in self.agents.items():
            score = sum(1 for exp in config["expertise"] if exp.lower() in task_lower)
            if score > best_score:
                best_score = score
                best_match = name
        return best_match

Input -> [Agent A] -> [Agent B] -> [Agent C] -> Output

class SequentialPipeline:
    """Execute agents in sequence, passing output from one to the next."""

    def __init__(self):
        self.stages: list[tuple[str, callable]] = []

    def add_stage(self, name: str, agent_fn) -> None:
        """Add a stage to the pipeline."""
        self.stages.append((name, agent_fn))

    def run(self, initial_input: str) -> dict:
        """Run the pipeline from start to finish.

        Returns:
            Dict with the final output and intermediate results.
        """
        current_input = initial_input
        results = {"stages": [], "final_output": ""}

        for name, agent_fn in self.stages:
            print(f"Running stage: {name}")
            output = agent_fn(current_input)
            results["stages"].append({
                "name": name,
                "input_preview": current_input[:200],
                "output_preview": output[:200],
            })
            current_input = output

        results["final_output"] = current_input
        return results

              -> [Agent A] ->
Input -> Fork -> [Agent B] -> Merge -> Output
              -> [Agent C] ->

import asyncio
from typing import Callable


async def parallel_execution(
    task: str,
    agents: dict[str, Callable],
    merger: Callable,
) -> str:
    """Execute multiple agents in parallel and merge their outputs.

    Args:
        task: The shared task description.
        agents: Dict mapping agent names to their async callable functions.
        merger: Function that combines all agent outputs into a final result.

    Returns:
        The merged output from all agents.
    """
    # Run all agents concurrently
    async def run_agent(name: str, fn: Callable) -> tuple[str, str]:
        result = await fn(task)
        return name, result

    tasks = [run_agent(name, fn) for name, fn in agents.items()]
    results = await asyncio.gather(*tasks)

    # Merge results
    agent_outputs = {name: output for name, output in results}
    final = merger(task, agent_outputs)
    return final

class HierarchicalSystem:
    """A hierarchical multi-agent system with a supervisor and workers."""

    def __init__(self, supervisor_fn, workers: dict[str, callable]):
        self.supervisor_fn = supervisor_fn
        self.workers = workers

    def run(self, task: str, max_rounds: int = 10) -> str:
        """Run the hierarchical system.

        The supervisor decomposes the task, delegates to workers, and
        synthesizes results. Iteration continues until the supervisor
        declares the task complete.
        """
        conversation_history = [f"Task: {task}"]

        for round_num in range(max_rounds):
            # Supervisor decides what to do next
            supervisor_context = "\n".join(conversation_history)
            delegation = self.supervisor_fn(supervisor_context)

            if "TASK_COMPLETE" in delegation:
                # Extract final answer
                return delegation.replace("TASK_COMPLETE:", "").strip()

            # Parse delegation: which worker and what subtask
            worker_name, subtask = self._parse_delegation(delegation)

            if worker_name not in self.workers:
                conversation_history.append(
                    f"Error: No worker named '{worker_name}'. "
                    f"Available: {list(self.workers.keys())}"
                )
                continue

            # Execute the subtask
            worker_result = self.workers[worker_name](subtask)
            conversation_history.append(
                f"Supervisor delegated to {worker_name}: {subtask}"
            )
            conversation_history.append(
                f"{worker_name} result: {worker_result}"
            )

        return "Max rounds reached without completing the task."

    def _parse_delegation(self, delegation: str) -> tuple[str, str]:
        """Parse the supervisor's delegation into worker name and subtask."""
        # Simple format: "DELEGATE worker_name: subtask description"
        if "DELEGATE" in delegation:
            parts = delegation.split("DELEGATE", 1)[1].strip()
            if ":" in parts:
                worker, subtask = parts.split(":", 1)
                return worker.strip(), subtask.strip()
        return "", delegation

AGENT_PROMPTS = {
    "researcher": """You are a Research Agent. Your role is to gather, analyze,
and synthesize information. You are thorough, evidence-based, and skeptical of
unverified claims. When you find information, always note the source.

Your capabilities:
- Search the web for information
- Read and summarize documents
- Identify gaps in available information
- Assess the reliability of sources

You communicate clearly and distinguish between facts, inferences, and opinions.""",

    "developer": """You are a Software Developer Agent. Your role is to write,
review, and debug code. You follow best practices, write clean and well-documented
code, and consider edge cases.

Your capabilities:
- Write code in Python, JavaScript, and SQL
- Run and test code
- Review code for bugs and improvements
- Design software architecture

You explain your technical decisions and trade-offs clearly.""",

    "critic": """You are a Critical Review Agent. Your role is to identify flaws,
risks, and areas for improvement in proposals, plans, and outputs. You are
constructive but unflinching in your assessment.

Your capabilities:
- Identify logical fallacies and weak arguments
- Spot missing considerations and edge cases
- Evaluate feasibility and risks
- Suggest specific improvements

You always provide actionable feedback, not just criticism.""",
}

class SpecializedAgent:
    """An agent with a specific role and set of tools."""

    def __init__(
        self,
        name: str,
        system_prompt: str,
        tools: list[dict],
        model: str = "gpt-4",
    ):
        self.name = name
        self.system_prompt = system_prompt
        self.tools = tools
        self.model = model
        self.conversation_history: list[dict] = []

    def process(self, message: str) -> str:
        """Process a message and return a response.

        The agent uses its specialized prompt and tools.
        """
        self.conversation_history.append({"role": "user", "content": message})

        # In a real implementation, this calls the LLM API with
        # the system prompt, tools, and conversation history
        response = llm_call(
            model=self.model,
            system=self.system_prompt,
            messages=self.conversation_history,
            tools=self.tools,
        )

        self.conversation_history.append({"role": "assistant", "content": response})
        return response

# Conceptual example showing AutoGen's design pattern
# (simplified for educational purposes)

class ConversableAgent:
    """Simplified version of AutoGen's ConversableAgent."""

    def __init__(
        self,
        name: str,
        system_message: str,
        llm_call,
        max_consecutive_auto_reply: int = 10,
    ):
        self.name = name
        self.system_message = system_message
        self.llm_call = llm_call
        self.max_consecutive_auto_reply = max_consecutive_auto_reply
        self.chat_messages: dict[str, list[dict]] = {}

    def generate_reply(self, messages: list[dict]) -> str:
        """Generate a reply based on the conversation history."""
        full_messages = [{"role": "system", "content": self.system_message}]
        full_messages.extend(messages)
        return self.llm_call(messages=full_messages)

    def initiate_chat(self, recipient: "ConversableAgent", message: str) -> list[dict]:
        """Start a conversation with another agent.

        The agents take turns responding until a termination
        condition is met.
        """
        conversation = []
        current_message = message

        for turn in range(self.max_consecutive_auto_reply * 2):
            if turn % 2 == 0:
                # This agent's turn
                sender, receiver = self, recipient
            else:
                # Other agent's turn
                sender, receiver = recipient, self

            conversation.append({
                "role": "user" if sender == self else "assistant",
                "sender": sender.name,
                "content": current_message,
            })

            # Check for termination
            if "TERMINATE" in current_message:
                break

            # Generate reply
            current_message = receiver.generate_reply(
                [{"role": "user", "content": m["content"]} for m in conversation]
            )

        return conversation

class GroupChat:
    """A group chat between multiple agents.

    Manages turn-taking and message routing.
    """

    def __init__(
        self,
        agents: list[ConversableAgent],
        max_rounds: int = 20,
        speaker_selection: str = "round_robin",  # or "auto" or "random"
    ):
        self.agents = {agent.name: agent for agent in agents}
        self.max_rounds = max_rounds
        self.speaker_selection = speaker_selection
        self.conversation: list[dict] = []

    def select_next_speaker(self, last_speaker: str) -> str:
        """Select the next agent to speak."""
        agent_names = list(self.agents.keys())

        if self.speaker_selection == "round_robin":
            current_idx = agent_names.index(last_speaker)
            next_idx = (current_idx + 1) % len(agent_names)
            return agent_names[next_idx]

        elif self.speaker_selection == "random":
            import random
            candidates = [n for n in agent_names if n != last_speaker]
            return random.choice(candidates)

        else:  # "auto" - use LLM to decide
            return self._llm_select_speaker(last_speaker)

    def _llm_select_speaker(self, last_speaker: str) -> str:
        """Use an LLM to select the most appropriate next speaker."""
        agent_descriptions = "\n".join(
            f"- {name}: {agent.system_message[:100]}..."
            for name, agent in self.agents.items()
        )
        context = "\n".join(
            f"{m['sender']}: {m['content'][:200]}" for m in self.conversation[-5:]
        )
        # In practice, call the LLM to select the best next speaker
        # based on the conversation context and agent descriptions
        return list(self.agents.keys())[0]  # Placeholder

    def run(self, initial_message: str, initiator: str) -> list[dict]:
        """Run the group chat.

        Args:
            initial_message: The starting message.
            initiator: Name of the agent who starts the conversation.

        Returns:
            The complete conversation history.
        """
        self.conversation = [{
            "sender": initiator,
            "content": initial_message,
        }]

        current_speaker = initiator

        for round_num in range(self.max_rounds):
            # Select next speaker
            next_speaker = self.select_next_speaker(current_speaker)
            agent = self.agents[next_speaker]

            # Generate reply
            messages = [
                {"role": "user", "content": f"[{m['sender']}]: {m['content']}"}
                for m in self.conversation
            ]
            reply = agent.generate_reply(messages)

            self.conversation.append({
                "sender": next_speaker,
                "content": reply,
            })

            # Check for termination
            if "TERMINATE" in reply:
                break

            current_speaker = next_speaker

        return self.conversation

SUPERVISOR_PROMPT = """You are a workflow supervisor managing a team of specialized agents.

Available agents:
{agent_descriptions}

Given the user's request, decide which agent(s) to invoke and in what order.
You can invoke agents sequentially or gather information from multiple agents.

For each step, respond in this format:
INVOKE <agent_name>: <instruction for that agent>

When you have enough information to provide a final answer:
FINAL_ANSWER: <your synthesized response>
"""


class SupervisorOrchestrator:
    """Supervisor-based orchestration for multi-agent systems."""

    def __init__(self, supervisor_llm_call, agents: dict[str, callable]):
        self.supervisor_llm_call = supervisor_llm_call
        self.agents = agents
        self.execution_log: list[dict] = []

    def run(self, task: str, max_steps: int = 10) -> dict:
        """Execute a task using supervisor-guided orchestration."""
        agent_descriptions = "\n".join(
            f"- {name}" for name in self.agents.keys()
        )
        system = SUPERVISOR_PROMPT.format(agent_descriptions=agent_descriptions)
        context = f"User request: {task}\n\nExecution history:\n"

        for step in range(max_steps):
            # Ask supervisor what to do next
            full_context = context + "\n".join(
                f"Step {i+1}: Invoked {log['agent']} -> {log['result'][:200]}"
                for i, log in enumerate(self.execution_log)
            )

            decision = self.supervisor_llm_call(system=system, prompt=full_context)

            if "FINAL_ANSWER:" in decision:
                answer = decision.split("FINAL_ANSWER:", 1)[1].strip()
                return {
                    "answer": answer,
                    "steps": self.execution_log,
                    "num_steps": len(self.execution_log),
                }

            if "INVOKE" in decision:
                agent_name, instruction = self._parse_invocation(decision)
                if agent_name in self.agents:
                    result = self.agents[agent_name](instruction)
                    self.execution_log.append({
                        "agent": agent_name,
                        "instruction": instruction,
                        "result": result,
                    })

        return {
            "answer": "Max steps reached.",
            "steps": self.execution_log,
            "num_steps": len(self.execution_log),
        }

    def _parse_invocation(self, decision: str) -> tuple[str, str]:
        """Parse 'INVOKE agent_name: instruction' from supervisor output."""
        parts = decision.split("INVOKE", 1)[1].strip()
        if ":" in parts:
            agent_name, instruction = parts.split(":", 1)
            return agent_name.strip(), instruction.strip()
        return "", parts

ROUTER_PROMPT = """Given the current task state, select the best agent to handle the next step.

Current state:
{state}

Available agents:
{agents}

Select ONE agent by responding with just the agent name."""


def dynamic_router(
    state: str,
    agents: dict[str, str],  # name -> description
    llm_call,
) -> str:
    """Dynamically route to the best agent for the current state."""
    agents_str = "\n".join(f"- {name}: {desc}" for name, desc in agents.items())
    prompt = ROUTER_PROMPT.format(state=state, agents=agents_str)
    selected = llm_call(prompt=prompt).strip()
    # Validate selection
    if selected in agents:
        return selected
    # Fallback: find closest match
    for name in agents:
        if name.lower() in selected.lower():
            return name
    return list(agents.keys())[0]  # Default to first agent

[Researcher: wrong fact] -> [Writer: uses wrong fact] -> [Editor: fixes grammar but not facts]

def majority_vote(agent_responses: dict[str, str], llm_call) -> str:
    """Resolve disagreement through majority voting.

    Uses an LLM to classify responses into groups and select
    the majority position.
    """
    responses_text = "\n".join(
        f"{name}: {response}" for name, response in agent_responses.items()
    )

    prompt = f"""The following agents have provided different responses.
Group them by their position and identify the majority view.

{responses_text}

Majority position (summarize the view held by most agents):"""

    return llm_call(prompt=prompt)

class CostTracker:
    """Track and limit costs in a multi-agent system."""

    def __init__(self, max_tokens: int = 100000, max_rounds: int = 50):
        self.max_tokens = max_tokens
        self.max_rounds = max_rounds
        self.tokens_used = 0
        self.rounds_completed = 0

    def record_usage(self, tokens: int) -> None:
        self.tokens_used += tokens
        self.rounds_completed += 1

    def can_continue(self) -> bool:
        return (
            self.tokens_used < self.max_tokens
            and self.rounds_completed < self.max_rounds
        )

    def usage_report(self) -> dict:
        return {
            "tokens_used": self.tokens_used,
            "token_budget_remaining": self.max_tokens - self.tokens_used,
            "rounds_completed": self.rounds_completed,
            "rounds_remaining": self.max_rounds - self.rounds_completed,
        }

"""
A two-agent debate system: Proposer vs Critic, evaluated by a Judge.

This demonstrates:
- Role-based agent specialization
- Structured multi-turn interaction
- Adversarial collaboration for better reasoning
- Final evaluation by a third-party judge

Reference: Du et al. (2023) "Improving Factuality and Reasoning in
Language Models through Multiagent Debate"

Requirements:
    pip install openai  (or any LLM client)
"""

from dataclasses import dataclass, field


@dataclass
class DebateConfig:
    """Configuration for the debate system."""

    max_rounds: int = 3
    min_rounds: int = 2
    proposer_model: str = "gpt-4"
    critic_model: str = "gpt-4"
    judge_model: str = "gpt-4"


PROPOSER_SYSTEM = """You are a Proposer in a structured debate. Your role is to:
1. Present clear, well-reasoned arguments in favor of your position.
2. Support your claims with evidence, examples, and logical reasoning.
3. Respond thoughtfully to the Critic's objections.
4. Acknowledge valid criticisms and refine your position when warranted.

Be persuasive but honest. Do not make claims you cannot support.
If the Critic raises a valid point, concede it and adjust your argument."""

CRITIC_SYSTEM = """You are a Critic in a structured debate. Your role is to:
1. Carefully analyze the Proposer's arguments for logical flaws,
   unsupported claims, missing evidence, and counterexamples.
2. Raise specific, substantive objections (not just vague disagreement).
3. Suggest alternative viewpoints or interpretations.
4. Acknowledge strong arguments while pressing on weak ones.

Be rigorous but fair. Your goal is not to "win" but to strengthen
the overall quality of reasoning through critical examination.
Do not be contrarian for its own sake."""

JUDGE_SYSTEM = """You are the Judge evaluating a debate between a Proposer and a Critic.

Evaluate the debate on these criteria:
1. **Argument Quality**: How well-supported and logically sound were the arguments?
2. **Evidence Use**: Were claims backed by concrete evidence or examples?
3. **Responsiveness**: Did each side address the other's points effectively?
4. **Intellectual Honesty**: Did participants acknowledge valid opposing points?
5. **Final Assessment**: What is the most well-supported conclusion given the debate?

Provide:
- A score for each side (1-10)
- Key strengths and weaknesses of each side
- Your assessment of the most defensible position
- Any important points that neither side addressed"""


class DebateAgent:
    """An agent participating in a debate."""

    def __init__(self, name: str, system_prompt: str, llm_call, model: str = "gpt-4"):
        self.name = name
        self.system_prompt = system_prompt
        self.llm_call = llm_call
        self.model = model

    def respond(self, conversation_history: list[dict]) -> str:
        """Generate a response given the conversation history."""
        messages = [{"role": "system", "content": self.system_prompt}]
        messages.extend(conversation_history)
        return self.llm_call(model=self.model, messages=messages)


class DebateSystem:
    """Orchestrates a structured debate between a Proposer and a Critic."""

    def __init__(self, llm_call, config: DebateConfig | None = None):
        self.config = config or DebateConfig()
        self.llm_call = llm_call

        # Create agents
        self.proposer = DebateAgent(
            "Proposer", PROPOSER_SYSTEM, llm_call, self.config.proposer_model
        )
        self.critic = DebateAgent(
            "Critic", CRITIC_SYSTEM, llm_call, self.config.critic_model
        )
        self.judge = DebateAgent(
            "Judge", JUDGE_SYSTEM, llm_call, self.config.judge_model
        )

    def run_debate(self, topic: str) -> dict:
        """Run a complete debate on a topic.

        Args:
            topic: The debate topic or question.

        Returns:
            Dict with the debate transcript, judge's verdict, and metadata.
        """
        transcript: list[dict] = []
        proposer_history: list[dict] = []
        critic_history: list[dict] = []

        print(f"{'='*60}")
        print(f"DEBATE TOPIC: {topic}")
        print(f"{'='*60}\n")

        # Round 1: Proposer opens
        opening_prompt = f"Present your opening argument on the following topic: {topic}"
        proposer_history.append({"role": "user", "content": opening_prompt})
        opening = self.proposer.respond(proposer_history)
        proposer_history.append({"role": "assistant", "content": opening})

        transcript.append({
            "round": 1,
            "speaker": "Proposer",
            "type": "opening",
            "content": opening,
        })
        print(f"[Round 1 - Proposer Opening]\n{opening}\n")

        # Debate rounds
        for round_num in range(1, self.config.max_rounds + 1):
            # Critic responds
            critic_prompt = (
                f"The Proposer argues:\n\n{opening if round_num == 1 else proposer_response}\n\n"
                f"Provide your critique (Round {round_num})."
            )
            critic_history.append({"role": "user", "content": critic_prompt})
            critique = self.critic.respond(critic_history)
            critic_history.append({"role": "assistant", "content": critique})

            transcript.append({
                "round": round_num,
                "speaker": "Critic",
                "type": "critique",
                "content": critique,
            })
            print(f"[Round {round_num} - Critic]\n{critique}\n")

            # Proposer responds to critique
            rebuttal_prompt = (
                f"The Critic objects:\n\n{critique}\n\n"
                f"Respond to the criticism and refine your argument (Round {round_num})."
            )
            proposer_history.append({"role": "user", "content": rebuttal_prompt})
            proposer_response = self.proposer.respond(proposer_history)
            proposer_history.append({"role": "assistant", "content": proposer_response})

            transcript.append({
                "round": round_num,
                "speaker": "Proposer",
                "type": "rebuttal",
                "content": proposer_response,
            })
            print(f"[Round {round_num} - Proposer Rebuttal]\n{proposer_response}\n")

        # Judge evaluates
        debate_text = self._format_for_judge(transcript)
        judge_prompt = (
            f"Evaluate the following debate on the topic: '{topic}'\n\n{debate_text}"
        )
        verdict = self.judge.respond([{"role": "user", "content": judge_prompt}])

        transcript.append({
            "round": "final",
            "speaker": "Judge",
            "type": "verdict",
            "content": verdict,
        })
        print(f"{'='*60}")
        print(f"[JUDGE'S VERDICT]\n{verdict}")
        print(f"{'='*60}")

        return {
            "topic": topic,
            "transcript": transcript,
            "verdict": verdict,
            "num_rounds": self.config.max_rounds,
            "total_turns": len(transcript),
        }

    def _format_for_judge(self, transcript: list[dict]) -> str:
        """Format the debate transcript for the judge."""
        lines = []
        for entry in transcript:
            if entry["speaker"] != "Judge":
                lines.append(
                    f"[{entry['speaker']} - {entry['type'].title()}]\n{entry['content']}\n"
                )
        return "\n".join(lines)


# ── Usage Example ─────────────────────────────────────────────────

def main():
    """Run a sample debate."""

    # Mock LLM call for demonstration
    call_count = {"n": 0}

    def mock_llm_call(model: str = "gpt-4", messages: list[dict] = None, **kwargs) -> str:
        """Simulate LLM responses for the debate."""
        call_count["n"] += 1
        last_msg = messages[-1]["content"] if messages else ""

        if "opening argument" in last_msg.lower():
            return (
                "I argue that open-source AI models are essential for democratic "
                "AI development. First, they enable transparency: researchers and "
                "the public can inspect model weights, training data, and behavior. "
                "Second, they reduce concentration of power: when only a few companies "
                "control AI, they become gatekeepers of a transformative technology. "
                "Third, open-source accelerates innovation through community contributions, "
                "as seen with Linux and the web."
            )
        elif "critique" in last_msg.lower() or "critic" in messages[0].get("content", "").lower():
            return (
                "While transparency is valuable, the Proposer overlooks significant risks. "
                "Open-source AI models can be misused for generating disinformation, "
                "creating deepfakes, or automating cyberattacks. The 'democratization' "
                "argument assumes all actors are benign, which is naive. Furthermore, "
                "truly 'open' AI requires open training data and compute, not just weights. "
                "Most 'open-source' models are still controlled by companies that choose "
                "what to release."
            )
        elif "respond to the criticism" in last_msg.lower():
            return (
                "The Critic raises valid concerns about misuse. However, restricting access "
                "does not prevent misuse; it merely shifts who can misuse. Closed models can "
                "also be jailbroken. The solution is not secrecy but robust safety research, "
                "which itself benefits from open access. On the 'controlled openness' point, "
                "I concede that current open-source AI is imperfect, but the trajectory is "
                "toward greater openness, and this should be encouraged."
            )
        elif "evaluate" in last_msg.lower():
            return (
                "VERDICT:\n\n"
                "Proposer Score: 7/10\n"
                "- Strengths: Clear structure, good examples (Linux, web), acknowledged "
                "valid criticism about 'controlled openness'.\n"
                "- Weaknesses: Insufficient engagement with the misuse argument.\n\n"
                "Critic Score: 8/10\n"
                "- Strengths: Specific counterexamples, distinguished between types of "
                "openness, practical concern about misuse.\n"
                "- Weaknesses: Did not engage with the 'restricting access shifts misuse' "
                "counterargument.\n\n"
                "Assessment: The debate reveals that the question is not binary. The most "
                "defensible position is graduated openness with safety evaluation, combining "
                "the transparency benefits the Proposer champions with the risk mitigation "
                "the Critic demands."
            )
        return "I acknowledge the points raised and will consider them carefully."

    # Run the debate
    config = DebateConfig(max_rounds=2)
    debate = DebateSystem(mock_llm_call, config)
    result = debate.run_debate(
        "Should AI models be open-source? Consider safety, innovation, and power dynamics."
    )

    print(f"\nDebate completed: {result['total_turns']} total turns across {result['num_rounds']} rounds.")


if __name__ == "__main__":
    main()

{
  "name": "financial-analysis-agent",
  "description": "Analyzes financial reports, generates summaries, and answers questions about company financials.",
  "url": "https://finance-agent.example.com",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  },
  "skills": [
    {
      "id": "financial-report-analysis",
      "name": "Financial Report Analysis",
      "description": "Analyzes 10-K, 10-Q, and annual reports to extract key metrics and trends.",
      "inputModes": ["text/plain", "application/pdf"],
      "outputModes": ["text/plain", "application/json"]
    },
    {
      "id": "earnings-qa",
      "name": "Earnings Q&A",
      "description": "Answers questions about a company's earnings based on public filings.",
      "inputModes": ["text/plain"],
      "outputModes": ["text/plain"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

"""
Conceptual example of A2A agent discovery and task delegation.

This illustrates how a client agent discovers a remote agent via its
Agent Card and delegates a task using the A2A protocol.
"""

import httpx


class A2AClient:
    """A client for interacting with A2A-compatible agents."""

    def __init__(self, base_url: str, auth_token: str | None = None):
        self.base_url = base_url.rstrip("/")
        self.auth_token = auth_token
        self.agent_card: dict | None = None

    def _headers(self) -> dict:
        headers = {"Content-Type": "application/json"}
        if self.auth_token:
            headers["Authorization"] = f"Bearer {self.auth_token}"
        return headers

    async def discover(self) -> dict:
        """Discover the remote agent by fetching its Agent Card."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/.well-known/agent.json",
                headers=self._headers(),
            )
            response.raise_for_status()
            self.agent_card = response.json()
            return self.agent_card

    def has_skill(self, skill_id: str) -> bool:
        """Check if the remote agent has a specific skill."""
        if not self.agent_card:
            raise RuntimeError("Call discover() first to fetch the Agent Card.")
        return any(
            s["id"] == skill_id for s in self.agent_card.get("skills", [])
        )

    async def send_task(self, message_text: str, task_id: str | None = None) -> dict:
        """Send a task to the remote agent.

        Args:
            message_text: The task instruction in natural language.
            task_id: Optional task ID (generated by server if omitted).

        Returns:
            The task object with its current status.
        """
        payload = {
            "jsonrpc": "2.0",
            "method": "tasks/send",
            "params": {
                "message": {
                    "role": "user",
                    "parts": [{"type": "text", "text": message_text}],
                },
            },
            "id": task_id or "auto",
        }

        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/a2a",
                json=payload,
                headers=self._headers(),
            )
            response.raise_for_status()
            return response.json()["result"]

    async def get_task_status(self, task_id: str) -> dict:
        """Poll the status of a previously submitted task."""
        payload = {
            "jsonrpc": "2.0",
            "method": "tasks/get",
            "params": {"id": task_id},
            "id": "status-check",
        }

        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/a2a",
                json=payload,
                headers=self._headers(),
            )
            response.raise_for_status()
            return response.json()["result"]


async def orchestrate_financial_analysis():
    """Example: A supervisor agent discovers and delegates to a financial agent."""

    # Step 1: Discover the remote agent
    finance_agent = A2AClient(
        base_url="https://finance-agent.example.com",
        auth_token="secret-token",
    )
    card = await finance_agent.discover()
    print(f"Discovered agent: {card['name']}")
    print(f"Skills: {[s['name'] for s in card['skills']]}")

    # Step 2: Check if the agent has the skill we need
    if not finance_agent.has_skill("financial-report-analysis"):
        print("Agent does not have the required skill. Searching for another agent...")
        return

    # Step 3: Send the task
    task = await finance_agent.send_task(
        "Analyze the Q3 2025 earnings report for Acme Corp. "
        "Focus on revenue growth, profit margins, and forward guidance."
    )
    print(f"Task created: {task['id']}, status: {task['status']}")

    # Step 4: Poll for completion (in production, use SSE streaming)
    import asyncio
    while task["status"] in ("submitted", "working"):
        await asyncio.sleep(2)
        task = await finance_agent.get_task_status(task["id"])
        print(f"Task status: {task['status']}")

    # Step 5: Process results
    if task["status"] == "completed":
        for artifact in task.get("artifacts", []):
            print(f"Received artifact: {artifact.get('name', 'unnamed')}")
            for part in artifact.get("parts", []):
                if "text" in part:
                    print(f"  Content: {part['text'][:200]}...")
    elif task["status"] == "input-required":
        print("Agent needs more information:")
        last_message = task.get("messages", [])[-1]
        print(f"  {last_message}")

"""
Conceptual example of ACP agent communication within an enterprise system.
"""


class ACPAgentDirectory:
    """A centralized directory for ACP agent discovery."""

    def __init__(self):
        self.agents: dict[str, dict] = {}

    def register(self, agent_id: str, metadata: dict) -> None:
        """Register an agent in the directory."""
        self.agents[agent_id] = {
            **metadata,
            "status": "available",
        }

    def discover(self, required_skill: str) -> list[dict]:
        """Find agents that have a specific skill."""
        return [
            {"id": agent_id, **info}
            for agent_id, info in self.agents.items()
            if required_skill in info.get("skills", [])
            and info["status"] == "available"
        ]

    def update_status(self, agent_id: str, status: str) -> None:
        """Update an agent's availability status."""
        if agent_id in self.agents:
            self.agents[agent_id]["status"] = status


class ACPMessage:
    """A message in the ACP protocol."""

    def __init__(self, thread_id: str, sender: str, content: list[dict]):
        self.thread_id = thread_id
        self.sender = sender
        self.content = content  # List of parts: text, file, data


class ACPAgent:
    """An agent that communicates via ACP."""

    def __init__(self, agent_id: str, skills: list[str], directory: ACPAgentDirectory):
        self.agent_id = agent_id
        self.skills = skills
        self.directory = directory
        self.inbox: list[ACPMessage] = []

        # Register with the directory on creation
        self.directory.register(agent_id, {
            "skills": skills,
            "description": f"Agent with skills: {', '.join(skills)}",
        })

    async def send_to(self, recipient_id: str, thread_id: str, content: list[dict]) -> None:
        """Send a message to another agent."""
        message = ACPMessage(
            thread_id=thread_id,
            sender=self.agent_id,
            content=content,
        )
        # In a real implementation, this would go through a message broker
        print(f"[ACP] {self.agent_id} -> {recipient_id}: {content}")

    async def request_work(self, skill: str, instruction: str) -> str | None:
        """Find an agent with a skill and request work."""
        candidates = self.directory.discover(skill)
        if not candidates:
            print(f"No available agent found with skill: {skill}")
            return None

        target = candidates[0]
        target_id = target["id"]
        self.directory.update_status(target_id, "busy")

        await self.send_to(
            recipient_id=target_id,
            thread_id=f"thread-{self.agent_id}-{target_id}",
            content=[{"type": "text", "text": instruction}],
        )
        return target_id
python
# Conceptual example of OpenAI Agents SDK pattern
# (simplified for educational purposes)

class Agent:
    def __init__(self, name: str, instructions: str, tools: list = None, handoffs: list = None):
        self.name = name
        self.instructions = instructions
        self.tools = tools or []
        self.handoffs = handoffs or []  # List of other Agents

class HandoffExample:
    """Demonstrates the handoff pattern."""

    def build_triage_system(self):
        """Build a customer service triage system with handoffs."""

        billing_agent = Agent(
            name="Billing Specialist",
            instructions="You handle billing questions: invoices, payments, refunds.",
            tools=[lookup_invoice, process_refund],
        )

        technical_agent = Agent(
            name="Technical Support",
            instructions="You handle technical issues: bugs, configuration, integration.",
            tools=[search_docs, create_ticket],
        )

        triage_agent = Agent(
            name="Triage Agent",
            instructions=(
                "You are the first point of contact. Determine whether the customer "
                "needs billing help or technical support, then hand off to the "
                "appropriate specialist."
            ),
            handoffs=[billing_agent, technical_agent],
        )

        return triage_agent

# Conceptual example of Claude Agent SDK pattern
# (simplified for educational purposes)

import anthropic


def build_research_agent():
    """Build a research agent using Claude Agent SDK patterns."""
    client = anthropic.Anthropic()

    tools = [
        {
            "name": "search_papers",
            "description": "Search academic papers by query.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "max_results": {"type": "integer", "default": 5},
                },
                "required": ["query"],
            },
        },
        {
            "name": "read_pdf",
            "description": "Extract text from a PDF document.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL of the PDF"},
                },
                "required": ["url"],
            },
        },
    ]

    # The agentic loop: send a message, process tool calls, repeat
    messages = [{"role": "user", "content": "Find recent papers on multi-agent debate."}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system="You are a research assistant. Use tools to find and analyze papers.",
            tools=tools,
            messages=messages,
        )

        # Check if the model wants to use a tool
        if response.stop_reason == "tool_use":
            # Execute tool calls and feed results back
            tool_results = execute_tools(response.content)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            # Model produced a final response
            return response.content

# Conceptual example of LangGraph pattern
# (simplified for educational purposes)

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    papers: list[dict]
    analysis: str
    needs_more_research: bool
    final_report: str


def search_node(state: ResearchState) -> dict:
    """Search for relevant papers."""
    papers = search_papers(state["query"])
    return {"papers": papers}


def analyze_node(state: ResearchState) -> dict:
    """Analyze the collected papers."""
    analysis = analyze_papers(state["papers"])
    needs_more = len(state["papers"]) < 5
    return {"analysis": analysis, "needs_more_research": needs_more}


def report_node(state: ResearchState) -> dict:
    """Generate the final report."""
    report = generate_report(state["analysis"], state["papers"])
    return {"final_report": report}


def should_continue(state: ResearchState) -> str:
    """Decide whether to search more or write the report."""
    if state["needs_more_research"]:
        return "search"  # Loop back to search
    return "report"  # Proceed to report generation


# Build the graph
# graph = StateGraph(ResearchState)
# graph.add_node("search", search_node)
# graph.add_node("analyze", analyze_node)
# graph.add_node("report", report_node)
# graph.set_entry_point("search")
# graph.add_edge("search", "analyze")
# graph.add_conditional_edges("analyze", should_continue, {"search": "search", "report": "report"})
# app = graph.compile()

# Conceptual example of CrewAI pattern
# (simplified for educational purposes)

class CrewAgent:
    def __init__(self, role: str, goal: str, backstory: str, tools: list = None):
        self.role = role
        self.goal = goal
        self.backstory = backstory
        self.tools = tools or []

class Task:
    def __init__(self, description: str, expected_output: str, agent: CrewAgent):
        self.description = description
        self.expected_output = expected_output
        self.agent = agent

# Define agents
researcher = CrewAgent(
    role="Senior Research Analyst",
    goal="Discover and analyze the latest trends in agentic AI.",
    backstory="A veteran AI researcher with 15 years of experience in multi-agent systems.",
)

writer = CrewAgent(
    role="Technical Writer",
    goal="Transform research findings into clear, engaging content.",
    backstory="A science communicator skilled at making complex topics accessible.",
)

# Define tasks
research_task = Task(
    description="Research the latest developments in agent communication protocols (A2A, ACP, MCP).",
    expected_output="A comprehensive summary with key findings and comparisons.",
    agent=researcher,
)

writing_task = Task(
    description="Write a blog post based on the research findings.",
    expected_output="A 1500-word blog post suitable for a technical audience.",
    agent=writer,
)

# The crew would execute: research_task -> writing_task