Multi-Agent Systems -- תזמור בקנה מידה

from anthropic import Anthropic

client = Anthropic()

def run_agent(system_prompt: str, user_input: str, model: str = "claude-sonnet-4-20250514") -> str:
    """Run a single agent and return its response."""
    response = client.messages.create(
        model=model,
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": user_input}]
    )
    return response.content[0].text

def content_pipeline(topic: str) -> dict:
    """Sequential pipeline: Research -> Write -> Edit."""

    # Agent 1: Researcher
    research = run_agent(
        system_prompt="You are a research specialist. Gather key facts, "
                      "statistics, and insights about the given topic. "
                      "Output a structured research brief.",
        user_input=f"Research the following topic thoroughly: {topic}"
    )
    print(f"[Researcher] Done: {len(research)} chars")

    # Agent 2: Writer
    draft = run_agent(
        system_prompt="You are a professional content writer. Using the "
                      "research brief provided, write a compelling article. "
                      "Use clear structure, engaging tone, and examples.",
        user_input=f"Write an article based on this research:\n\n{research}"
    )
    print(f"[Writer] Done: {len(draft)} chars")

    # Agent 3: Editor
    final = run_agent(
        system_prompt="You are a meticulous editor. Review the article for "
                      "clarity, accuracy, grammar, and flow. Fix any issues "
                      "and return the polished final version.",
        user_input=f"Edit and improve this article:\n\n{draft}"
    )
    print(f"[Editor] Done: {len(final)} chars")

    return {
        "research": research,
        "draft": draft,
        "final": final
    }

# Usage
result = content_pipeline("AI agents in Israeli startups 2026")
print(result["final"])

import { generateText } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';

async function runAgent(
  systemPrompt: string,
  userInput: string,
  model = 'claude-sonnet-4-20250514'
): Promise<string> {
  const { text } = await generateText({
    model: anthropic(model),
    system: systemPrompt,
    prompt: userInput,
    maxTokens: 2048,
  });
  return text;
}

async function contentPipeline(topic: string) {
  // Agent 1: Researcher
  const research = await runAgent(
    'You are a research specialist. Gather key facts and insights.',
    `Research the following topic thoroughly: ${topic}`
  );
  console.log(`[Researcher] Done: ${research.length} chars`);

  // Agent 2: Writer
  const draft = await runAgent(
    'You are a professional content writer. Write a compelling article.',
    `Write an article based on this research:\n\n${research}`
  );
  console.log(`[Writer] Done: ${draft.length} chars`);

  // Agent 3: Editor
  const final = await runAgent(
    'You are a meticulous editor. Polish the article for clarity and accuracy.',
    `Edit and improve this article:\n\n${draft}`
  );
  console.log(`[Editor] Done: ${final.length} chars`);

  return { research, draft, final: final };
}

// Usage
const result = await contentPipeline('AI agents in Israeli startups 2026');

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def run_agent_async(name: str, system_prompt: str, user_input: str) -> dict:
    """Run a single agent asynchronously."""
    response = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_input}]
    )
    result = response.content[0].text
    print(f"[{name}] Done")
    return {"agent": name, "result": result}

async def parallel_research(query: str) -> list[dict]:
    """Fan-out: search multiple sources in parallel."""
    agents = [
        ("Market Analyst", "You analyze market trends and competitive landscape."),
        ("Tech Researcher", "You analyze technology stacks and technical feasibility."),
        ("Financial Analyst", "You analyze costs, revenue models, and financial viability."),
        ("User Researcher", "You analyze user needs, pain points, and demand signals."),
    ]

    tasks = [
        run_agent_async(name, prompt, f"Analyze: {query}")
        for name, prompt in agents
    ]

    # All agents run in parallel
    results = await asyncio.gather(*tasks)
    return results

