目录

1. 什么是多智能体系统

1.1 定义与核心思想

1.2 为什么需要多智能体

1.3 单Agent vs Multi-Agent 对比

1.4 典型结构示例：Supervisor + 子Agent

2. NotRequired vs Required vs Optional

3. 五大协作模式

3.1 模式概览

3.2 模式1：Subagents（子代理）

3.2.1 什么是 Subagents 架构

3.2.2 两种实现方式

3.2.3 上下文隔离机制

3.2.4 确保子代理被正确调用

3.2.5 按需向子代理注入上下文

3.2.6 控制子代理返回结果

3.2.7 同步 vs 异步执行

3.3 模式2：Handoffs（交接）

3.3.1 什么是Handoffs架构

3.3.2 两种实现方式

3.3.3 关键：为什么要包含ToolMessage

3.4 模式3：Skills（技能）

3.4.1 什么是Skills架构

3.4.2 skill VS MCP VS 提示词

3.4.3 显示清理 与 子图隔离

3.4.4 状态追踪 Skill 加载记录

3.4.5 按 Skill 大小选择加载策略

3.5 模式4：Router（路由器）

3.5.1 什么是Router架构

3.5.2 Router vs Subagents 

3.5.2 路由机制

3.5.3 有状态Router

3.6 模式5：Custom Workflow（自定义工作流）

3.6.1 什么是Custom Workflow架构

3.6.2 基础实现

1. 什么是多智能体系统

1.1 定义与核心思想

  • 多智能体系统是由 2 个及以上智能体构成的协作系统。在 LangGraph 框架下，它被具体实现为包含多个 Node 的 StateGraph，每个节点封装为一个独立的 Agent，共同协作完成宏观任务
  • 系统的质量取决于每个智能体能否访问到完成任务所需的正确数据。在 LangGraph 中，这体现为对全局 State 和 MessagesState 的精细设计，通过精准的 State 路由来实现信息的隔离与传递，避免 Token 爆炸和模型“走神”
  • 设计首要原则是“能用工作流，别用自主性”，通常是“宏观上的工作流+ 微观节点的 自主性”
  • Multi-agent（多智能体）系统通过协调多个专业化组件来处理复杂的工作流程。但并非所有复杂任务都需要这种方案——单个智能体配合合适的工具和提示词，往往也能达到类似效果

1.2 为什么需要多智能体

当开发者说“多智能体”时，通常需要以下一种或多种能力：

能力说明
上下文管理在模型上下文窗口有限的情况下，有选择地呈现相关专业知识
分布式开发不同团队可以独立开发和维护各自的能力，以清晰的边界组合成更大的系统
并行执行为了任务启动多个专业工作者，并发执行以加快结果产出

1.3 单Agent vs Multi-Agent 对比

维度单AgentMulti-Agent
决策方式所有事情由一个Agent决策拆分为“调度员+专家小组”
提示词复杂度逻辑复杂时，提示词容易膨胀每个Agent的Prompt简单清晰
调试难度难以定位问题架构清晰，便于理解和展示
上下文管理所有信息挤在一个窗口上下文隔离，各司其职

1.4 典型结构示例：Supervisor + 子Agent

用户提问：“明天北京天气怎么样？适合户外跑步吗？”
                    ↓
            Supervisor（调度员）
                    │
        ├─① 调用 ask_weather_expert("北京明天天气")
        │      └─→ 返回 “明天北京晴天，25°C，微风”
        │
        ├─② 调用 ask_date_expert("明天是星期几？")
        │      └─→ 返回 “明天是星期六”
        │
        └─③ 调用 ask_life_advisor("日期：星期六，天气：晴天25°C微风。请给跑步建议。")
                       └─→ 返回 “天气晴朗舒适，非常适合户外跑步！建议清晨或傍晚避免阳光直射。”
                    ↓
            Supervisor 整合并回复用户

关键说明：

  • 子代理无状态：每个子代理每次调用都是独立的，不记忆历史对话
  • 主管维护上下文：Supervisor通过工具调用串联信息
  • 并行执行：Supervisor可以同时调用多个子代理，然后等待结果

2. NotRequired vs Required vs Optional

class SupportState(AgentState):
    """客户支持工作流的状态。"""
    current_step: str
    warranty_status: Optional[Literal["in_warranty", "out_of_warranty"]]
    issue_type: NotRequired[Literal["hardware", "software"]]
类型标记对字典键的要求对键值的约束说明
不写任何标记（默认）必须提供值可以是任意类型（含 None）节点必须强制写入该字段
Optional[...]必须提供值允许为 None

键必须存在，只是值可以为空

容易导致状态字典无意义地塞满 None

NotRequired[...]可完全省略值若存在，则必须符合类型真正的“按需存在”。该键在图中可以“有”或“无”，完全取决于哪些节点需要它

3. 五大协作模式

3.1 模式概览

模式工作方式核心特点
Subagents主代理以工具形式协调子代理，所有路由经过主代理集中控制、上下文隔离
Handoffs行为根据状态动态变化，工具调用更新状态变量触发路由切换状态驱动、顺序约束
Skills按需加载专业提示词和知识，单个代理保持控制权渐进式披露、轻量级
Route路由步骤分类输入，分发给一个或多个专业代理，最后合成响应并行查询、结果合成
Custom Workflow使用 LangGraph 构建定制化执行流程完全控制、灵活组合

重要：这五种模式并不互斥，在实际系统里也经常组合使用

  • 模式选型指南
模式何时使用何时避免
Subagents多个不同领域、需要集中控制、子代理不需要直接与用户对话只需要几个工具的简单场景
Handoffs需要强制执行顺序约束、多阶段对话流程不需要状态驱动的简单场景
Skills单智能体多种能力、不同团队独立开发、需要按需加载知识需要在Skills间强制执行特定约束
Router泾渭分明的垂直领域、需要并行查询多个来源需要维护持续对话状态
Custom Workflow标准模式无法满足需求、需要结合确定性逻辑与智能体行为标准模式已足够
  • 实践建议
  1. 从简单开始：先用单Agent+工具，不够再拆分
  2. 上下文是第一优先级：每个智能体能否访问到正确的数据，决定了系统的质量
  3. 善用中间件：HumanInTheLoopMiddleware、SummarizationMiddleware 等让代码更简洁
  4. 检查点必须配置：跨轮次状态持久化、人机交互恢复都需要 checkpointer
  5. 五种模式可组合：如 Router 可作为 Custom Workflow 的一个节点

