Multi-Agent Orchestration

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

class OrchestratorState(TypedDict):
    messages: Annotated[list, operator.add]
    next_agent: str
    results: dict

def supervisor(state: OrchestratorState):
    """Central supervisor that routes tasks to agents."""
    llm = ChatOpenAI(model="gpt-4")
    response = llm.invoke(
        f"""You are a supervisor managing agents: researcher, writer, reviewer.
        Based on the current state, decide which agent should act next.
        Respond with just the agent name or 'FINISH' if done.
        Current results: {state['results']}"""
    )
    return {"next_agent": response.content.strip()}

def researcher_agent(state):
    # Research logic here
    return {"results": {"research": "findings..."}}

def writer_agent(state):
    # Writing logic here
    return {"results": {"draft": "content..."}}

# Build orchestration graph
graph = StateGraph(OrchestratorState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", lambda s: s["next_agent"], {
    "researcher": "researcher",
    "writer": "writer",
    "FINISH": END,
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")

# Shared state approach
class SharedState(TypedDict):
    task_queue: list[str]
    completed: list[dict]
    agent_status: dict[str, str]

# Message passing approach
class AgentMessage:
    sender: str
    receiver: str
    content: str
    message_type: str  # "request", "response", "handoff"

# Task routing based on capability
def route_task(task: str, agents: list) -> str:
    """Route task to the most capable agent."""
    capabilities = {
        "researcher": ["search", "analyze", "summarize"],
        "coder": ["code", "debug", "test"],
        "writer": ["write", "edit", "format"],
    }
    for agent, skills in capabilities.items():
        if any(s in task.lower() for s in skills):
            return agent
    return "researcher"  # default