async def fan_out_with_aggregation(query: str) -> str:
    """Fan-out + aggregation: parallel research then synthesize."""
    # Step 1: Fan-out (parallel)
    results = await parallel_research(query)

    # Step 2: Aggregate (single agent synthesizes all results)
    combined = "\n\n".join(
        f"## {r['agent']}\n{r['result']}" for r in results
    )

    synthesis = await run_agent_async(
        "Synthesizer",
        "You synthesize multiple research reports into a coherent summary "
        "with key insights and recommendations.",
        f"Synthesize these reports:\n\n{combined}"
    )
    return synthesis["result"]

# Usage
result = asyncio.run(fan_out_with_aggregation("Building an AI tutoring platform for Israel"))
print(result)

import { generateText } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';

interface AgentResult {
  agent: string;
  result: string;
}

async function runAgentAsync(
  name: string, systemPrompt: string, userInput: string
): Promise<AgentResult> {
  const { text } = await generateText({
    model: anthropic('claude-sonnet-4-20250514'),
    system: systemPrompt,
    prompt: userInput,
  });
  console.log(`[${name}] Done`);
  return { agent: name, result: text };
}

async function parallelResearch(query: string): Promise<string> {
  const agents = [
    { name: 'Market Analyst', prompt: 'You analyze market trends.' },
    { name: 'Tech Researcher', prompt: 'You analyze technology feasibility.' },
    { name: 'Financial Analyst', prompt: 'You analyze costs and revenue.' },
    { name: 'User Researcher', prompt: 'You analyze user needs and demand.' },
  ];

  // All agents run in parallel with Promise.all
  const results = await Promise.all(
    agents.map(a => runAgentAsync(a.name, a.prompt, `Analyze: ${query}`))
  );

  // Synthesize
  const combined = results.map(r => `## ${r.agent}\n${r.result}`).join('\n\n');
  const synthesis = await runAgentAsync(
    'Synthesizer',
    'Synthesize multiple reports into a coherent summary.',
    `Synthesize:\n\n${combined}`
  );

  return synthesis.result;
}

import json
from anthropic import Anthropic

client = Anthropic()

WORKERS = {
    "developer": {
        "description": "Writes code, fixes bugs, implements features",
        "system": "You are an expert software developer. Write clean, tested code.",
    },
    "designer": {
        "description": "Creates UI/UX designs, wireframes, style guides",
        "system": "You are a UI/UX designer. Create clear, user-friendly designs.",
    },
    "tester": {
        "description": "Writes tests, finds bugs, validates functionality",
        "system": "You are a QA engineer. Write thorough tests and find edge cases.",
    },
    "documenter": {
        "description": "Writes docs, README files, API documentation",
        "system": "You are a technical writer. Write clear, comprehensive docs.",
    },
}

def supervisor_loop(task: str, max_steps: int = 6) -> list[dict]:
    """Supervisor dynamically assigns tasks to workers."""
    worker_list = "\n".join(
        f"- {name}: {w['description']}" for name, w in WORKERS.items()
    )

    history = []
    supervisor_messages = []

    for step in range(max_steps):
        supervisor_prompt = f"""You are a project manager. Your team:
{worker_list}

Task: {task}

Work done so far:
{json.dumps(history, indent=2) if history else "Nothing yet."}

Decide the next action. Respond with JSON:
{{"action": "delegate", "worker": "", "instruction": ""}}
or if the task is complete:
{{"action": "complete", "summary": ""}}"""

        supervisor_messages.append({"role": "user", "content": supervisor_prompt})
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system="You are a project supervisor. Respond with valid JSON only.",
            messages=supervisor_messages
        )

        decision_text = response.content[0].text
        supervisor_messages.append({"role": "assistant", "content": decision_text})

        try:
            decision = json.loads(decision_text)
        except json.JSONDecodeError:
            break

        if decision["action"] == "complete":
            print(f"[Supervisor] Task complete: {decision['summary']}")
            history.append({"step": step + 1, "action": "complete", "summary": decision["summary"]})
            break

        worker_name = decision["worker"]
        instruction = decision["instruction"]
        print(f"[Supervisor] Step {step+1}: Delegating to {worker_name}")

        # Run the worker
        worker = WORKERS[worker_name]
        worker_response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            system=worker["system"],
            messages=[{"role": "user", "content": instruction}]
        )
        result = worker_response.content[0].text
        history.append({
            "step": step + 1, "worker": worker_name,
            "instruction": instruction,
            "result": result[:500]  # Truncate for context management
        })

    return history

