一天一个AI项目｜多Agent助手实战
手把手带你拆解一个 GitHub 开源项目，用 LangGraph 从零搭建多 Agent 智能助手。Supervisor 架构 + 5 个专业子 Agent，代码不到 3000 行，适合写进简历的实战项目。
三、架构详解
用户消息 (Telegram / Slack)
|
v
+---------------+
| Manager Agent | GPT-4o + SQLite记忆
+-------+-------+
|
| SendMessage 工具（动态注入）
|
+-----+-----+-----+-----+
v v v v v
消息 日历 待办 协作 搜索
Agent Agent Agent Agent Agent
(每个子Agent: GPT-4o-mini + 专属工具集)
三个核心概念
概念1：Agent = LLM + Tools + Prompt
每个 Agent 都是一个 ReAct Agent（Reasoning + Acting），它会：
- 思考：根据 System Prompt 和用户输入进行推理
- 行动：调用自己的 Tools 去执行操作
- 观察：拿到 Tool 的返回结果
- 循环：直到任务完成
概念2：Supervisor 模式
Manager 不直接做事，它的唯一工具就是 SendMessage——给子 Agent 发消息。它的职责是：
- 理解用户意图
- 决定调用哪个子 Agent
- 如果需要多步骤，按顺序调度多个子 Agent
- 汇总所有结果，回复用户
概念3：动态工具注入
子 Agent 在初始化时并不知道彼此的存在。是 Orchestrator（编排器）在启动时，自动给 Manager 注入了一个 SendMessage 工具，这个工具的参数是根据所有子 Agent 的 name 和 description 动态生成的。
四、核心代码逐行讲解
4.1 Agent 基类
这是所有 Agent 的父类，定义了 Agent 的基本结构：
from langgraph.prebuilt import create_react_agent
from src.utils import get_llm_by_provider
class Agent:
def __init__(self, name, description, system_prompt,
tools, sub_agents, model, temperature, memory=None):
# name: Agent的名字，如 "calendar_agent"
# description: 一句话描述，Orchestrator用它生成路由信息
# system_prompt: 指令提示词，定义Agent的行为边界
# tools: 这个Agent可以调用的工具列表
# sub_agents: 下级Agent列表（只有Manager才有）
# model: 格式为 "provider/model"，如 "openai/gpt-4o"
# memory: 可选的记忆存储（只有Manager用）
self.name = name
self.description = description
self.system_prompt = system_prompt
self.tools = tools
self.sub_agents = sub_agents
self.model = model
self.temperature = temperature
self.memory = memory
self.agent = None # 延迟初始化
def invoke(self, *args, **kwargs):
if not self.agent:
self.initiat_agent() # 第一次调用时才创建
print(f"--- Calling {self.name} ---")
return self.agent.invoke(*args, **kwargs)
def initiat_agent(self):
# 根据 provider/model 格式获取对应的LLM实例
llm = get_llm_by_provider(self.model, self.temperature)
# 用 LangGraph 创建 ReAct Agent
self.agent = create_react_agent(
llm,
tools=self.tools,
state_modifier=self.system_prompt,
# Manager有memory，子Agent设为False避免冲突
**({"checkpointer": self.memory} if self.memory
else {"checkpointer": False})
)
学习要点：
create_react_agent是 LangGraph 提供的工厂函数，一行代码就能创建一个完整的 ReAct Agentstate_modifier就是 System Prompt，定义了 Agent 的角色和行为checkpointer是记忆机制，传入 SQLite 存储就能实现多轮对话
4.2 Orchestrator 编排器
这是整个系统最精妙的部分——它负责让 Agent 之间能互相通信：
from pydantic import Field, create_model
class AgentsOrchestrator:
def __init__(self, main_agent, agents):
self.main_agent = main_agent
self.agents = agents
self.agent_mapping = {} # name -> Agent 的映射表
self._populate_agent_mapping() # 建立映射
self._add_send_message_tool() # 核心：动态注入通信工具
def invoke(self, message, **kwargs):
# 统一入口：把用户消息包装成 LangGraph 的消息格式
messages = {"messages": [("human", message)]}
response = self.main_agent.invoke(messages, **kwargs)
# 取最后一条消息作为最终回复
return response["messages"][-1].content
def _populate_agent_mapping(self):
# 建立 name -> Agent 对象的映射，方便后续路由
for agent in self.agents:
self.agent_mapping[agent.name] = agent
def _create_dynamic_send_message_tool(self, agent):
# 动态生成 SendMessage 工具的输入格式
# 关键：把所有子Agent的name和description拼成提示
recipients_description = "\n".join(
f"{sub.name}: {sub.description}"
for sub in agent.sub_agents if sub.description
)
# 用 Pydantic 动态创建输入模型
DynamicInput = create_model(
f"{agent.name}SendMessageInput",
recipient=(str, Field(..., description=recipients_description)),
message=(str, Field(..., description="要发送的消息内容")),
)
tool = SendMessage(args_schema=DynamicInput)
tool.agent_mapping = self.agent_mapping
return tool
def _add_send_message_tool(self):
# 遍历所有Agent，给有子Agent的（即Manager）注入SendMessage
for agent in self.agents:
if hasattr(agent, "sub_agents") and agent.sub_agents:
tool = self._create_dynamic_send_message_tool(agent)
agent.tools.append(tool)
agent.initiat_agent() # 重新初始化以绑定新工具
这段代码为什么精妙？
- 解耦：子 Agent 完全不知道其他 Agent 的存在，所有路由逻辑由 Orchestrator 统一管理
- 动态性：用 Pydantic 的 create_model 动态生成工具的输入格式，新增子 Agent 零代码改动
- LLM 原生路由：Manager 看到 SendMessage 工具的 description 里列着所有子 Agent 的名字和职责，自然就知道该"发消息"给谁
4.3 组装完整系统
class PersonalAssistant:
def __init__(self):
# === 子Agent区 ===
msg_agent = Agent(
name="msg_agent",
description="处理消息收发相关任务",
system_prompt=MSG_PROMPT.format(
current_date=datetime.now().strftime("%Y-%m-%d"),
current_time=datetime.now().strftime("%H:%M")
),
tools=[ReadMsg(), SendMsg(), FindContacts()],
model="openai/gpt-4o-mini",
temperature=0.1,
sub_agents=[]
)
calendar_agent = Agent(
name="calendar_agent",
description="管理日程和会议",
system_prompt=CALENDAR_PROMPT.format(...),
tools=[GetEvents(), CreateEvent(), FindContacts()],
model="openai/gpt-4o-mini",
temperature=0.1,
sub_agents=[]
)
# todo_agent, collab_agent, researcher_agent 同理...
# === Manager ===
conn = sqlite3.connect("db/checkpoints.sqlite")
manager = Agent(
name="manager",
description="",
system_prompt=MANAGER_PROMPT.format(...),
tools=[], # 不手动加工具！Orchestrator会自动注入
sub_agents=[msg_agent, calendar_agent,
todo_agent, collab_agent, researcher_agent],
model="openai/gpt-4o",
temperature=0.1,
memory=SqliteSaver(conn)
)
# === 编排器把一切串起来 ===
self.orchestrator = AgentsOrchestrator(
main_agent=manager,
agents=[manager, msg_agent, calendar_agent,
todo_agent, collab_agent, researcher_agent]
)
五、快速上手
# 1. 克隆
git clone 仓库地址（GitHub搜索personal-ai-assistant）
cd AI-personal-assistant
# 2. 虚拟环境
python -m venv venv
source venv/bin/activate
# 3. 安装依赖
pip install -r requirements.txt
# 4. 配置
cp .env.example .env
# 编辑 .env，按注释填入你的各项服务配置
# 5. 启动
python app.py
六、文件结构速查
src/
├── agents/
│ ├── base/
│ │ ├── agent.py # Agent基类，所有Agent的模板
│ │ └── agents_orchestrator.py # 编排器，多Agent通信的核心
│ └── personal_assistant.py # 组装入口，把所有Agent串起来
├── channels/ # Telegram/Slack 消息渠道适配
├── prompts/ # 每个Agent的System Prompt
│ ├── manager_agent.py # Manager的指令：如何路由任务
│ ├── calendar_agent.py # 日历Agent的指令
│ └── ...
├── tools/ # 每个Agent的工具实现
│ ├── calendar/ # 查询日程、创建日程
│ ├── research/ # 网页搜索、内容抓取
│ └── ...
└── utils.py # LLM provider 路由
阅读顺序建议：agent.py -> agents_orchestrator.py -> personal_assistant.py -> 挑一个感兴趣的 tools/ 和 prompts/ 看