3.2 模式1：Subagents（子代理）

3.2.1 什么是 Subagents 架构

Subagents架构中，一个中央主代理（常称为监督者/Supervisor）通过将子代理包装为工具来协调它们。主代理决定调用哪个子代理、提供什么输入，以及如何组合结果

核心特性：

  • 集中控制：所有路由都通过主代理
  • 上下文隔离：主代理隔离子代理如何完成的全部过程噪声，子代理只获取输入并返回结果
  • 子代理无状态：每次被调用均启动全新上下文并在任务完成后丢弃
  • 并行执行：主代理可以在单次回合中调用多个子代理
  • 无直接用户交互：子代理将结果返回给主代理，而非用户。但可中断获取必要信息

    仅需在顶层代理添加检查点 checkpointer ，子代理自动继承

3.2.2 两种实现方式

维度Tool per agent（每个代理一个工具）Single dispatch tool（单一分派工具）
主代理视角看到多个独立工具仅看到一个统一工具
调用方式根据任务选择不同工具函数调用同一工具，通过参数指定具体代理
适用场景

子代理数量少且固定，

需对每个工具的输入/输出做精细定制

子代理数量多、由多团队独立开发、需频繁增删
  • 代码对比示例：
# ========== Tool per agent ==========
@tool
def call_research(query: str) -> str:
    return research_agent.invoke({"messages": [{"role": "user", "content": query}]})["messages"][-1].text

@tool
def call_writer(query: str) -> str:
    return writer_agent.invoke({"messages": [{"role": "user", "content": query}]})["messages"][-1].text

main_agent = create_agent(model="...", tools=[call_research, call_writer])

# ========== Single dispatch tool ==========
SUBAGENTS = {"research": research_agent, "writer": writer_agent}

@tool
def task(agent_name: str, description: str) -> str:
    agent = SUBAGENTS[agent_name]
    return agent.invoke({"messages": [{"role": "user", "content": description}]})["messages"][-1].text

main_agent = create_agent(model="...", tools=[task])  # 仅需一个工具！

3.2.3 上下文隔离机制

对比维度Tool-per-AgentSingle Dispatch Tool核心差异
返回结果父上下文中会追加多条名称各异的ToolMessage。不同工具返回的文本结构五花八门，杂乱堆积在主窗口父上下文仅追加唯一名称的Task消息。所有子代理的返回结果全部塞进同一个ToolMessage里，仅description参数内容不同Single Dispatch对父上下文的污染更“规整”，极大降低主LLM解析历史时“迷失”的概率
路由决策决策重担压在LLM身上：主代理需从N个工具签名（名称+描述）中做语义匹配，若两个工具描述接近，LLM易混淆，且工具描述本身占用了主代理的上下文窗口空间路由抽象为参数填充：主代理只需在task工具中填agent_name参数。路由匹配逻辑从“黑盒语义推理”降级为“白盒查字典”Single Dispatch在主代理的上下文窗口中，将“路由决策信息”压缩到了极致，释放了主窗口的认知Token

3.2.4 确保子代理被正确调用

主代理依靠工具的名称和描述决策。根据系统复杂度，有三种声明策略：

  • 策略A：系统提示枚举（适合 < 10 个固定代理）
    直接在主代理 system_prompt 中罗列可用代理及其职责
  • 策略B：枚举约束（适合需类型安全的中型系统）
    使用 Enum 限制 agent_name 参数，提供 IDE 自动补全与校验
  • 策略C：基于工具发现（适合 > 10 个动态代理）
    提供 list_agents 工具让主代理按需查询，实现渐进式披露，降低 Prompt Token 消耗

策略A：

main_agent = create_agent(
    model="...",
    tools=[task],
    system_prompt=(
        "你负责协调专业子代理。"
        "可用代理如下：\n"
        "- research：研究与事实核查\n"
        "- writer：内容创作与编辑\n"
        "- reviewer：代码与文档审查\n"
        "请使用 task 工具来分配工作。"
    ),
)

策略B：

from enum import Enum

class AgentName(str, Enum):
    RESEARCH = "research"
    WRITER = "writer"
    REVIEWER = "reviewer"

@tool
def task(
    agent_name: AgentName,  # 枚举约束
    description: str
) -> str:
    """为一项任务启动一个临时的子代理。"""
    # 实现逻辑（此处省略）
    pass
​

策略C示例：

@tool
def list_agents(query: str = "") -> str:
    """按关键词筛选并返回可用子代理列表及描述。"""
    return format_agent_list(search_agent_registry(query))

main_agent = create_agent(
    model="...",
    tools=[task, list_agents],  # 主代理先查再用
    system_prompt="先用list_agents查找可用代理，再用task分配任务。"
)

3.2.5 按需向子代理注入上下文

  • 目标：让子智能体能够访问完整的对话历史，解决指代不明或依赖历史信息的问题
  • 实现方法：在定义子智能体的包装工具时，通过 ToolRuntime 参数访问当前运行状态，从中提取所需的历史消息，并手动构造一个包含完整上下文的提示词发送给子智能体
from langchain.tools import tool, ToolRuntime

@tool
def schedule_event(request: str, runtime: ToolRuntime) -> str:
    """使用自然语言安排日历事件。"""
    # 从运行状态中获取完整历史，并提取初始用户原始问题
    original_user_msg = next(
        msg for msg in runtime.state["messages"] if msg.type == "human"
    )
    # 构造融合“全局背景”与“当前局部任务”的提示词
    enhanced_prompt = (
        f"用户原始询问：\n{original_user_msg.text}\n\n"
        f"当前需完成的具体子任务：\n{request}"
    )
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": enhanced_prompt}]
    })
    return result["messages"][-1].text

3.2.6 控制子代理返回结果

  • 原因：默认返回子代理的最终文本，监督者可能还需依赖 LLM 二次解析才能提取关键字段
  • 实现：将返回值改为 JSON 结构化数据，降低父代理推理开销与幻觉风险
import json