# Usage
results = supervisor_loop("Build a REST API for a todo app with tests and docs")

from anthropic import Anthropic
import json

client = Anthropic()

AGENTS = {
    "triage": {
        "system": """You are a customer service triage agent. Analyze the customer's
request and decide which department handles it best.
Respond with JSON: {"handoff_to": "billing|technical|sales|none", "reason": "..."}
If you can answer directly (simple greetings, general info), use "none".""",
        "can_handoff_to": ["billing", "technical", "sales"]
    },
    "billing": {
        "system": """You are a billing specialist. Handle payment, invoice, and
account balance questions. If the issue is technical, hand off.
To hand off, include in your response: HANDOFF:technical or HANDOFF:triage""",
        "can_handoff_to": ["technical", "triage"]
    },
    "technical": {
        "system": """You are a technical support specialist. Handle product issues,
bugs, and how-to questions. If it's a billing issue, hand off.
To hand off, include: HANDOFF:billing or HANDOFF:triage""",
        "can_handoff_to": ["billing", "triage"]
    },
    "sales": {
        "system": """You are a sales specialist. Handle pricing questions, plan
upgrades, and new feature inquiries. Hand off if needed.
To hand off, include: HANDOFF:billing or HANDOFF:triage""",
        "can_handoff_to": ["billing", "triage"]
    },
}

def swarm_conversation(user_message: str, max_handoffs: int = 3) -> str:
    """Process a customer request through the swarm."""
    current_agent = "triage"
    handoff_count = 0
    context = f"Customer message: {user_message}"

    while handoff_count <= max_handoffs:
        agent = AGENTS[current_agent]
        print(f"[{current_agent.upper()}] Processing...")

        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=agent["system"],
            messages=[{"role": "user", "content": context}]
        )
        result = response.content[0].text

        # Check for handoff
        if "HANDOFF:" in result:
            target = result.split("HANDOFF:")[1].strip().split()[0]
            if target in agent["can_handoff_to"]:
                print(f"[{current_agent.upper()}] Handing off to {target}")
                current_agent = target
                context = f"Handed off from previous agent.\nOriginal: {user_message}\nContext: {result}"
                handoff_count += 1
                continue

        # Triage returns JSON
        if current_agent == "triage":
            try:
                decision = json.loads(result)
                if decision["handoff_to"] != "none":
                    current_agent = decision["handoff_to"]
                    context = f"Routed by triage: {decision['reason']}\nCustomer: {user_message}"
                    handoff_count += 1
                    continue
            except (json.JSONDecodeError, KeyError):
                pass

        return result

    return "I apologize, let me connect you with a human agent."

# Usage
print(swarm_conversation("I was charged twice for my subscription last month"))
print(swarm_conversation("How do I reset my password?"))

from anthropic import Anthropic

client = Anthropic()

