04 Subagents 子智能体：主管派活，专家干活

1
sequenceDiagram
2
    participant U as 用户
3
    participant M as 主管 Agent
4
    participant R as 调研子智能体
5
    U->>M: "东京天气怎么样？"
6
    M->>R: research("东京天气")
7
    Note over M: 阻塞等待结果…
8
    R-->>M: "当前 72°F，晴"
9
    M-->>U: "东京现在 72°F，晴天"

1
sequenceDiagram
2
    participant U as 用户
3
    participant M as 主管 Agent
4
    participant J as 作业系统
5
    participant C as 合同审查子智能体
6
    U->>M: "帮我审一下这份并购合同"
7
    M->>J: run_agent("legal_reviewer", task)
8
    J->>C: 启动 Agent
9
    J-->>M: job_id: "job_123"
10
    M-->>U: "已开始审查（job_123）"
11
    Note over C: 正在审 150+ 页…
12
    U->>M: "进度怎么样？"
13
    M->>J: check_status(job_id)
14
    J-->>M: "running"
15
    M-->>U: "还在审…"
16
    U->>M: "好了吗？"
17
    M->>J: check_status(job_id)
18
    J-->>M: "completed"
19
    M->>J: get_result(job_id)
20
    J-->>M: 合同分析结果
21
    M-->>U: "审查完成：[结论]"

1
graph LR
2
    A[用户] --> B[主管 Agent]
3
    B --> C[子智能体 A]
4
    B --> D[子智能体 B]
5
    B --> E[子智能体 C]
6
    C --> B
7
    D --> B
8
    E --> B
9
    B --> F[回复用户]

1
from langchain.tools import tool
2
from langchain.agents import create_agent
3

4
# 每个子智能体 = 一个独立工具（入参/返回值都能单独设计）
5
subagent = create_agent(model="...", tools=[...])
6

7
@tool("subagent_name", description="subagent_description")
8
def call_subagent(query: str):
9
    result = subagent.invoke({"messages": [{"role": "user", "content": query}]})
10
    return result["messages"][-1].content
11

12
main_agent = create_agent(model="...", tools=[call_subagent])

1
graph LR
2
    A[用户] --> B[主管 Agent]
3
    B --> C{"task 工具<br/>agent_name, description"}
4
    C -->|research| D[调研 Agent]
5
    C -->|writer| E[写作 Agent]
6
    C -->|reviewer| F[审查 Agent]
7
    D --> C
8
    E --> C
9
    F --> C
10
    C --> B
11
    B --> G[回复用户]

1
from langchain.agents import AgentState
2
from langchain.tools import tool, ToolRuntime
3

4
class CustomState(AgentState):
5
    example_state_key: str
6

7
@tool("subagent1_name", description="subagent1_description")
8
def call_subagent1(query: str, runtime: ToolRuntime[None, CustomState]):
9
    # 按需把消息"加工"成适合子智能体的输入
10
    subagent_input = some_logic(query, runtime.state["messages"])
11
    result = subagent1.invoke({
12
        "messages": subagent_input,
13
        # 也可以顺便传其它 state 字段；
14
        # 但这些字段必须在"主 Agent 和子智能体"两边的 state schema 里都定义。
15
        "example_state_key": runtime.state["example_state_key"],
16
    })
17
    return result["messages"][-1].content

1
from typing import Annotated
2
from langchain.tools import InjectedToolCallId
3
from langgraph.types import Command
4

5
@tool("subagent1_name", description="subagent1_description")
6
def call_subagent1(
7
    query: str,
8
    tool_call_id: Annotated[str, InjectedToolCallId],
9
) -> Command:
10
    result = subagent1.invoke({"messages": [{"role": "user", "content": query}]})
11
    return Command(update={
12
        # 把子智能体的额外状态带回来
13
        "example_state_key": result["example_state_key"],
14
        "messages": [
15
            ToolMessage(
16
                content=result["messages"][-1].content,
17
                tool_call_id=tool_call_id,
18
            )
19
        ],
20
    })