高级多智能体

前置阅读：多智能体概览 · 多智能体模式

本章覆盖两种高级模式（Skills、Custom Workflow）以及多 Agent 系统的共性问题：内存共享、权限分级、错误处理、最佳实践。

必须深刻理解，不能跳过：worker 与编排的分离在高级模式中的体现。

• Skills 不是 LangChain create_agent 的内置能力，而是 Deep Agents 的约定式 Harness。LangChain 侧只做交叉引用。
• Custom Workflow 是 LangGraph StateGraph 的直接使用--它不是 create_agent 的扩展，而是绕过 create_agent 的封装，直接操作图拓扑。
• 权限分级 在 worker 层用 create_agent + middleware（如 HumanInTheLoopMiddleware）实现，在编排层用 LangGraph interrupt 实现。

Skills 技能模式

概念

Skills 模式将 Agent 的能力封装为可复用的技能单元。与 Subagents 不同，技能不是独立的 Agent，而是一组工具 + prompt 片段 + 使用指引的组合，可以"插入"到任意 Agent 中使用。

类比软件工程中的组合优于继承：与其创建一个"什么都会"的 Agent，不如让 Agent 按需装载技能模块。

落地方式：Deep Agents

Skills 是 Deep Agents 的约定式能力，不在 LangChain create_agent 中内置。Deep Agents 在 LangGraph/LangChain Agent 之上提供了 Skills 的标准封装、文件系统、上下文管理等能力。

不要在 LangChain 中手搓 Skills

旧教程可能用 dataclass + create_agent 手动拼接"技能"（工具列表 + prompt 片段）。这只是工具分组，不是真正的 Skills 模式--缺乏标准化的技能发现、加载、上下文隔离机制。如果你需要 Skills，应直接使用 Deep Agents Skills。

代码示例：Deep Agents Skills（交叉引用）

以下为 Deep Agents Skills 的使用概要，完整教程见 Deep Agents Skills：

# pip install deepagents
from deepagents import create_deep_agent

# Deep Agents 内置 Skills 机制，技能 = 工具 + 指引 + 上下文隔离
agent = create_deep_agent(
    model=model,
    skills=[sql_skill, search_skill, code_skill],  # 按 Deep Agents Skills 规范定义
    # Deep Agents 还提供：文件系统、Subagent、上下文管理、HITL 默认能力
)

如果你只需要简单的"工具分组"（不需要标准化的 Skills 机制），在 LangChain 侧可以用 create_agent + 分组工具列表近似：

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

model = init_chat_model(os.environ["LLM_MODEL"])

# ---- 简单的工具分组（不是真正的 Skills，仅做能力复用）----
@tool
def execute_sql(query: str) -> str:
    """执行 SQL 查询并返回结果"""
    return f"SQL 执行结果：查询 '{query}' 返回 15 行数据"

@tool
def web_search(query: str) -> str:
    """搜索网页"""
    return f"搜索结果：找到 5 条关于 '{query}' 的信息"

# 数据分析 Agent = 基础 + SQL + 搜索
data_analyst = create_agent(
    model,
    tools=[execute_sql, web_search],
    system_prompt=(
        "你是数据分析师，帮助用户从数据中获取洞察。"
        "你具备 SQL 查询和网络搜索能力。"
    ),
)

# 轻量搜索助手 = 基础 + 搜索
search_assistant = create_agent(
    model,
    tools=[web_search],
    system_prompt="你是搜索助手，帮助用户查找信息。",
)

何时升级到 Deep Agents Skills：

• 需要标准化的技能发现和加载机制
• 需要技能的上下文隔离（避免技能间互相干扰）
• 需要技能与文件系统、Subagent 协同
• 需要技能的可复用性跨项目共享

详见 Deep Agents Skills 和 Deep Agents 概览。

Custom Workflow 自定义工作流

概念

当 langgraph-supervisor / langgraph-swarm 的编排模式无法满足需求时，可以使用 LangGraph StateGraph 构建完全自定义的多 Agent 工作流。LangGraph 提供了图（Graph）抽象，支持条件分支、并行执行、循环和人工干预。

何时从编排包升级到 StateGraph

场景	langgraph-supervisor / swarm	LangGraph StateGraph
简单路由分发	足够	过度设计
固定流程的多步骤	足够	可选
条件分支 + 循环	勉强	推荐
并行执行多个 Agent	不支持	原生支持
人工审批中断	不直接支持	原生支持（interrupt）
持久化检查点	不直接支持	原生支持（checkpointer）
生产级错误恢复	有限	完善

代码示例：条件路由工作流

