LangChain 多智能体架构：深入解析 Supervisor 与 Handoffs 模式

uv add langgraph-supervisor langchain-openai

"""Run the langgraph supervisor sample with DeepSeek model."""
from langchain_openai import ChatOpenAI

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

model = ChatOpenAI(
    model="deepseek-chat",
    api_key="sk-...",
    base_url="https://api.deepseek.com"
)

# Create specialized agents

def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b

def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

def web_search(query: str) -> str:
    """Search the web for information."""
    return (
        "Here are the headcounts for each of the FAANG companies in 2024:\n"
        "1. **Facebook (Meta)**: 67,317 employees.\n"
        "2. **Apple**: 164,000 employees.\n"
        "3. **Amazon**: 1,551,000 employees.\n"
        "4. **Netflix**: 14,000 employees.\n"
        "5. **Google (Alphabet)**: 181,269 employees."
    )

math_agent = create_react_agent(
    model=model,
    tools=[add, multiply],
    name="math_expert",
    prompt="You are a math expert. Always use one tool at a time."
)

research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="You are a world class researcher with access to web search. Do not do any math."
)

# Create supervisor workflow
workflow = create_supervisor(
    [research_agent, math_agent],
    model=model,
    prompt=(
        "You are a team supervisor managing a research expert and a math expert. "
        "For current events, use research_agent. "
        "For math problems, use math_agent."
    )
)

# Compile and run
app = workflow.compile()
result = app.invoke({
    "messages": [
        {
            "role": "user",
            "content": "what's the combined headcount of the FAANG companies in 2024?"
        }
    ]
})

print("\n=== Final Answer ===")
messages = result.get("messages", []) if isinstance(result, dict) else []
if messages:
    last = messages[-1]
    output = getattr(last, "content", str(last))
    print(output)
    print("\n=== Trace ===")
    for i, m in enumerate(messages):
        name = getattr(m, "name", None) or "unknown"
        typ = getattr(m, "type", None) or type(m).__name__
        print(f"{i:02d}. {name} [{typ}]")
else:
    print(result)

=== Final Answer ===
Based on the research conducted, the combined headcount of FAANG companies in 2024 is **approximately 1,977,586 employees**.

Here's the breakdown:
- **Meta (Facebook)**: 67,317 employees
- **Apple**: 164,000 employees
- **Amazon**: 1,551,000 employees
- **Netflix**: 14,000 employees
- **Google (Alphabet)**: 181,269 employees

**Total: 1,977,586 employees**

Amazon accounts for the vast majority (over 78%) of FAANG's total workforce, largely due to its massive warehouse and logistics operations. The other four companies combined make up about 22% of the total headcount.

=== Trace ===
00. unknown [human]
01. supervisor [ai]
02. transfer_to_research_expert [tool]
03. research_expert [ai]
04. research_expert [ai]
05. transfer_back_to_supervisor [tool]
06. supervisor [ai]

workflow = create_supervisor(
    agents=[agent1, agent2],
    output_mode="full_history"
)

=== Trace ===
00. unknown [human]
01. supervisor [ai]
02. transfer_to_research_expert [tool]
03. research_expert [ai]
04. web_search [tool]
05. research_expert [ai]
06. web_search [tool]
07. research_expert [ai]
08. web_search [tool]
09. research_expert [ai]
10. web_search [tool]
11. research_expert [ai]
12. web_search [tool]
13. research_expert [ai]
14. web_search [tool]
15. research_expert [ai]
16. research_expert [ai]
17. transfer_back_to_supervisor [tool]
18. supervisor [ai]

research_team = create_supervisor(
    [research_agent, math_agent],
    model=model,
    supervisor_name="research_supervisor"
).compile(name="research_team")

writing_team = create_supervisor(
    [writing_agent, publishing_agent],
    model=model,
    supervisor_name="writing_supervisor"
).compile(name="writing_team")

top_level_supervisor = create_supervisor(
    [research_team, writing_team],
    model=model,
    supervisor_name="top_level_supervisor"
).compile(name="top_level_supervisor")

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()
store = InMemoryStore()

model = ...
research_agent = ...
math_agent = ...

workflow = create_supervisor(
    [research_agent, math_agent],
    model=model,
    prompt="You are a team supervisor managing a research expert and a math expert.",
)

# Compile with checkpointer/store
app = workflow.compile(
    checkpointer=checkpointer,
    store=store
)

from langgraph_supervisor import create_handoff_tool
workflow = create_supervisor(
    [research_agent, math_agent],
    tools=[
        create_handoff_tool(agent_name="math_expert", name="assign_to_math_expert", description="Assign task to math expert"),
        create_handoff_tool(agent_name="research_expert", name="assign_to_research_expert", description="Assign task to research expert")
    ],
    model=model,
)