def debate(topic: str, rounds: int = 3) -> str:
    """Two agents debate a topic. A judge synthesizes the best answer."""

    writer_history = []
    critic_history = []

    for round_num in range(1, rounds + 1):
        print(f"\n--- Round {round_num} ---")

        # Writer produces or revises
        if round_num == 1:
            writer_input = f"Write a thorough analysis of: {topic}"
        else:
            writer_input = (
                f"The critic said:\n{criticism}\n\n"
                f"Revise your analysis to address these points."
            )

        writer_history.append({"role": "user", "content": writer_input})
        writer_resp = client.messages.create(
            model="claude-sonnet-4-20250514", max_tokens=2048,
            system="You are an expert analyst. Write accurate, well-sourced analysis.",
            messages=writer_history
        )
        analysis = writer_resp.content[0].text
        writer_history.append({"role": "assistant", "content": analysis})
        print(f"[Writer] {len(analysis)} chars")

        # Critic reviews
        critic_input = (
            f"Critically review this analysis. Find errors, gaps, weak arguments, "
            f"and unsupported claims:\n\n{analysis}"
        )
        critic_history.append({"role": "user", "content": critic_input})
        critic_resp = client.messages.create(
            model="claude-sonnet-4-20250514", max_tokens=1024,
            system="You are a rigorous critic and fact-checker. Find every flaw.",
            messages=critic_history
        )
        criticism = critic_resp.content[0].text
        critic_history.append({"role": "assistant", "content": criticism})
        print(f"[Critic] Found issues")

    # Judge synthesizes the final answer
    judge_input = (
        f"Two experts debated '{topic}' for {rounds} rounds.\n\n"
        f"Final analysis:\n{analysis}\n\n"
        f"Final criticism:\n{criticism}\n\n"
        f"Synthesize the best possible answer, incorporating valid criticisms."
    )
    judge_resp = client.messages.create(
        model="claude-sonnet-4-20250514", max_tokens=2048,
        system="You are an impartial judge. Synthesize the strongest answer.",
        messages=[{"role": "user", "content": judge_input}]
    )
    return judge_resp.content[0].text

# Usage
result = debate("Will AI agents replace most SaaS products by 2028?", rounds=3)
print(result)

from dataclasses import dataclass, field
from typing import Any

@dataclass
class PipelineState:
    """Shared state accessible by all agents."""
    task: str = ""
    research: str = ""
    draft: str = ""
    review: str = ""
    final: str = ""
    metadata: dict = field(default_factory=dict)
    errors: list[str] = field(default_factory=list)
    status: str = "pending"  # pending, in_progress, complete, failed

    def log(self, agent: str, message: str):
        """Log an action to shared state."""
        if "log" not in self.metadata:
            self.metadata["log"] = []
        self.metadata["log"].append({"agent": agent, "message": message})

# Each agent reads from and writes to the same state
def researcher(state: PipelineState) -> PipelineState:
    state.status = "in_progress"
    state.log("researcher", "Starting research")
    # ... call LLM, write to state.research ...
    state.research = "Research results here..."
    state.log("researcher", "Research complete")
    return state

def writer(state: PipelineState) -> PipelineState:
    state.log("writer", "Starting draft")
    # ... use state.research to write draft ...
    state.draft = f"Article based on: {state.research[:100]}..."
    state.log("writer", "Draft complete")
    return state

interface AgentMessage {
  from: string;
  to: string;
  type: 'task' | 'result' | 'error' | 'handoff';
  payload: any;
  correlationId: string;
  timestamp: Date;
}

class MessageBus {
  private queues: Map<string, AgentMessage[]> = new Map();
  private listeners: Map<string, ((msg: AgentMessage) => void)[]> = new Map();

  send(message: AgentMessage): void {
    const queue = this.queues.get(message.to) || [];
    queue.push(message);
    this.queues.set(message.to, queue);

    // Notify listeners
    const handlers = this.listeners.get(message.to) || [];
    handlers.forEach(handler => handler(message));
  }

  receive(agentName: string): AgentMessage | undefined {
    const queue = this.queues.get(agentName) || [];
    return queue.shift();
  }

  subscribe(agentName: string, handler: (msg: AgentMessage) => void): void {
    const handlers = this.listeners.get(agentName) || [];
    handlers.push(handler);
    this.listeners.set(agentName, handlers);
  }
}

// Usage
const bus = new MessageBus();
const correlationId = crypto.randomUUID();

bus.send({
  from: 'orchestrator',
  to: 'researcher',
  type: 'task',
  payload: { topic: 'AI agents market in Israel' },
  correlationId,
  timestamp: new Date(),
});