以下示例用 LangGraph StateGraph 构建条件分支工作流。worker 用 create_agent 构建，但编排逻辑由 StateGraph 的 add_node + add_conditional_edges 直接控制。

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain.messages import HumanMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Literal
import operator

model = init_chat_model(os.environ["LLM_MODEL"])

# ============ 状态定义 ============
class WorkflowState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    category: str   # 分类结果
    result: str     # 最终结果

# ============ 节点定义 ============
def classify_node(state: WorkflowState) -> dict:
    """分类节点：判断问题类型"""
    last_message = state["messages"][-1].content
    response = model.invoke([
        {"role": "system", "content": "将用户问题分类为 'technical' 或 'business' 或 'general'。只返回分类标签。"},
        {"role": "user", "content": last_message}
    ])
    return {"category": response.content.strip().lower()}

# 技术处理 worker
@tool
def debug_code(code: str) -> str:
    """调试代码问题"""
    return f"代码分析完成：发现 2 个潜在问题"

@tool
def suggest_fix(issue: str) -> str:
    """建议修复方案"""
    return f"修复建议：{issue} -> 使用缓存优化 + 异步处理"

tech_agent = create_agent(
    model,
    tools=[debug_code, suggest_fix],
    system_prompt="你是技术专家。",
    name="tech_agent",
)

def tech_node(state: WorkflowState) -> dict:
    """技术处理节点：调用 tech_agent worker"""
    result = tech_agent.invoke({"messages": state["messages"]})
    return {"result": result["messages"][-1].content}

# 业务处理 worker
@tool
def market_analysis(topic: str) -> str:
    """市场分析"""
    return f"{topic} 市场分析：增长率 15%，竞争格局稳定"

business_agent = create_agent(
    model,
    tools=[market_analysis],
    system_prompt="你是商业顾问。",
    name="business_agent",
)

def business_node(state: WorkflowState) -> dict:
    """业务处理节点：调用 business_agent worker"""
    result = business_agent.invoke({"messages": state["messages"]})
    return {"result": result["messages"][-1].content}

def general_node(state: WorkflowState) -> dict:
    """通用处理节点：直接用模型回复"""
    response = model.invoke(state["messages"])
    return {"result": response.content}

# 汇总节点
def summary_node(state: WorkflowState) -> dict:
    """汇总结果"""
    response = model.invoke([
        {"role": "system", "content": "将以下处理结果整理为用户友好的回复"},
        {"role": "user", "content": state["result"]}
    ])
    return {"messages": [response]}

# ============ 条件路由 ============
def route_by_category(state: WorkflowState) -> Literal["tech", "business", "general"]:
    """根据分类结果路由"""
    category = state.get("category", "general")
    if "technical" in category:
        return "tech"
    elif "business" in category:
        return "business"
    return "general"

# ============ 构建图 ============
workflow = StateGraph(WorkflowState)

# 添加节点
workflow.add_node("classify", classify_node)
workflow.add_node("tech", tech_node)
workflow.add_node("business", business_node)
workflow.add_node("general", general_node)
workflow.add_node("summary", summary_node)

# 添加边
workflow.add_edge(START, "classify")
workflow.add_conditional_edges("classify", route_by_category)
workflow.add_edge("tech", "summary")
workflow.add_edge("business", "summary")
workflow.add_edge("general", "summary")
workflow.add_edge("summary", END)

# 编译并执行
app = workflow.compile()
result = app.invoke({
    "messages": [HumanMessage(content="我们的 API 响应时间最近变慢了，该怎么排查？")]
})
print(result["result"])

人工审批中断（HITL）

LangGraph 的 interrupt + Command(resume=) 允许工作流在任意节点暂停，等待人工审批后继续：

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

def approval_node(state: WorkflowState) -> dict:
    """审批节点：暂停等待人工确认"""
    # interrupt 暂停图执行，payload 返回给调用方
    approval = interrupt({
        "question": "确认发布以下结果？",
        "content": state["result"],
    })
    # 人工恢复后，approval 是恢复时传入的值
    if approval.get("action") == "approve":
        return {"result": f"[已发布] {state['result']}"}
    else:
        return {"result": "[已驳回] 用户未批准"}

# 编译时需要 checkpointer 才能恢复中断
workflow = StateGraph(WorkflowState)
workflow.add_node("classify", classify_node)
workflow.add_node("approval", approval_node)
workflow.add_edge(START, "classify")
workflow.add_edge("classify", "approval")
workflow.add_edge("approval", END)

app = workflow.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "session-1"}}