@tool
def schedule_event(request: str) -> str:
    result = calendar_agent.invoke({"messages": [{"role": "user", "content": request}]})
    # 返回结构化数据，而不只是纯文本
    return json.dumps({
        "status": "success",
        "event_id": "evt_123",
        "summary": result["messages"][-1].text  # 保留人类可读摘要
    })
  • 子代理在执行工具后，可能只返回工具调用成功的简单确认，导致最终返回信息缺少关键细节
  • 必须在子代理的系统提示词中显式声明输出规范，强制要求其将工具执行结果整合进最终回复
# 必须在日历子代理的 System Prompt 中追加明确约束
CALENDAR_AGENT_PROMPT = (
    "你是日历日程安排助手。"
    # ... 原有指令 ...
    "【关键输出要求】在最终回复中，你必须明确包含：事件标题、具体起止时间、参与人列表。"
    "绝对禁止只回复‘已创建’而不提供上述详情。"
)

3.2.7 同步 vs 异步执行

子代理执行可以是同步（阻塞）或异步（后台）。

模式主代理行为最适合场景权衡
同步等待子代理完成主代理需要结果才能继续简单，但会阻塞对话
异步子代理后台运行时主代理继续执行独立任务，用户无需等待响应式，但更复杂
  • 同步（默认）
# 默认同步调用：主代理等待子代理完成
result = calendar_agent.invoke({
    "messages": [{"role": "user", "content": request}]
})

适用场景：

  • 主代理需要子代理结果来形成响应
  • 任务有顺序依赖（如：获取数据→分析→响应）
  • 子代理失败应阻塞主代理响应

异步

异步指主代理启动后台作业并不阻塞地继续执行，不同于Python的 async/await

核心架构流程

  1. 主代理：接收用户请求 → 调用 submit_task 工具 → 将任务数据推入消息队列（Redis/DB）→ 立即返回 Task ID 给用户（不等待子代理）
  2. 后台 Worker（独立进程/服务）：持续监听队列 → 拉取任务 → 在独立的内存空间中启动子代理（全新上下文）→ 执行多步推理 → 将最终结果写回共享存储（DB）
  3. 主代理：用户稍后询问 → 主代理调用 check_task_status 工具 → 从 DB 读取结果返回

适用场景：

  • 子代理工作独立于主对话流程
  • 用户应能在工作进行时继续聊天
  • 希望并行运行多个独立任务

3.3 模式2：Handoffs（交接）

3.3.1 什么是Handoffs架构

Handoffs 是一种让 Agen t根据当前所处阶段动态切换行为的架构模式

核心机制：

  • 状态驱动行为：基于状态变量改变行为
  • 工具触发转换：工具通过更新状态变量来实现在不同状态间移动
  • 状态持久化：状态在对话的多个轮次间持续存在
  • 直接用户交互：每个状态的配置都直接处理用户消息

适用场景：

  • 需要强制执行顺序约束（如：仅在满足前置条件后才解锁特定功能）
  • 代理需要在不同状态下与用户直接对话
  • 构建多阶段的对话流程（如客户支持场景）

核心机制：

  • 创建⼀个返回 Command 对象的工具，用于更新状态，从而触发向新步骤或代理的转换
@tool
def transfer_to_specialist(runtime) -> Command:
    """转至专业代理机构。"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content="转至专业人员处理",
                    tool_call_id=runtime.tool_call_id
                )
            ],
            "current_step": "specialist"  # 触发行为改变
        }
    )

3.3.2 两种实现方式

在大多数交接场景中，使用单代理 + 中间件方法更简单。只有在需要定制化代理实现时（例如，节点本身是一个包含反思或检索步骤的复杂图），才使用多代理子图方法

方式一：单Agent + 中间件（推荐）

单个代理基于状态改变其行为，中间件拦截每次模型调用并动态调整系统提示词和可用工具

from typing import Callable, Literal
from typing_extensions import NotRequired

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import (
    ModelRequest,
    ModelResponse,
    SummarizationMiddleware,
    wrap_model_call,
)
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

# =============================================
# 1. 定义自定义状态 (State)
# =============================================

# 定义可能的工作流步骤：保修收集员、问题分类器、解决方案专家
SupportStep = Literal["warranty_collector", "issue_classifier", "resolution_specialist"]


class SupportState(AgentState):
    """客户支持工作流的状态。"""
    current_step: NotRequired[SupportStep]
    warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]]
    issue_type: NotRequired[Literal["hardware", "software"]]


# =============================================
# 2. 创建管理状态流转的工具 (Tools)
# =============================================

@tool
def record_warranty_status(
    status: Literal["in_warranty", "out_of_warranty"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """记录客户的保修状态并转换到问题分类步骤。"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"保修状态已记录为：{status}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "warranty_status": status,
            "current_step": "issue_classifier",  # 进入问题分类器
        }
    )