from dataclasses import dataclass, field
from typing import Any
import threading

@dataclass
class MultiAgentState:
    """Thread-safe shared state with per-agent namespaces."""

    # Per-agent namespaces -- no conflicts
    agent_outputs: dict[str, Any] = field(default_factory=dict)

    # Shared append-only log
    messages: list[dict] = field(default_factory=list)

    # Lock for thread-safe writes
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def set_output(self, agent_name: str, key: str, value: Any):
        """Each agent writes to its own namespace."""
        with self._lock:
            if agent_name not in self.agent_outputs:
                self.agent_outputs[agent_name] = {}
            self.agent_outputs[agent_name][key] = value

    def get_output(self, agent_name: str, key: str) -> Any:
        """Read any agent's output."""
        return self.agent_outputs.get(agent_name, {}).get(key)

    def add_message(self, from_agent: str, content: str):
        """Append-only message log -- never loses data."""
        with self._lock:
            self.messages.append({
                "from": from_agent,
                "content": content,
                "index": len(self.messages)
            })

    def get_all_outputs(self) -> dict:
        """Get a snapshot of all agent outputs."""
        with self._lock:
            return dict(self.agent_outputs)

# Usage
state = MultiAgentState()
state.set_output("researcher", "findings", "Found 3 key trends...")
state.set_output("analyst", "recommendation", "Invest in vertical AI...")
state.add_message("researcher", "Research phase complete")
state.add_message("analyst", "Analysis phase complete")

# No conflicts -- each agent has its own namespace
print(state.get_output("researcher", "findings"))
print(state.get_all_outputs())

from anthropic import Anthropic

client = Anthropic()

MAX_DELEGATION_DEPTH = 3  # Prevent infinite delegation chains

def specialist_agent(
    name: str,
    system_prompt: str,
    task: str,
    available_specialists: dict[str, str],
    depth: int = 0
) -> str:
    """Agent that can delegate to specialists, with depth limit."""

    if depth >= MAX_DELEGATION_DEPTH:
        # Escalate to human instead of delegating further
        return f"[ESCALATION] Task too complex after {depth} delegations. Need human review: {task}"

    specialist_list = "\n".join(
        f"- {k}: {v}" for k, v in available_specialists.items()
    )

    full_prompt = f"""{system_prompt}

You can delegate subtasks to specialists:
{specialist_list}

To delegate, respond with: DELEGATE:specialist_name:instruction
To answer directly, just respond normally.
Current delegation depth: {depth}/{MAX_DELEGATION_DEPTH}"""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        system=full_prompt,
        messages=[{"role": "user", "content": task}]
    )
    result = response.content[0].text

    # Check for delegation
    if "DELEGATE:" in result:
        parts = result.split("DELEGATE:")[1].strip().split(":", 1)
        target = parts[0].strip()
        instruction = parts[1].strip() if len(parts) > 1 else task

        if target in available_specialists:
            print(f"[{name}] Delegating to {target} (depth {depth+1})")
            sub_result = specialist_agent(
                name=target,
                system_prompt=available_specialists[target],
                task=instruction,
                available_specialists=available_specialists,
                depth=depth + 1
            )
            # Agent can use the sub-result
            return f"[{name}] used [{target}]'s work:\n{sub_result}"

    return f"[{name}]: {result}"

# Usage
specialists = {
    "coder": "You write Python code. Clean, tested, documented.",
    "reviewer": "You review code for bugs, security, and best practices.",
    "documenter": "You write clear technical documentation.",
}

result = specialist_agent(
    name="project_lead",
    system_prompt="You are a project lead. Break tasks into subtasks and delegate.",
    task="Create a function to validate Israeli phone numbers with tests and docs",
    available_specialists=specialists
)
print(result)

import { generateText } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';

interface Specialist {
  name: string;
  description: string;
  systemPrompt: string;
  model: string;  // Different models per specialist
}