# 第一次调用会中断
result = app.invoke(
    {"messages": [HumanMessage(content="...")]},
    config,
)
# 检查是否中断（result 中有 __interrupt__ 或 messages 显示中断）

# 人工审批后恢复
result = app.invoke(
    Command(resume={"action": "approve"}),
    config,
)
print(result["result"])

关于 LangGraph 的完整教程，请参阅 LangGraph 概览 和 子图编排。

多 Agent 内存共享

多个 Agent 协作时，如何共享上下文是核心挑战。以下是三种方案：

方案一：消息传递（最常用）

Agent 之间通过工具参数或 state 传递必要信息，每次交互显式传入上下文：

@tool
def delegate_to_analyst(task: str, context: str) -> str:
    """委派分析任务。
    Args:
        task: 具体的分析任务
        context: 之前步骤的结果摘要，作为分析依据
    """
    result = analyst_agent.invoke({
        "messages": [
            {"role": "system", "content": f"参考上下文：{context}"},
            {"role": "user", "content": task}
        ]
    })
    return result["messages"][-1].content

优点：简单直接，无隐式依赖。缺点：上下文可能在多次传递中丢失细节。

方案二：共享 State（LangGraph）

使用 LangGraph 的 StateGraph，所有节点共享同一个状态对象：

class SharedState(TypedDict):
    messages: Annotated[list, operator.add]
    research_data: str      # 搜索 Agent 写入
    analysis_result: str    # 分析 Agent 写入
    final_report: str       # 写作 Agent 写入

# 每个节点都能读写 SharedState
def research_node(state: SharedState) -> dict:
    # 读取 messages，写入 research_data
    return {"research_data": "搜索到的原始数据..."}

def analysis_node(state: SharedState) -> dict:
    # 读取 research_data，写入 analysis_result
    data = state["research_data"]
    return {"analysis_result": f"基于 {data} 的分析结果..."}

优点：所有 Agent 共享完整上下文，无信息丢失。缺点：需要 LangGraph，状态可能膨胀。

方案三：外部存储

对于长期记忆或跨会话共享，使用 LangGraph Store 或向量数据库：

from langgraph.store.memory import InMemoryStore
from langchain.embeddings import init_embeddings

# 共享的 Store（生产环境用 PostgresStore）
store = InMemoryStore()

@tool
def save_to_memory(namespace: str, key: str, content: str) -> str:
    """保存信息到共享记忆"""
    store.put(namespace=namespace, key=key, value={"content": content})
    return "已保存"

@tool
def recall_from_memory(namespace: str, query: str) -> str:
    """从共享记忆中检索相关信息"""
    items = store.search(namespace=namespace, query=query, limit=3)
    return "\n".join(item.value["content"] for item in items)

权限分级与人工干预

多 Agent 系统常需要按角色分级权限--只读 Agent 不能写数据，账单 Agent 不能发邮件。LangChain 提供两层权限控制：

Worker 层：create_agent + middleware

在 worker 层用 HumanInTheLoopMiddleware 对敏感工具做审批拦截：

import os
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.types import Command
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model(os.environ["LLM_MODEL"])

