架构级重构: - Supervisor节点:分析任务、分解子任务、智能调度Agent - Worker节点:各专业Agent(subgraph)独立执行子任务 - Aggregator节点:并行结果自动聚合 - Send API并行:多Agent同时处理不同子任务 - Agent注册表:AgentRegistry管理5个Agent - weather_agent: 天气专家 - math_agent: 数学专家 - knowledge_agent: 知识专家 - mcp_agent: MCP工具调用 - general_agent: 通用助手(兜底) - 共享State:messages/subtasks/results/final_answer - Supervisor输出JSON格式任务计划(parallel/single/direct)
6.5 KiB
6.5 KiB
黄庄三号四能力 Agent
基于 LangGraph 的多能力 AI Agent,集成 FC / MCP / 思考模式 / Skill 四种核心能力,v3.0 支持多Agent交互。
核心架构:Supervisor + Worker + Aggregator
用户输入 → Supervisor(分析/分解/调度)
│
┌──────────┼──────────┬──────────┐
▼ ▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
│天气 │ │数学 │ │知识 │ │MCP │ ← 各Worker Agent
│专家 │ │专家 │ │专家 │ │Agent │
└──┬───┘ └──┬───┘ └──┬───┘ └──┬───┘
│ │ │ │
└─────────┴─────────┴─────────┘
│
Aggregator(聚合)
│
最终回复
关键能力:
- 并行执行:多个Agent同时处理不同子任务(Send API)
- 串行执行:单Agent直接处理简单任务
- 结果聚合:多Agent结果自动合并为连贯回复
- 智能调度:Supervisor自动分析任务、选择Agent
四项能力
| 能力 | 方案 | 支持度 |
|---|---|---|
| Function Call | LangGraph 原生 ToolNode | ★★★★★ |
| MCP | langchain-mcp-adapters + 确定性路由 | ★★★★ |
| 思考模式 | 自建 think_node + CoT 推理链 | ★★★★ |
| Skill | 自建 SkillRegistry 注册机制 | ★★★★ |
目录结构
hz3-agent/
├── config.yaml ← 一个文件管所有配置(模型、MCP、路由关键词)
├── agent.py ← Agent 主程序,从配置加载
├── mcp_server.py ← MCP 服务器(示例:时间/字符统计/UUID)
├── tools/ ← 工具目录,丢 .py 文件即注册
│ ├── weather.py
│ ├── calculator.py
│ └── knowledge.py
├── skills/ ← 技能目录,丢 .yaml 文件即注册
│ ├── weather_analyst.yaml
│ ├── math_tutor.yaml
│ └── knowledge_explorer.yaml
└── README.md
快速开始
cd hz3-agent
# 自动测试(不带MCP)
python3 agent.py --test
# 自动测试(带MCP)
python3 agent.py --mcp --test
# 交互模式(带MCP)
python3 agent.py --mcp
# 交互模式(不带MCP)
python3 agent.py
新增工具
在 tools/ 目录下创建 .py 文件,暴露 TOOLS 列表即可:
# tools/stock.py
from langchain_core.tools import tool
@tool
def get_stock_price(symbol: str) -> str:
"""查询股票价格"""
return f"{symbol}: 当前价格 123.45"
TOOLS = [get_stock_price] # 必须!供自动扫描
重启 Agent 即自动加载,无需改任何其他文件。
新增技能
支持两种格式:
格式1: .yaml 文件(简洁定义)
在 skills/ 目录下创建 .yaml 文件:
# skills/stock_analyst.yaml
name: stock_analyst
description: "股票分析师 - 查询股价并给出分析"
prompt: |
你是股票分析师。根据以下行情给出分析建议:
行情数据:{input}
tools:
- get_stock_price
格式2: OpenClaw 格式(目录 + SKILL.md + scripts/)
在 skills/ 目录下创建目录,放入 SKILL.md 和 scripts/:
skills/
└── my-skill/
├── SKILL.md ← OpenClaw 标准格式
└── scripts/
└── do_something.py
SKILL.md 格式:
---
name: my-skill
description: "我的自定义技能"
---
# 我的自定义技能
技能的详细说明和使用方法...
Agent 会自动扫描 SKILL.md 的 frontmatter 和 scripts/ 目录,执行脚本并将输出传给 LLM。
路由关键词
无论哪种格式,都需要在 config.yaml 的 skill_keywords 中添加路由:
skill_keywords:
stock_analyst: ["股价", "股票", "行情"]
my-skill: ["关键词1", "关键词2"]
新增 MCP 服务器
在 config.yaml 的 mcp_servers 列表中添加一项:
mcp_servers:
- name: "hz3-tools" # 已有的
command: "python3"
args: ["mcp_server.py"]
transport: "stdio"
route_keywords:
get_current_time: ["几点", "时间"]
count_chars: ["统计字符", "字符数"]
generate_uuid: ["UUID"]
- name: "my-new-server" # 新增的
command: "python3"
args: ["path/to/my_server.py"]
transport: "stdio"
route_keywords:
my_tool: ["关键词1", "关键词2"]
启动时加 --mcp 参数即可连接所有配置的 MCP 服务器。
架构流程
用户输入 → think(思考) → skill_route(路由)
│
┌──────────┴──────────┐
匹配MCP工具 匹配Skill
直接调用 进入skill_exec
│ │
返回结果 执行工具+提示词模板
│ │
└──────────┬─────────┘
│
agent(主节点)
│ │
有tool_calls 无tool_calls
│ │
tools节点 END
│
回到agent
代码调用
from agent import run_agent, build_graph, load_config, scan_tools, scan_skills
config = load_config()
tools = scan_tools()
skills = scan_skills()
graph = await build_graph(config, skills, None, tools)
result = await run_agent("黄庄天气怎么样?", graph)
print(result["reply"]) # 回复
print(result["thinking"]) # 思考过程
print(result["skill"]) # 使用的技能
关键设计决策
- MCP 确定性路由 — 关键词匹配后直接 session.call_tool(),绕过模型不调工具的问题
- 思考结果合并 — 与 system prompt 合并为单条 SystemMessage,避免干扰工具调用
- AsyncExitStack 长连接 — MCP session 在 Agent 生命周期内保持
- 迭代保护 — agent 节点迭代超过阈值强制结束,防死循环
- 配置驱动 — 所有新增/调整通过 config.yaml + 丢文件完成,不改源码
依赖
langgraph>=1.1
langchain-openai>=1.2
langchain-mcp-adapters>=0.2
mcp>=1.0
pyyaml>=6.0