const specialists: Specialist[] = [
  {
    name: 'code_expert',
    description: 'Programming questions, debugging, code review',
    systemPrompt: 'You are an expert programmer. Write clean, tested code.',
    model: 'claude-sonnet-4-20250514',
  },
  {
    name: 'data_analyst',
    description: 'Data analysis, SQL, visualization, statistics',
    systemPrompt: 'You are a data analyst. Provide clear, data-driven insights.',
    model: 'claude-sonnet-4-20250514',
  },
  {
    name: 'writer',
    description: 'Content writing, editing, copywriting, translation',
    systemPrompt: 'You are a professional writer. Create compelling content.',
    model: 'claude-haiku-4-20250514',  // Cheaper model for writing
  },
];

async function triageAndRoute(userQuery: string): Promise<string> {
  // Step 1: Triage with cheap model
  const specialistList = specialists
    .map(s => `- ${s.name}: ${s.description}`)
    .join('\n');

  const { text: routing } = await generateText({
    model: anthropic('claude-haiku-4-20250514'),  // Cheap triage
    system: `Route the user's request to the right specialist.
Available specialists:\n${specialistList}
Respond with just the specialist name, nothing else.`,
    prompt: userQuery,
  });

  const selected = specialists.find(s => s.name === routing.trim());
  if (!selected) {
    // Fallback to first specialist
    return triageAndRoute(userQuery);
  }

  console.log(`[Triage] Routed to: ${selected.name}`);

  // Step 2: Run specialist with appropriate model
  const { text: answer } = await generateText({
    model: anthropic(selected.model),
    system: selected.systemPrompt,
    prompt: userQuery,
  });

  return answer;
}

from claude_agent_sdk import Agent, Tool

# Define specialist agents
researcher = Agent(
    name="researcher",
    model="claude-sonnet-4-20250514",
    system="You are a research specialist. Find facts and data.",
    tools=[web_search_tool, database_tool]
)

writer = Agent(
    name="writer",
    model="claude-sonnet-4-20250514",
    system="You are a professional writer.",
    tools=[]  # No tools needed -- pure generation
)

# Orchestrator agent that can delegate to sub-agents
orchestrator = Agent(
    name="orchestrator",
    model="claude-sonnet-4-20250514",
    system="""You coordinate a team. Available sub-agents:
    - researcher: finds facts and data
    - writer: writes polished content
    Delegate tasks as needed.""",
    supported_agents=[researcher, writer]
)

# The orchestrator decides which sub-agent to call
result = orchestrator.run("Write an article about AI adoption in Israeli healthcare")

from agents import Agent, handoff, Runner

billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing questions. If technical, hand off to tech.",
    handoffs=["technical_agent"]
)

technical_agent = Agent(
    name="Technical Specialist",
    instructions="Handle technical issues. If billing, hand off to billing.",
    handoffs=["billing_agent"]
)

triage_agent = Agent(
    name="Triage",
    instructions="Route to the right specialist based on the question.",
    handoffs=[billing_agent, technical_agent]
)

# Run -- the triage agent routes automatically
result = await Runner.run(triage_agent, "I was charged twice for my subscription")
print(result.final_output)  # Handled by billing_agent

from google.adk.agents import (
    LlmAgent, SequentialAgent, ParallelAgent, LoopAgent
)

# Define individual agents
researcher = LlmAgent(name="researcher", model="gemini-2.5-pro",
    instruction="Research the given topic.")
writer = LlmAgent(name="writer", model="gemini-2.5-pro",
    instruction="Write an article from the research.")
reviewer = LlmAgent(name="reviewer", model="gemini-2.5-pro",
    instruction="Review and score the article 1-10.")

# Sequential: researcher -> writer -> reviewer
pipeline = SequentialAgent(
    name="content_pipeline",
    sub_agents=[researcher, writer, reviewer]
)

# Parallel: run multiple researchers at once
multi_research = ParallelAgent(
    name="parallel_research",
    sub_agents=[market_researcher, tech_researcher, user_researcher]
)