@tool
def query_data(sql: str) -> str:
    """只读查询"""
    return f"查询结果：{sql} 返回 5 行"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（敏感操作）"""
    return f"邮件已发送至 {to}"

@tool
def apply_refund(account_id: str, amount: float) -> str:
    """申请退款（敏感操作）"""
    return f"退款已申请：{account_id} ¥{amount}"

# 只读 Agent：无审批
readonly_agent = create_agent(
    model,
    tools=[query_data],
    system_prompt="你是只读分析助手，只能查询数据。",
    name="readonly_agent",
)

# 审批 Agent：敏感工具需要人工确认
approval_agent = create_agent(
    model,
    tools=[query_data, send_email, apply_refund],
    system_prompt="你是带审批的助手。发邮件和退款需要人工确认。",
    middleware=[
        HumanInTheLoopMiddleware(
            # interrupt_on 指定哪些工具需要人工审批
            interrupt_on={
                "send_email": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                "apply_refund": True,  # 简写：只允许 approve/reject
            },
        ),
    ],
    name="approval_agent",
)

# 执行：需要 checkpointer 才能恢复中断
app = approval_agent  # create_agent 返回已编译的图
config = {"configurable": {"thread_id": "session-1"}}

# 第一次调用：触发 send_email 时会中断
result = app.invoke(
    {"messages": [{"role": "user", "content": "给 boss@company.com 发邮件说退款已处理"}]},
    config,
)

# 人工审批后恢复
result = app.invoke(
    Command(resume={"action": "approve"}),
    config,
)
print(result["messages"][-1].content)

原生语义：HumanInTheLoopMiddleware(interrupt_on=...) 在 worker 的 Agent Loop 中插入中断点。当 LLM 决定调用 send_email 时，middleware 拦截工具调用并触发 interrupt，等待 Command(resume=) 恢复。这是 LangGraph 的 interrupt 机制在 create_agent middleware 层的封装。

编排层：LangGraph interrupt

在编排层（StateGraph）用 interrupt 对整个节点做审批：

from langgraph.types import interrupt, Command

def publish_node(state: SharedState) -> dict:
    """发布节点：需要人工确认才能发布"""
    approval = interrupt({
        "content": state["final_report"],
        "message": "确认发布？",
    })
    if approval.get("action") == "approve":
        return {"published": True}
    return {"published": False}

错误处理

多 Agent 系统的错误处理比单 Agent 更复杂--任何一个子 Agent 的失败都可能影响整体流程。

策略一：子 Agent 级别的重试

import time

def invoke_with_retry(agent, messages, max_retries=3):
    """带重试的 Agent 调用"""
    for attempt in range(max_retries):
        try:
            result = agent.invoke({"messages": messages})
            return result["messages"][-1].content
        except Exception as e:
            if attempt == max_retries - 1:
                return f"[错误] Agent 调用失败（{max_retries} 次重试后）：{str(e)}"
            wait = 2 ** attempt
            print(f"第 {attempt + 1} 次失败，{wait}s 后重试：{e}")
            time.sleep(wait)

策略二：降级回退

当专家 Agent 不可用时，回退到通用 Agent：

@tool
def route_to_expert(question: str) -> str:
    """路由到专家 Agent，不可用时降级到通用回答"""
    try:
        result = expert_agent.invoke(
            {"messages": [{"role": "user", "content": question}]}
        )
        return result["messages"][-1].content
    except Exception:
        # 降级：用通用模型直接回答
        response = model.invoke([{"role": "user", "content": question}])
        return f"[降级回复] {response.content}"

策略三：LangGraph 超时与错误处理

LangGraph 1.2+ 提供节点级超时和错误处理（Python 专属），比手动 signal.alarm 更可靠：

from langgraph.types import TimeoutPolicy
from langgraph.errors import NodeError

# 超时控制：add_node 的 timeout 参数
workflow.add_node(
    "tech",
    tech_node,
    timeout=30,  # 秒；也可传 TimeoutPolicy(run_timeout=30, idle_timeout=60)
)

# 错误处理：add_node 的 error_handler 参数
def tech_error_handler(error: NodeError) -> dict:
    """技术节点出错时的降级处理"""
    return {
        "result": f"[技术节点降级] {error.node} 执行失败：{error.error}",
    }

workflow.add_node("tech", tech_node, error_handler=tech_error_handler)

旧版（仅用于读旧项目）

旧教程可能用 signal.alarm 实现超时。该方案仅适用于 Unix 系统，且不支持 async。LangGraph 1.2+ 的 timeout= 参数是跨平台方案。

最佳实践

1. 最小权限原则

每个 Agent 只装载它需要的工具，避免"全能 Agent"：

# 好的做法：按职责分配工具
billing_agent = create_agent(
    model,
    tools=[check_balance, process_refund],
    system_prompt="你是账单专家。",
    name="billing_agent",
)
readonly_agent = create_agent(
    model,
    tools=[query_data],
    system_prompt="你是只读分析助手。",
    name="readonly_agent",
)

# 不好的做法：所有 Agent 都能做所有事
agent = create_agent(model, tools=[all_100_tools], system_prompt="你是万能助手")

2. 明确的角色边界

每个 Agent 的 system_prompt 应该清晰定义：能做什么、不能做什么、什么时候应该 handoff：

system_prompt = """你是账单处理专家。
能做：查询余额、处理退款、解释账单明细
不能做：技术问题排查、密码重置、账户注销
遇到不能做的事：使用 handoff 工具交接给对应专家"""

3. 可观测性

在生产环境中，记录每个 Agent 的调用链：

import logging

logger = logging.getLogger("multi_agent")

@tool
def delegate_with_logging(agent_name: str, task: str) -> str:
    """带日志的任务委派"""
    logger.info(f"[委派] {agent_name} <- {task[:50]}...")
    try:
        result = agents[agent_name].invoke(
            {"messages": [{"role": "user", "content": task}]}
        )
        response = result["messages"][-1].content
        logger.info(f"[完成] {agent_name} -> {response[:50]}...")
        return response
    except Exception as e:
        logger.error(f"[失败] {agent_name}: {e}")
        raise

4. 测试策略

多 Agent 系统的测试需要分层进行：

# 单元测试：单个 worker 的工具调用
def test_billing_agent_check_balance():
    result = billing_agent.invoke(
        {"messages": [{"role": "user", "content": "查询账户 A001 的余额"}]}
    )
    assert "余额" in result["messages"][-1].content

# 集成测试：编排层的交接
def test_handoff_triage_to_billing():
    result = app.invoke(
        {"messages": [{"role": "user", "content": "我上个月的账单有问题"}]},
        config={"configurable": {"thread_id": "test-1"}},
    )
    assert "账单" in result["messages"][-1].content

# 端到端测试：完整流程
def test_full_research_pipeline():
    result = supervisor_app.invoke(
        {"messages": [{"role": "user", "content": "研究 2024 年 AI 安全进展"}]},
        config={"configurable": {"thread_id": "test-2"}},
    )
    assert len(result["messages"][-1].content) > 100

5. 成本控制

多 Agent 系统的 LLM 调用次数成倍增长，需要关注成本：

• 使用分层模型：路由器用便宜快速的模型，专家用强模型（通过环境变量配置不同模型）
• 缓存频繁查询：对确定性工具的结果做缓存
• 限制迭代次数：在 invoke 时通过 config={"recursion_limit": N} 防止无限循环（不是 create_agent 的参数）
• 监控 Token 用量：追踪每个 Agent 的 Token 消耗

# 正确：recursion_limit 在 invoke 时通过 config 传入
result = app.invoke(
    {"messages": [...]},
    config={
        "configurable": {"thread_id": "session-1"},
        "recursion_limit": 25,
    },
)

旧版（仅用于读旧项目）

旧教程可能写 create_agent(model, tools, recursion_limit=25)。recursion_limit 不是 create_agent 的构造参数，这样写会抛 TypeError。正确做法是在 invoke / stream 时通过 config 传入。

前端类比

多 Agent 的成本控制类似前端的性能预算（Performance Budget）：

• 给每个 Agent 设定 Token 预算，就像给每个页面设定 JS bundle 大小上限
• 路由器用小模型 ≈ 关键路径用轻量代码，非关键路径懒加载
• 缓存工具结果 ≈ HTTP 缓存 / Service Worker 缓存

核心原则一样：在用户无感知的地方省成本，在用户体验关键点投入资源。

常见问题

Q：Skills 模式和给 Agent 加更多工具有什么区别？

A：Skills 模式（Deep Agents）的核心价值是标准化封装和复用。一个技能包含工具 + 指引 + 上下文隔离，是一个完整的能力单元。直接给 create_agent 加工具只是增加了函数列表，缺乏标准化的发现、加载和隔离机制。如果需要真正的 Skills，用 Deep Agents。

Q：Custom Workflow 什么时候才值得用？

A：满足以下任一条件时考虑 LangGraph StateGraph：

• 需要条件分支（if/else 路由）
• 需要并行执行多个 Agent
• 需要人工审批中断（interrupt）
• 需要持久化检查点和错误恢复
• 流程复杂度超过 langgraph-supervisor / langgraph-swarm 的表达能力

Q：多 Agent 系统的延迟怎么优化？

A：

1. 并行化：用 LangGraph 并行执行无依赖的子 Agent
2. 流式输出：使用 stream_mode="messages" 减少用户感知延迟
3. 模型选择：非关键路径用快速模型
4. 预热缓存：对常见查询预计算结果

Q：HumanInTheLoopMiddleware 和 LangGraph interrupt 有什么区别？

A：

• HumanInTheLoopMiddleware 在 worker 层（create_agent 内部）拦截特定工具调用，适合"某个工具需要审批"的场景
• LangGraph interrupt 在 编排层（StateGraph 节点）暂停整个节点，适合"整个流程步骤需要审批"的场景
• 两者底层都是 LangGraph 的 interrupt + Command(resume=) 机制

下一步

• 多智能体概览 - 回顾五种模式和选型决策
• 多智能体模式 - Subagents / Handoffs / Router 核心模式
• LangGraph 概览 - LangGraph 完整教程入口
• 子图编排 - LangGraph 中的多 Agent 子图模式
• Deep Agents Skills - Skills 技能模式的标准实现
• Deep Agents Subagents - 长周期研究、多文件多 Subagent 的高级 Harness

参考资源

• LangChain create_agent 文档
• LangGraph Multi-Agent 概念
• LangGraph StateGraph 文档
• HumanInTheLoopMiddleware API
• Deep Agents 文档