@tool
def record_issue_type(
    issue_type: Literal["hardware", "software"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """记录问题类型并转换到解决方案专家步骤。"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"问题类型已记录为：{issue_type}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "issue_type": issue_type,
            "current_step": "resolution_specialist",  # 进入解决方案专家
        }
    )


@tool
def escalate_to_human(reason: str) -> str:
    """将案例升级给人工支持专员。"""
    # 在实际系统中，这里会创建工单、通知员工等。
    return f"正在升级给人工支持。原因：{reason}"


@tool
def provide_solution(solution: str) -> str:
    """向客户提供解决方案。"""
    return f"已提供解决方案：{solution}"


# =============================================
# 3. 定义步骤配置映射 (Configuration)
# =============================================

# 保修收集员提示词
WARRANTY_COLLECTOR_PROMPT = """你是一名客户支持专员，负责帮助解决设备问题。
当前步骤：保修验证

在此步骤中，你需要：
1. 询问他们的设备是否在保修期内
2. 根据用户的回答，使用 record_warranty_status 记录他们的回答并进入下一步

对话要自然友好，不要一次问多个问题。"""

# 问题分类器提示词
ISSUE_CLASSIFIER_PROMPT = """你是一名客户支持专员，负责帮助解决设备问题。
当前步骤：问题分类

客户信息：保修状态为 {warranty_status}

在此步骤中，你需要：
1. 请客户描述他们遇到的问题
2. 判断问题是硬件问题（物理损坏、部件破损）还是软件问题（应用崩溃、性能问题）
3. 使用 record_issue_type 记录分类结果并进入下一步

如果无法确定，请在分类前先提出澄清性问题。"""

# 解决方案专家提示词
RESOLUTION_SPECIALIST_PROMPT = """你是一名客户支持专员，负责帮助解决设备问题。
当前步骤：解决方案

客户信息：保修状态为 {warranty_status}，问题类型为 {issue_type}

在此步骤中，你需要：
1. 对于软件问题：使用 provide_solution 提供故障排除步骤
2. 对于硬件问题：
   - 如果在保修期内：使用 provide_solution 说明保修维修流程
   - 如果已过保修期：使用 escalate_to_human 升级以便提供付费维修选项

提供的解决方案要具体且有帮助。"""

# 步骤配置：将步骤名称映射到 (提示词, 工具, 所需状态)
STEP_CONFIG = {
    "warranty_collector": {
        "prompt": WARRANTY_COLLECTOR_PROMPT,
        "tools": [record_warranty_status],
        "requires": [],
    },
    "issue_classifier": {
        "prompt": ISSUE_CLASSIFIER_PROMPT,
        "tools": [record_issue_type],
        "requires": ["warranty_status"],
    },
    "resolution_specialist": {
        "prompt": RESOLUTION_SPECIALIST_PROMPT,
        "tools": [provide_solution, escalate_to_human],
        "requires": ["warranty_status", "issue_type"],
    },
}


# =============================================
# 4. 创建动态配置中间件 (Middleware)
# =============================================

@wrap_model_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """根据当前步骤配置智能体的行为。"""
    # 获取当前步骤（首次交互默认为 warranty_collector）
    current_step = request.state.get("current_step", "warranty_collector")

    # 查找步骤配置
    step_config = STEP_CONFIG[current_step]

    # 验证所需的状态是否存在
    for key in step_config["requires"]:
        if request.state.get(key) is None:
            raise ValueError(f"在进入 {current_step} 之前必须设置 {key}")

    # 使用状态值格式化提示词（字典解包按键匹配替换占位符）
    system_prompt = step_config["prompt"].format(**request.state)

    # 注入系统提示词和步骤专用工具
    request = request.override(
        system_prompt=system_prompt,
        tools=step_config["tools"],
    )
    return handler(request)


# =============================================
# 5. 组装智能体 (Agent)
# =============================================

# 创建带有步骤配置和摘要中间件的智能体
agent = create_agent(
    "gpt-5.6-luna",
    tools=[
        record_warranty_status,
        record_issue_type,
        provide_solution,
        escalate_to_human,
    ],
    state_schema=SupportState,
    middleware=[
        apply_step_config,
        # 随着对话进行，上下文可能过长。
        # # 引入 SummarizationMiddleware 对早期消息进行摘要压缩。
        # SummarizationMiddleware(
        #     model="gpt-5.6-luna",
        #     trigger=("tokens", 4000),
        #     keep=("messages", 10)
        # )
    ],
    checkpointer=InMemorySaver(),  # 必须：用于跨对话轮次保留状态
)


# =============================================
# 6. 测试示例
# =============================================

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "thread_1"}}

    # 第一轮：用户报修
    result = agent.invoke(
        {"messages": [HumanMessage("你好，我的手机屏幕碎了")]},
        config
    )

    # 第二轮：回答保修状态
    result = agent.invoke(
        {"messages": [HumanMessage("是的，还在保修期内")]},
        config
    )

    # 第三轮：描述具体问题
    result = agent.invoke(
        {"messages": [HumanMessage("屏幕是摔坏的，物理上有裂痕")]},
        config
    )

    # 第四轮：寻求解决方案
    result = agent.invoke(
        {"messages": [HumanMessage("我该怎么办？")]},
        config
    )

    # 打印完整对话历史
    for msg in result['messages']:
        msg.pretty_print()

方式二：多Agent子图

独立的代理作为图中的独立节点存在，交接工具使用 Command.PARENT 在代理节点之间导航

from typing import Literal

from langchain.agents import AgentState, create_agent
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing_extensions import NotRequired


# =============================================
# 1. 定义包含 active_agent 追踪器的状态
# =============================================

class MultiAgentState(AgentState):
    active_agent: NotRequired[str]


# =============================================
# 2. 创建交接工具
# =============================================

@tool
def transfer_to_sales(
    runtime: ToolRuntime,
) -> Command:
    """转接到销售代理。"""
    last_ai_message = next(
        msg for msg in reversed(runtime.state["messages"])
        if isinstance(msg, AIMessage)
    )
    transfer_message = ToolMessage(
        content="已从支持代理转接到销售代理",
        tool_call_id=runtime.tool_call_id,
    )
    return Command(
        goto="sales_agent",
        update={
            "active_agent": "sales_agent",
            "messages": [last_ai_message, transfer_message],
        },
        graph=Command.PARENT,
    )


@tool
def transfer_to_support(
    runtime: ToolRuntime,
) -> Command:
    """转接到支持代理。"""
    last_ai_message = next(
        msg for msg in reversed(runtime.state["messages"])
        if isinstance(msg, AIMessage)
    )
    transfer_message = ToolMessage(
        content="已从销售代理转接到支持代理",
        tool_call_id=runtime.tool_call_id,
    )
    return Command(
        goto="support_agent",
        update={
            "active_agent": "support_agent",
            "messages": [last_ai_message, transfer_message],
        },
        graph=Command.PARENT,
    )


# =============================================
# 3. 使用交接工具创建代理
# =============================================

sales_agent = create_agent(
    model="gpt-5-mini",
    tools=[transfer_to_support],
    system_prompt=(
        "你是一个销售代理，负责处理销售咨询。"
        "如果用户询问技术问题或寻求支持，请转接到支持代理。"
    ),
)

support_agent = create_agent(
    model="gpt-5-mini",
    tools=[transfer_to_sales],
    system_prompt=(
        "你是一个支持代理，负责处理技术问题。"
        "如果用户询问价格或购买事宜，请转接到销售代理。"
    ),
)


# =============================================
# 4. 创建调用代理的节点函数
# =============================================

def call_sales_agent(state: MultiAgentState) -> Command:
    """调用销售代理的节点。"""
    response = sales_agent.invoke(state)
    return response


def call_support_agent(state: MultiAgentState) -> Command:
    """调用支持代理的节点。"""
    response = support_agent.invoke(state)
    return response


# =============================================
# 5. 创建路由函数，判断是结束还是继续
# =============================================

def route_after_agent(
    state: MultiAgentState,
) -> Literal["sales_agent", "support_agent", "__end__"]:
    """基于 active_agent 进行路由，如果代理在没有交接的情况下结束，则返回 END。"""
    messages = state.get("messages", [])

    # 检查最后一条消息 - 如果是没有工具调用的 AIMessage，我们已完成
    if messages:
        last_msg = messages[-1]
        if isinstance(last_msg, AIMessage) and not last_msg.tool_calls:
            return "__end__"

    # 否则，路由到激活的代理
    active = state.get("active_agent", "sales_agent")
    return active if active else "sales_agent"


def route_initial(
    state: MultiAgentState,
) -> Literal["sales_agent", "support_agent"]:
    """根据状态路由到激活的代理，默认为销售代理。"""
    return state.get("active_agent") or "sales_agent"


# =============================================
# 6. 构建图
# =============================================

builder = StateGraph(MultiAgentState)
builder.add_node("sales_agent", call_sales_agent)
builder.add_node("support_agent", call_support_agent)

# 基于初始 active_agent 进行条件路由启动
builder.add_conditional_edges(START, route_initial, ["sales_agent", "support_agent"])

# 在每个代理之后，检查是否结束或路由到另一个代理
builder.add_conditional_edges(
    "sales_agent",
    route_after_agent,
    ["sales_agent", "support_agent", END]
)
builder.add_conditional_edges(
    "support_agent",
    route_after_agent,
    ["sales_agent", "support_agent", END]
)

graph = builder.compile()


# =============================================
# 7. 测试执行
# =============================================

if __name__ == "__main__":
    result = graph.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "你好，我在登录账户时遇到问题，能帮帮我吗？",
                }
            ]
        }
    )

    for msg in result["messages"]:
        msg.pretty_print()

3.3.3 关键：为什么要包含ToolMessage

当LLM调用工具时，它期望得到一个响应。具有匹配 tool_call_id 的 ToolMessage 完成了这个请求-响应循环——没有它，对话历史将变得格式错误。交接工具更新 messages 时，这是必需的

3.4 模式3：Skills（技能）

3.4.1 什么是Skills架构

Skill 是以提示词为核心驱动、绑定专属工具集和上下文的原子性能力单元
在 Skills 架构中，智能体根据任务需求动态加载对应的 Skill 获取专项能力。该能力的激活范围仅限于当前任务步骤或子图（Sub-graph）内，执行完毕后通过子图隔离或显式状态清理机制，将中间过程从上下文中剥离，仅保留最终结果，以此保证上下文的纯净与系统的可扩展性

  • 全量加载：启动就将所有技能的内容塞进系统提示词→上下文爆炸、注意力分散、成本剧增
  • 渐进式披露 ：系统提示词里只放技能名称和简单描述（轻量级索引），哪个技能用到了再通过工具调用去加载完整内容→上下文清爽、按需加载、无限扩展

核心特征：

  • 提示词驱动的专项化：Skills主要由专门的提示词定义
  • 渐进式披露：Skills根据上下文或用户需求变得可用
  • 轻量级组合：Skills比完整的子智能体更简单
  • 团队分布式开发：不同团队可以独立开发和维护各自的Skills

3.4.2 skill VS MCP VS 提示词

对比维度SkillMCP原生提示词
核心本质有状态的能力容器（提示词模板 + 专属工具集 + 上下文约束 + 生命周期管理）标准化的数据/工具接入总线（类比 USB-C 接口，统一外部资源接入的 RPC 协议）无状态的文本输入（仅有自然语言指令，无绑定逻辑）
驱动与执行机制图状态（State）驱动。由 LangGraph 节点调度，运行时动态编译为 System Prompt 注入，并绑定额外的 Tool 权限协议驱动。通过 Client-Server 模式暴露 Resources（数据）和 Tools（函数），需显式调用 mcp_call_tool静态语义驱动。完全依赖 LLM 的上下文理解，本身不触发任何代码执行或外部调用
上下文管理策略渐进式披露 + 瞬时隔离。索引常驻，完整长文本在执行瞬间注入，执行完毕即通过子图屏障或清理节点剥离，State 仅保留最终结果持久化句柄引用。仅将 MCP 返回的极简结果写入上下文，原始大数据（如文件内容）保持在 MCP Server 内存中，按指针索引全量硬塞。无论多长（如万字设定），都塞入当前轮次的上下文窗口，无自动卸载机制，全靠人工截断
工具绑定方式专属紧耦合。绑定的工具集是硬编码进 Skill 能力定义中的（例如“财报 Skill”只绑 query_finance_db）动态松耦合。通过统一接口 tools/list 动态发现，Agent 需先感知可用工具列表，再跨进程调用纯文本映射。仅用文字“描述”工具功能，实际工具的路由逻辑需要外部代码额外挂载
多团队/版本治理逻辑隔离强。不同团队维护各自 Skill 的 Prompt 和 Tool 白名单，通过 Reducer 命名空间防止 State 键值冲突物理隔离强。独立的 MCP Server 进程，有自己的生命周期和权限控制，Skill 可跨多个 MCP 调取数据全局混战。所有提示词拼接在一个字符串中，不同团队撰写的指令极易相互覆盖或注意力稀释
在 LangGraph 中的典型落位Subgraph（子图）节点 或 动态 System Prompt 构造器（Pre-model 钩子）Node 内部的 ToolNode 调用。即 agent_node 生成 tool_calls 后，路由至 tools_node 执行 MCP 请求ChatPromptTemplate 的基础占位符，是最底层的原材料

3.4.3 显示清理 与 子图隔离

from typing import Optional, Literal
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import os

# ==================== 1. 定义 Skill 数据结构及模拟数据 ====================
class SkillState(MessagesState):
    skill_context: Optional[str]  # 仅存当前激活的Skill完整内容（长文本）

SKILLS = [
    {
        "name": "sales_analytics",
        "description": "用于销售数据分析的数据库模式与业务逻辑",
        "content": """
        【模拟长文本】销售分析数据库模式（约5000字）：
        - customers 表：customer_id, name, email, signup_date, region
        - orders 表：order_id, customer_id, order_date, total_amount, status
        - order_items 表：item_id, order_id, product_id, quantity, price
        关键业务逻辑：
        1. 计算月度销售额需 JOIN orders 和 order_items，过滤 status='completed'
        2. 区域分析需结合 customers.region 进行 GROUP BY
        """
    },
    {
        "name": "hr_policy",
        "description": "公司内部人力资源政策与休假制度",
        "content": """
        【模拟长文本】2025年员工手册摘要（约3000字）：
        - 年假：入职满1年享5天，满3年享10天，满5年享15天
        - 病假：需提供医院证明，每年最多20天带薪病假
        - 调休：周末加班可申请调休，需在当季度内使用完毕
        """
    }
]

def get_skill_index_prompt() -> str:
    """生成轻量级索引（只含名字和描述）"""
    return "\n".join([f"- **{s['name']}**: {s['description']}" for s in SKILLS])

def load_skill_content(name: str) -> str:
    """根据名称加载Skill完整长文本"""
    for s in SKILLS:
        if s["name"] == name:
            return s["content"]
    return ""

# ==================== 2. 定义工具（轻量确认，不含长文本） ====================
@tool
def load_skill(skill_name: str) -> str:
    """加载指定Skill的完整内容到当前工作上下文。"""
    # 只返回确认句，大文本由后续逻辑从 SKILLS 常量中读取
    return f"已成功加载技能: {skill_name}"

tools = [load_skill]

# ==================== 3. 核心：Agent节点（注入与清理） ====================
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 基础系统提示词：只有索引，没有长文本
BASE_SYSTEM_PROMPT = f"""你是一个智能助手。
可用Skills索引（加载后可使用详细能力）：
{get_skill_index_prompt()}
当需要专项能力时，请调用 load_skill 工具。
"""

def agent_node(state: SkillState):
    # 3.1 构建动态系统提示词（渐进式披露的核心）
    sys_content = BASE_SYSTEM_PROMPT
    skill_context = state.get("skill_context")
    
    if skill_context:
        # 若State中存在Skill长文本，则临时拼入System Prompt
        sys_content += f"\n\n【当前激活的专项能力详情】\n{skill_context}"

    system_msg = SystemMessage(content=sys_content)
    
    # 3.2 调用大模型
    response = model.invoke([system_msg] + state["messages"])
    
    # 3.3 【显式清理逻辑】执行完毕后立即剥离上下文
    updates = {"messages": [response]}
    if not response.tool_calls:
        # 如果模型本次回复没有要求调用工具，说明当前任务已结束，清空Skill内容
        updates["skill_context"] = None  # 显式清理！
    
    return updates

# ==================== 4. 路由逻辑 ====================
def should_continue(state: SkillState) -> Literal["tools", END]:
    """判断是否需要执行工具"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# ==================== 5. 【合并节点】工具执行 + 技能加载一体化 ====================
# 预置一个标准工具执行器
base_tool_executor = ToolNode(tools)

def tools_with_loading_node(state: SkillState):
    """
    合并节点：
    1. 执行所有的 tool_calls（包括 load_skill 和其他业务工具）
    2. 精准回溯解析参数，若调用了 load_skill，立即将长文本写入 skill_context
    """
    # 5.1 执行所有工具，获取 ToolMessage 列表
    result = base_tool_executor.invoke(state)  
    tool_messages = result["messages"]  # 这是返回给 LLM 的消息列表
    
    # 5.2 从历史 State 中回溯查找原本的 AIMessage，提取 skill_name
    skill_content_to_load = None
    for msg in reversed(state["messages"]):
        if isinstance(msg, AIMessage) and msg.tool_calls:
            for tc in msg.tool_calls:
                # 精准匹配工具名（防止误判其他工具）
                if tc.get("name") == "load_skill":
                    skill_name = tc["args"].get("skill_name")
                    if skill_name:
                        skill_content_to_load = load_skill_content(skill_name)
                    break
            if skill_content_to_load:
                break

    # 5.3 构造返回更新（同时携带工具执行结果和 Skill 长文本）
    updates = {"messages": tool_messages}
    if skill_content_to_load:
        updates["skill_context"] = skill_content_to_load
        print(f"[合并节点] 已加载技能长文本（{len(skill_content_to_load)} 字符）")
    else:
        # 如果这次调用没有涉及 load_skill，但之前遗留了 skill_context，保持原状（由 Agent 负责清理）
        pass
    
    return updates

# ==================== 6. 构建并编译图（极简版） ====================
builder = StateGraph(SkillState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tools_with_loading_node)  # 合并后的节点

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")  # 工具执行+加载后，直接回到 Agent

graph = builder.compile()

# ==================== 7. 测试与验证 ====================
if __name__ == "__main__":
    print("=" * 60)
    print(" 开始测试 Skill 渐进式披露 + 显式清理架构（合并节点版）")
    print("=" * 60)

    # 测试用例1：需要销售分析能力
    test_input = {"messages": [HumanMessage(content="帮我查一下上个月的销售总额是多少？")]}
    
    print(f"\n>>> 用户提问: {test_input['messages'][0].content}")
    print("-" * 60)
    
    # 逐轮流式打印状态变化（便于观察加载与清理）
    for idx, event in enumerate(graph.stream(test_input, stream_mode="updates")):
        print(f"\n--- 步骤 {idx+1} 更新 ---")
        for node_name, update in event.items():
            print(f"节点: {node_name}")
            
            if node_name == "agent":
                if "messages" in update:
                    msg = update["messages"][0]
                    content_preview = msg.content[:150] + "..." if len(msg.content) > 150 else msg.content
                    print(f"模型回复预览: {content_preview}")
                
                if "skill_context" in update and update["skill_context"] is None:
                    print("  显式清理触发：skill_context 已置空！")
                    
            if node_name == "tools":
                if update.get("skill_context"):
                    print(f"  长文本加载成功（长度: {len(update['skill_context'])} 字符）")
                # 打印工具执行结果
                if "messages" in update:
                    for msg in update["messages"]:
                        if hasattr(msg, "content"):
                            print(f"工具返回: {msg.content}")
    
    print("\n" + "=" * 60)
    print("  测试结束。关键流程回顾：")
    print("1. Agent 看到索引 -> 调用 load_skill")
    print("2. tools 合并节点执行工具 + 自动回溯加载长文本到 State")
    print("3. Agent 第二轮看到长文本 -> 生成最终专业回答")
    print("4. Agent 因无新工具调用 -> 触发清理，skill_context = None")
    print("=" * 60)

  • 子图隔离

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
from typing import TypedDict


# ---------- 1. 父图状态定义 ----------
class ParentState(TypedDict):
    user_query: str
    skill_name: str
    final_answer: str  # 父图只关心这个结果


# ---------- 2. 构建 Skill 子图工厂 ----------
def create_skill_subgraph(skill_name: str, skill_content: str):
    """根据Skill内容动态构建一个独立的子图"""

    # 子图内部可以使用任意复杂的 ReAct 循环
    sub_builder = StateGraph(MessagesState)

    # 子图内部专用模型（可以绑定不同模型或不同参数）
    sub_model = ChatOpenAI(model="gpt-4o")

    # 子图内部系统提示词，直接硬塞完整 Skill 长文本（因为仅在此子图生命周期内有效）
    SUB_SYS_PROMPT = f"""你是专精于该领域的专家。
请严格依据以下技能上下文回答用户问题：
{skill_content}
"""

    def sub_agent_node(state: MessagesState):
        sys_msg = SystemMessage(content=SUB_SYS_PROMPT)
        response = sub_model.invoke([sys_msg] + state["messages"])
        return {"messages": [response]}

    # 若子图内部也需要工具，同样在这里添加 tool_node 即可
    sub_builder.add_node("sub_agent", sub_agent_node)
    sub_builder.add_edge(START, "sub_agent")
    sub_builder.add_edge("sub_agent", END)

    return sub_builder.compile()


# ---------- 3. 父图的执行节点 ----------
def execute_skill_node(state: ParentState):
    # 3.1 根据父图传入的skill_name加载原始内容
    skill_content = load_skill_content(state["skill_name"])  # 复用前面的加载函数

    # 3.2 动态构建子图实例（可按需缓存）
    sub_graph = create_skill_subgraph(state["skill_name"], skill_content)

    # 3.3 调用子图，传入用户问题
    # 注意：子图的输入输出仅限此处定义，外部State不受影响
    sub_result = sub_graph.invoke({
        "messages": [HumanMessage(content=state["user_query"])]
    })

    # 3.4 仅提取子图最终的文字回答，写回父图
    final_text = sub_result["messages"][-1].content
    return {"final_answer": final_text}


# ---------- 4. 编译父图 ----------
parent_builder = StateGraph(ParentState)
parent_builder.add_node("execute_skill", execute_skill_node)
parent_builder.add_edge(START, "execute_skill")
parent_builder.add_edge("execute_skill", END)

parent_graph = parent_builder.compile()

# ---------- 5. 调用示例 ----------
# 调用时，父图内部完全看不到子图的中间推理消息
# result = parent_graph.invoke({"user_query": "上个月销售额?", "skill_name": "sales_analytics"})

3.4.4 状态追踪 Skill 加载记录

  • 目的：明确记录哪些 Skill 已被加载，避免重复加载，并为后续权限判断提供依据
  • 实现：扩展 AgentState，新增 skills_loaded: list[str] 字段
from langchain.agents.middleware import AgentState
from typing import NotRequired

class CustomState(AgentState):
    skills_loaded: NotRequired[list[str]]  # 已加载的skill名称列表

3.4.5 按 Skill 大小选择加载策略

Skill 大小推荐策略
< 1K tokens直接嵌入系统提示词 + 缓存（避免工具调用开销）
1K – 10K tokens按需加载（本文方案）
> 10K tokens分页/检索（如只加载目录，搜索后再取片段）或层级探索

3.5 模式4：Router（路由器）

3.5.1 什么是Router架构

Router架构引入一个专门的 “路由”步骤，负责对用户输入进行分类，并精准地定向到一个或多个专门的智能体

核心特性：

  • 查询分解：Router能分析并拆解复杂查询
  • 并行调用：可以零个或多个专业智能体被并行调用
  • 结果合成：能将来自不同智能体的结果整合成连贯的最终响应

适用场景：

  • 拥有泾渭分明的垂直领域（如代码、文档、内部讨论各自独立）
  • 需要并行查询多个来源来获取完整信息
  • 需要合成多个来源的结果

3.5.2 Router vs Subagents 

Router 是“一次性任务分派器”，而 Subagents 是“拥有记忆的协调者”

  • Router：专门的单次分类步骤，不维护持续对话状态
  • Subagents：维护对话上下文的完整代理，动态决定多轮调用
对比维度Router（路由器架构）Subagents（子代理架构）
核心设计模式Map-Reduce（映射-归约）
先拆分任务并分发，再聚合结果		Supervisor-Worker（监督者-工作者）
主管Agent拥有完整记忆，将子Agent视为“高级工具”进行多轮调度
执行模式单次/静态路由
通常只执行一轮：分类 -> 并行调用 -> 合成。流程结束后即终止		动态/循环调用
主代理（Supervisor）具备推理能力，可根据对话上下文反复、多轮决定调用哪个子工具，直至任务完成
状态与记忆无状态（Stateless）
每次请求独立处理，不记忆历史对话。若需多轮对话，必须将整个Router包装成一个Tool交给上层Agent		有状态（Stateful）
主代理维护完整的对话历史和上下文（通过checkpointer持久化）。子代理虽然是无状态的，但主代理记得之前调用过谁、得到了什么结果
决策机制结构化分类（Classification）
通常使用LLM的with_structured_output或关键词匹配进行分类，决策逻辑相对固定		工具推理（Tool Calling）
主代理通过系统提示词和工具描述，像人类一样“思考”下一步该调用哪个子代理，决策更灵活、更接近自然语言
子代理角色垂直领域专家（知识检索）
子代理通常用于从不同数据源（API、数据库）拉取信息并返回原始搜索结果		任务执行者（操作执行）
子代理不仅查信息，还执行具体业务操作（如创建日历事件、发送邮件），并返回确认结果
用户交互方式间接交互
子代理不与用户直接对话，所有结果汇总到Router的“合成节点”后，统一回复用户		隔离交互
默认子代理不与用户对话（返回给主代理）。但支持在子代理内部使用HumanInTheLoopMiddleware实现人工审查（如发邮件前请求用户确认）
并行执行原生强力支持（Fan-out）
通过LangGraph的 Send API，将一个查询拆解为多个子问题，真正并行调用多个子Agent，互不阻塞		支持并发（需模型配合）
主代理可以在一个回合中同时调用多个子Agent（如果模型支持并行工具调用），但通常更倾向于串行或按需调用
结果合成专门的归约节点（Synthesizer）
有独立的synthesize_results节点，专门负责合并来自不同来源的碎片化信息，去重并生成连贯答案		主代理自然语言总结
主代理接收到子工具返回的文本后，不经过特定归约节点，而是依靠自身LLM能力直接将结果组织成回复
适用场景知识库问答 / 多源检索（RAG）
当你的知识天然分散在多个独立系统（如代码库、文档库、聊天记录）且只需“查询”时		生产力自动化 / 任务编排（Task Automation）
当需要跨系统执行操作（如“建会议+发提醒邮件”），且操作有先后依赖关系时
扩展性弱（需改图结构）
增加新数据源需要修改状态定义（state）和路由逻辑（route_to_agents），侵入性较强		强（约定优于配置）
新增子代理只需在工具列表或SUBAGENTS注册表中添加，主代理代码无需改动（即插即用）
复杂度中等（图结构固定）
流程图清晰（分类->并行->合成），易于调试和监控每个环节的耗时		较高（代理随机性）
主代理的下一步行动由LLM实时生成，具有不确定性，调试时需要查看详细的推理链（Chain of Thought）
典型代码特征 StateGraph + Send + add_conditional_edgescreate_agent + @tool 包装器 + create_supervisor（或自定义工具列表）

3.5.2 路由机制

  • 单一路由（带状态传递 + 兜底）：使用 Command 将请求路由到单⼀、最匹配的专家 Agent
def route_node(state: dict) -> Command:
    query = state["messages"][-1].content
    target = fake_classify(query)
    
    # 传递 update，确保目标节点能拿到 query
    return Command(
        goto=target,
        update={"routed_query": query}  # 显式传递，防止丢失
    )
  • 并行多路由
def parallel_router(state: RouterState) -> List[Send]:
    classifications = classify_for_parallel(state["messages"][-1].content)
    
    # 返回 Send 列表，LangGraph 自动并行调度
    return [
        Send(
            c["agent"], 
            {"messages": [HumanMessage(content=c["query"])]}  # 每个子代理只拿自己的子查询
        ) 
        for c in classifications
    ]
  • 完整示例
第一步：定义状态

class RouterState(TypedDict):
    query: str
    classifications: list[Classification]
    results: Annotated[list[AgentOutput], operator.add]  # 自动合并并行结果
    final_answer: str

第二步：分类节点

def classify_query(state: RouterState) -> dict:
    """分析查询，决定调用哪些Agent"""
    structured_llm = llm.with_structured_output(ClassificationResult)
    result = structured_llm.invoke([
        {"role": "system", "content": "分析此查询，确定需要咨询哪些知识库..."},
        {"role": "user", "content": state["query"]}
    ])
    return {"classifications": result.classifications}

第三步：路由节点（使用Send实现并行）

def route_to_agents(state: RouterState) -> list[Send]:
    """将查询并行发送到多个智能体"""
    return [
        Send(c["source"], {"query": c["query"]})
        for c in state["classifications"]
    ]

第四步：构建工作流

workflow = (
    StateGraph(RouterState)
    .add_node("classify", classify_query)
    .add_node("github", query_github)
    .add_node("notion", query_notion)
    .add_node("slack", query_slack)
    .add_node("synthesize", synthesize_results)
    .add_edge(START, "classify")
    .add_conditional_edges("classify", route_to_agents, ["github", "notion", "slack"])
    .add_edge("github", "synthesize")
    .add_edge("notion", "synthesize")
    .add_edge("slack", "synthesize")
    .add_edge("synthesize", END)
    .compile()
)
​

3.5.3 有状态Router

对于多轮对话，推荐使用 工具包装器 方式：

@tool
def search_knowledge_base(query: str) -> str:
    """跨多个知识源搜索信息"""
    result = workflow.invoke({"query": query})
    return result["final_answer"]

# 创建带记忆的对话智能体
conversational_agent = create_agent(
    model="gpt-4o",
    tools=[search_knowledge_base],
    checkpointer=InMemorySaver(),
)

3.6 模式5：Custom Workflow（自定义工作流）

3.6.1 什么是Custom Workflow架构

自定义工作流架构使用 LangGraph 定义专属的执行流程，即可以直接在任何LangGraph节点中调用LangChain Agent，将自定义工作流的灵活性与预构建 Agent 的便利性结合起来

核心特点：

  • 完全掌控图结构
  • 将确定性逻辑与智能体行为相结合
  • 支持顺序、分支、循环、并行等多种执行模式
  • 可将其他架构模式作为工作流中的一个节点嵌入

适用场景：当标准模式（如子智能体、技能等）无法满足需求时

3.6.2 基础实现

from langchain.agents import create_agent
from langgraph.graph import StateGraph, START, END

# 创建智能体
agent = create_agent(model="gpt-4o", tools=[...])

def agent_node(state: State) -> dict:
    """一个调用LangChain智能体的LangGraph节点"""
    result = agent.invoke({
        "messages": [{"role": "user", "content": state["query"]}]
    })
    return {"answer": result["messages"][-1].content}

# 构建工作流
workflow = (
    StateGraph(State)
    .add_node("agent", agent_node)
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)
Logo
乐奇 Rokid 开放社区

更多推荐

cover

善弈者通盘无妙招：乐奇Rokid没有“灵光一闪”

avatar 乐奇 Rokid 开放社区
cover

极刻觅镜 | AI眼镜日报｜乐奇二代光学升级、Rokid Prism打通桌面智能体

avatar 乐奇 Rokid 开放社区

用 Rokid AIUI 做一个「专注时刻助手」：把一段专注时间挂在眼前

我用“开始专注 25 分钟背单词”和“专注 10 分钟”等输入方式，检查应用是否被正确调起、页面是否收到参数，以及倒计时和任务名是否正确显示。一次典型操作是：先说“开始专注 25 分钟背单词”，确认眼前卡片显示了正确的任务和时间，再选择“开始”。用户可以通过类似“开始专注 25 分钟背单词”的表达唤起智能体，页面接收任务名称和时长，显示待机倒计时。超过单次时长范围时，需要给出清楚的反馈。它没有任务

avatar 乐奇 Rokid 开放社区