# Loop: keep refining until quality score >= 8
refinement_loop = LoopAgent(
    name="refinement",
    sub_agents=[writer, reviewer],
    max_iterations=3,
    stop_condition=lambda state: state.get("score", 0) >= 8
)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages

class TeamState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str

def supervisor_node(state: TeamState) -> dict:
    """Supervisor decides which worker to call next."""
    # Call LLM to decide routing
    response = llm.invoke(
        f"Given the conversation, who should act next? "
        f"Options: researcher, writer, reviewer, FINISH"
    )
    return {"next_agent": response.content}

def researcher_node(state: TeamState) -> dict:
    result = llm.invoke("Research: " + state["messages"][-1].content)
    return {"messages": [{"role": "assistant", "content": f"[Research] {result.content}"}]}

def writer_node(state: TeamState) -> dict:
    result = llm.invoke("Write article from: " + state["messages"][-1].content)
    return {"messages": [{"role": "assistant", "content": f"[Writer] {result.content}"}]}

def route(state: TeamState) -> Literal["researcher", "writer", "reviewer", "__end__"]:
    if state["next_agent"] == "FINISH":
        return "__end__"
    return state["next_agent"]

# Build the graph
graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route)

# Workers always return to supervisor
for worker in ["researcher", "writer", "reviewer"]:
    graph.add_edge(worker, "supervisor")

app = graph.compile()

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Senior Researcher",
    goal="Find comprehensive data about the topic",
    backstory="Expert researcher with 10 years of experience",
    llm="claude-sonnet-4-20250514",
    allow_delegation=True
)

writer = Agent(
    role="Content Writer",
    goal="Write engaging, accurate articles",
    backstory="Award-winning journalist",
    llm="claude-sonnet-4-20250514"
)

editor = Agent(
    role="Editor",
    goal="Ensure quality, accuracy, and clarity",
    backstory="Veteran editor at a major publication",
    llm="claude-sonnet-4-20250514"
)

research_task = Task(
    description="Research AI agent adoption in Israeli tech companies",
    agent=researcher,
    expected_output="Structured research brief with facts and sources"
)

write_task = Task(
    description="Write a 1000-word article based on the research",
    agent=writer,
    expected_output="Polished article ready for review"
)

edit_task = Task(
    description="Review and polish the article",
    agent=editor,
    expected_output="Publication-ready article"
)

# Hierarchical process: a manager agent coordinates
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.hierarchical,
    manager_llm="claude-sonnet-4-20250514"
)

result = crew.kickoff()

import uuid
import time
import json
from dataclasses import dataclass, field
from typing import Any

@dataclass
class TraceEvent:
    correlation_id: str
    agent_name: str
    event_type: str  # "start", "llm_call", "tool_call", "handoff", "error", "end"
    timestamp: float
    data: dict = field(default_factory=dict)
    duration_ms: float = 0

class MultiAgentTracer:
    """Distributed tracing for multi-agent systems."""

    def __init__(self):
        self.events: list[TraceEvent] = []

    def start_run(self) -> str:
        correlation_id = str(uuid.uuid4())[:8]
        self.log(correlation_id, "system", "run_start", {})
        return correlation_id

    def log(self, correlation_id: str, agent: str, event_type: str, data: dict):
        event = TraceEvent(
            correlation_id=correlation_id,
            agent_name=agent,
            event_type=event_type,
            timestamp=time.time(),
            data=data
        )
        self.events.append(event)
        # Real-time logging
        print(f"[{correlation_id}][{agent}] {event_type}: "
              f"{json.dumps(data, ensure_ascii=False)[:100]}")

    def trace_agent(self, correlation_id: str, agent_name: str):
        """Context manager for tracing an agent's execution."""
        class AgentTrace:
            def __init__(self, tracer, cid, name):
                self.tracer = tracer
                self.cid = cid
                self.name = name
                self.start_time = None

            def __enter__(self):
                self.start_time = time.time()
                self.tracer.log(self.cid, self.name, "start", {})
                return self

            def __exit__(self, exc_type, exc_val, exc_tb):
                duration = (time.time() - self.start_time) * 1000
                if exc_type:
                    self.tracer.log(self.cid, self.name, "error",
                                   {"error": str(exc_val), "duration_ms": duration})
                else:
                    self.tracer.log(self.cid, self.name, "end",
                                   {"duration_ms": duration})

        return AgentTrace(self, correlation_id, agent_name)

    def get_timeline(self, correlation_id: str) -> list[dict]:
        """Get ordered timeline of events for a run."""
        events = [e for e in self.events if e.correlation_id == correlation_id]
        events.sort(key=lambda e: e.timestamp)
        return [
            {"agent": e.agent_name, "event": e.event_type,
             "data": e.data, "time": e.timestamp}
            for e in events
        ]

# Usage in a multi-agent pipeline
tracer = MultiAgentTracer()
cid = tracer.start_run()

with tracer.trace_agent(cid, "researcher"):
    tracer.log(cid, "researcher", "llm_call", {"model": "sonnet", "tokens": 1500})
    research = "..." # actual LLM call

with tracer.trace_agent(cid, "writer"):
    tracer.log(cid, "writer", "llm_call", {"model": "sonnet", "tokens": 2000})
    draft = "..."  # actual LLM call

# View the full timeline
for event in tracer.get_timeline(cid):
    print(f"  {event['agent']}: {event['event']}")

from dataclasses import dataclass

@dataclass
class AgentCostConfig:
    name: str
    model: str
    avg_input_tokens: int
    avg_output_tokens: int
    calls_per_run: float  # Average number of calls per run

# Pricing per 1M tokens (March 2026)
MODEL_PRICING = {
    "claude-opus-4": {"input": 15.0, "output": 75.0},
    "claude-sonnet-4": {"input": 3.0, "output": 15.0},
    "claude-haiku-4": {"input": 0.25, "output": 1.25},
    "gpt-5": {"input": 2.50, "output": 10.0},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gemini-2.5-pro": {"input": 1.25, "output": 5.0},
    "gemini-2.5-flash": {"input": 0.15, "output": 0.60},
}

def estimate_run_cost(agents: list[AgentCostConfig]) -> dict:
    """Estimate total cost for one multi-agent run."""
    total = 0
    breakdown = []

    for agent in agents:
        pricing = MODEL_PRICING[agent.model]
        input_cost = (agent.avg_input_tokens / 1_000_000) * pricing["input"]
        output_cost = (agent.avg_output_tokens / 1_000_000) * pricing["output"]
        agent_cost = (input_cost + output_cost) * agent.calls_per_run

        breakdown.append({
            "agent": agent.name,
            "model": agent.model,
            "cost_per_call": f"${input_cost + output_cost:.4f}",
            "calls": agent.calls_per_run,
            "total": f"${agent_cost:.4f}"
        })
        total += agent_cost

    return {
        "total_per_run": f"${total:.4f}",
        "total_per_1000_runs": f"${total * 1000:.2f}",
        "total_per_day_100_runs": f"${total * 100:.2f}",
        "breakdown": breakdown
    }

# Example: Content Pipeline with 4 agents
pipeline_agents = [
    AgentCostConfig("triage", "claude-haiku-4", 500, 100, 1),
    AgentCostConfig("researcher", "claude-sonnet-4", 2000, 1500, 1),
    AgentCostConfig("writer", "claude-sonnet-4", 3000, 2000, 1),
    AgentCostConfig("editor", "claude-sonnet-4", 4000, 2000, 1),
]

estimate = estimate_run_cost(pipeline_agents)
print(f"Cost per run: {estimate['total_per_run']}")
print(f"Cost per 1000 runs: {estimate['total_per_1000_runs']}")
for item in estimate["breakdown"]:
    print(f"  {item['agent']} ({item['model']}): {item['total']}")