企业官网建设流程全解析

最近很多朋友问我同一个问题：多智能体到底怎么上手？看了不少概念文章，一打开代码库还是懵。我觉得与其啃复杂框架，不如先手搓一个“AI任务调度助手”，把多智能体的核心思路跑通。这篇文章就用LangChain生态里最常用的工具，搭一个能拆解任务、分派Agent、汇总结果的简单调度器。看完你就能理解多智能体的协作逻辑，也能直接跑出一版可用的Demo。

先说清楚这个Demo要解决什么问题。平时我们调大模型，基本是一个Agent从头干到尾。比如让它写一份行业分析报告，它又要联网查资料，又要整理数据，还要润色文字，最后出来的内容常常顾此失彼。多智能体的思路是：把一个大任务拆成多个子任务，每个子任务交给一个专职Agent去处理——有负责调研的，有负责写初稿的，有负责审核的。调度器（Orchestrator）负责拆任务、发任务、收集结果。这种结构的好处是职责清晰，每个Agent只干一件事，模型就不容易“精神分裂”，而且可以针对不同子任务选不同的模型或提示词。

我选的实现路线是LangGraph，不是LangChain早期的AgentExecutor。LangGraph是LangChain团队专门做Agent编排的库，它把Agent流程定义成一张图：节点是Agent的动作，边是节点之间的流转，状态是节点间传递的数据。这种“状态图”的方式特别适合调度场景，因为你可以清楚看到任务流转到哪一步、状态里存了什么，出问题也容易排查。

1. 整体设计与思路拆解

1.1 为什么多智能体反而比单个大模型好用

单个大模型Agent不是不能用，而是在处理复杂任务时有几个痛点。第一，上下文容易被污染。一个Agent既要处理用户需求，又要调用工具，还要根据中间结果调整策略，所有信息都堆在一个上下文窗口里，跑着跑着就忘了前面的关键约束。第二，提示词写起来非常纠结。想让一个Agent既懂“拆任务”又懂“调工具”又懂“汇总报告”，这个系统提示词可能要上千字，而且模型很容易顾此失彼。第三，调试困难。任务哪一步出错了，只能看一长串对话历史，定位问题很痛苦。

多智能体架构把这些问题拆开了。每个Agent只维护一小段上下文，提示词短小专注，一个环节出问题，只看那个Agent的状态即可。像我这版调度助手，就分了三个角色：任务规划Agent负责拆解需求、执行Agent负责具体干活、质检Agent负责检查结果。三个Agent各司其职，调度器只在它们之间传状态，不干涉具体执行。这样跑起来之后，我可以单独调试任何一个环节，比如觉得写稿质量不好，就只优化写稿Agent的提示词，其他都不用动。

1.2 选LangGraph做编排的理由

如果你去搜“LangChain多智能体”，会看到很多旧教程还在用initialize_agent或者AgentExecutor。坦白说那套东西做简单问答还行，做调度就力不从心了。AgentExecutor是一个封装好的黑盒，你很难在里面插入“先检查再决定下一步”的逻辑。后来LangChain团队推出LangGraph，思路变成“把Agent流程画成图”，这对调度场景是降维打击。

LangGraph里有两个核心概念：State（状态）和Graph（图）。State是一个全局数据对象，所有节点都能读写它；Graph就是节点和边的连接。调度器运行时，从入口节点开始，按边走到下一个节点，直到走到END节点。每个节点可以是一个Python函数，也可以是一个内置的Agent节点。节点执行完之后返回一个字典，更新State。边可以是普通边（无条件跳转），也可以是条件边（根据当前State内容决定跳转到哪个节点）。这套表达能力和灵活性比AgentExecutor强太多了，而且调试时能看到完整的运行轨迹。

1.3 调度助手的架构长什么样

我这版Demo的流程图，用文字描述大概是这样的：入口节点接收用户输入，把任务文本存进State；然后进入规划节点，让大模型拆解任务，输出一个子任务列表，每个子任务包含任务ID、任务描述、负责角色；接下来进入执行阶段，每个子任务落到对应的Agent节点上执行；所有子任务执行完后，进入汇总节点，把各Agent的输出整理成最终答复。

为了让新手好理解，我故意没做复杂的并行分发。实际项目中可以让多个Agent并行执行，但在入门Demo里，串行执行更容易看清每个节点的输入输出。我在设计时刻意加了一个“人工审核”节点：规划节点拆解完任务后，先不急着执行，而是把任务清单返回给用户，用户确认后再往下走。这样既演示了LangGraph的中断与恢复机制，也防止模型拆出奇怪的任务列表浪费token。这个设计思路来自实际踩坑——模型自己拆任务，很容易拆出不可执行的子任务，加一道人工确认，效果立刻不一样。

2. 核心细节解析与实操要点

2.1 这些概念先搞清楚再动手

在写代码之前，有几个LangGraph的高频概念需要先有感觉。第一个是State，它决定节点间怎么传数据。你用TypedDict定义一个状态结构，比如{"input": str, "tasks": list, "results": dict}。所有节点函数接收一个state参数，返回一个更新字典，框架会自动合并。第二个是节点，节点就是普通Python函数，签名是(state) -> dict。需要注意节点返回的字典不会覆盖整个State，只会更新你指定的key。第三个是边。普通边用add_edge，表示执行完节点A必然去节点B。条件边用add_conditional_edges，你需要传一个路由函数，函数根据当前State计算一个分支名，框架根据分支名找到对应的目标节点。第四个是状态记忆，LangGraph默认会把State传给每个节点，但不会自动保存历史，如果你需要多轮对话记忆，要在State里加一个messages字段专门存历史消息。

2.2 模型与工具怎么选型

这版Demo不绑定云厂商，我推荐用兼容OpenAI接口的本地推理服务，比如你在本地启动一个模型服务，然后通过ChatOpenAI(base_url="http://localhost:11434/v1", api_key="not-needed", model="local-chat-model")来调用。这样做的原因是：入门阶段调试频率很高，如果每调一次都走API付费，心理上压力大，而且本地模型的响应延迟低，调试体验更流畅。不过要注意本地模型的能力参差不齐，规划Agent这种需要结构化输出的环节，建议选指令跟随能力强的模型。

工具方面，执行Agent可以用内置的TavilySearchResults做联网检索，也可以写一个简单的file_tool读写本地文件。我这版Demo里做了一个自定义工具叫now_time，用来返回当前时间。你别小看这个工具，很多模型对“当前时间”没有概念，写日报类任务时总会瞎编日期。给Agent挂一个取当前时间的工具，单位时间内出错的概率立刻降下来。这其实说明了一个道理：多智能体系统里，工具不在多，而在准。

2.3 结构化输出是调度器的生命线

在多智能体调度里，最怕的就是模型返回一段乱七八糟的文本，下游节点没法解析。规划节点的输出必须是结构化的子任务清单，否则没法分派。所以我会给规划节点用强格式约束——要么用Pydantic的with_structured_output，要么在提示词里明确要求JSON格式，再加一个JSON解析兜底函数。实测中，提示词写得好不好直接决定解析成功率。我总结了一个比较稳的JSON提示词模板：要求模型只输出JSON，不要输出任何解释；提供JSON示例；字段用蛇形命名；明确说“如果不符合条件，返回空列表而不是报错”。

除了模型输出，节点之间的消息格式也要统一。我这版里每个Agent返回的output字段固定是字符串字符串，汇总节点直接拼接。如果你想传更富的数据，比如置信度、来源链接，就用字典或Pydantic模型，并写在State定义里。切忌一会儿返回字符串一会儿返回字典，排查问题时会疯掉。

3. 实操过程与核心环节实现

3.1 环境安装与目录结构

先创建一个干净的虚拟环境，然后安装依赖。我用的是langgraph，它会自动带起langchain-core等相关包。命令就一条：

pip install langgraph langchain-core langchain-openai

如果你要用联网搜索工具，再加一个tavily-python，并设置API Key。不用工具也能跑，Demo可以先用一个模拟工具替代。

我的项目结构很简单，只有两个文件：state.py放状态定义，agent.py放节点和构图逻辑。新手先不要学别人一上来搞十几个文件，单个文件跑通再拆分。

3.2 定义状态结构

我这样定义State：

from typing import TypedDict, List, Dict, Optional class SubTask(TypedDict): task_id: str desc: str owner: str status: str class AgentState(TypedDict): user_request: str tasks: List[SubTask] current_task: Optional[SubTask] results: Dict[str, str] error: Optional[str]

这个结构里的核心是results，它是一个字典，key是子任务ID，value是Agent执行后的文本结果。汇总节点只认这个字典，不用关心中间过程。current_task字段是给执行节点用的，每次从tasks里取一个未处理的任务放进去，执行完再写回results。error字段用来记录错误信息，方便条件边判断是否需要重试。

3.3 写规划节点

规划节点的核心工作是把用户请求变成子任务清单。我用LangChain的with_structured_output来做，感觉比手动解析JSON稳得多。定义好子任务的数据结构：

from pydantic import BaseModel, Field class SubTaskOut(BaseModel): task_id: str = Field(description="任务编号，例如 T01") desc: str = Field(description="任务描述") owner: str = Field(description="负责角色名称") class TaskPlan(BaseModel): tasks: List[SubTaskOut] = Field(description="子任务清单")

接着写规划节点函数：

from langchain_openai import ChatOpenAI system_planner = """ 你是一个项目规划助手。用户会给你一个完整需求，你需要把它拆解为多个独立的子任务。 要求： 1. 每个子任务只包含一个明确的动作。 2. 子任务数量控制在3到5个。 3. 角色限定为：research 或 write 或 review。 4. 最终汇总不需要单独拆任务。 只输出结构化结果，不允许额外解释。用户需求：{request} """ def plan_node(state: AgentState) -> dict: llm = ChatOpenAI(model="local-chat-model", temperature=0) structured_llm = llm.with_structured_output(TaskPlan) prompt = system_planner.format(request=state["user_request"]) plan = structured_llm.invoke(prompt) tasks = [SubTask(task_id=t.task_id, desc=t.desc, owner=t.owner, status="pending") for t in plan.tasks] return {"tasks": tasks}

这里把temperature设为0，是为了让模型减少发散，输出更可预测。实际对比过，temperature=0.7时偶尔会拆出玄幻的任务描述，0就很稳。

3.4 写执行节点和质检节点

执行节点不区分是research还是write，统一走一个通用执行函数。判断标准就看owner字段：如果是research，就调用搜索工具；如果是write，就让模型基于已有资料写文本；如果是review，就检查文本长度和关键词。

def execute_node(state: AgentState): if not state.get("tasks"): return {"error": "no tasks"} for t in state["tasks"]: if t["status"] == "pending": state["current_task"] = t if t["owner"] == "research": output = do_research(t["desc"]) elif t["owner"] == "write": output = do_write(t["desc"]) else: output = do_review(t["desc"]) state["results"][t["task_id"]] = output t["status"] = "done" break return {"results": state["results"], "tasks": state["tasks"]}

这样写有一个好处：一次执行一个子任务，下次再进入节点时继续处理下一个，逻辑简单且不容易死循环。do_write其实就是一次普通的大模型生成调用，do_review可以简单检查字数，超过500字就算过。

3.5 组装LangGraph并加入人工确认

最关键的是构图。我用了StateGraph，然后在规划节点之后加了一个条件边，根据用户输入的“是否自动执行”标志决定是去执行还是返回给用户确认。如果你在终端环境跑，用户确认要修改State，LangGraph提供了一个interrupt_before参数，可以在进入执行节点前中断，然后你手动修改State再恢复运行。

from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver g = StateGraph(AgentState) g.add_node("planner", plan_node) g.add_node("executor", execute_node) g.add_node("summarizer", sum_node) g.set_entry_point("planner") g.add_edge("planner", "executor") g.add_conditional_edges("executor", check_remaining_tasks, {"continue": "executor", "done": "summarizer"}) g.add_edge("summarizer", END) app = g.compile(checkpointer=MemorySaver())

关键在于check_remaining_tasks：

def check_remaining_tasks(state): if any(t["status"] == "pending" for t in state["tasks"]): return "continue" return "done"

这样执行节点会反复进入，直到所有任务做完，才进入汇总。如果你想要人工确认，可以在compile时传interrupt_before=["executor"]，然后在调用invoke后访问app.get_state(config)，修改state后调用app.update_state(config, values)，再继续invoke。这个机制我第一次用的时候觉得很绕，后来发现它其实是LangGraph给我们留的“安全刹车”。

3.6 总结节点与整体运行效果

汇总节点就是把state["results"]里的各段输出整合一下，有时也会让大模型把多个结果转成一篇连贯的最终报告。我这里是自己拼接加简单润色，因为初版要尽量少依赖大模型，跑得快一些。

def sum_node(state: AgentState): parts = [state["results"][t["task_id"]] for t in state["tasks"]] final_text = "\n\n".join(parts) return {"final_answer": final_text}

运行一个例子看看效果。用户输入“帮我整理一篇关于本地模型推理加速的调研短文”，规划节点拆出来的任务大概是“搜集相关工具及性能数据”“撰写调研短文”“检查字数是否满足要求”。执行节点分别处理后，汇总节点把三块内容拼起来。整体跑下来耗时可能比单个Agent长一点，但每一段内容的质量是稳定的，不会出现“前面还在讲推理加速，后面突然扯到部署成本”这种跑题情况。

4. 常见问题与排查技巧实录

4.1 模型不遵守JSON输出指令

这个问题我几乎每次做多智能体项目都会遇到。要么是模型输出里带着解释性文字，要么是JSON字段名不对。解决办法有两个方向：第一，换更强指令跟随的模型，如果本地模型达不到要求，就在结构化输出后面加一个兜底解析函数，用正则把第一个{到最后一个}切出来再json.loads；第二，在提示词里把“不遵守格式的后果”写清楚，比如“任何非JSON输出都会被系统丢弃，你将拿不到任何报酬”，实测下来威慑力比“请遵守JSON格式”管用。

4.2 调度器一直在执行节点里打转

这个问题多半出在状态更新逻辑上。比如忘了把已处理的任务改成done，或者check_remaining_tasks判断条件写反了。我排查时会在每个节点入口打印一下State的关键字段：

print(">>>", state.get("current_task"), state.get("results"))

确认任务状态是否更新。还有个隐蔽坑：LangGraph的State更新是按key覆盖的，如果你在节点里直接修改了传入的列表对象而没有返回新对象，框架可能感知不到变化。所以返回时要显式返回tasks和results。

4.3 多个Agent上下文互相污染

如果你把多个Agent塞进同一个State里，彼此历史消息混在一起，很容易造成模型行为错乱。我用的隔离技巧是：每个Agent维护自己的消息列表key，比如research_messages、write_messages，节点之间互不相干。汇总节点也不去读这些列表，只读最后的结果字符串。这样设计虽然牺牲了一点信息传递效率，但换来了调试时极大的清爽感。

4.4 性能太慢怎么办

多智能体一定比单Agent慢，因为模型被调用了多次。如果你觉得Demo响应时间长，优先看两点：第一，是不是本地模型跑在CPU上，如果是，换GPU或换个小一点的量化模型；第二，是不是子任务拆得太多，控制在3个以内能显著提速。还有一个技巧是并行化，模型支持并发的话，把执行节点改造成线程池并发执行，整体耗时能压缩到只比单任务慢一点。但入门阶段不推荐并行，因为并发引入的竞态问题特别不好排查。

5. 从Demo到真正可用的任务调度助手

5.1 加一个简单记忆功能

目前的Demo是无状态的，用户每次提问都是全新上下文。实际使用时，多轮对话很重要。LangGraph支持MemorySaver，但那是给调度器存状态用的，不是模型记忆。模型记忆需要你自己把历史对话传给模型。我这版Demo里在写Agent的系统提示词里插入了历史要点列表，每次执行任务前把上几轮的关键结论传进去。注意不要传原始对话，模型会分不清新旧任务；传经过压缩的要点会更稳。

5.2 接入团队协作场景

这个调度助手稍加改造，就能模拟一个“内容生产线”：策划Agent出选题，调研Agent找资料，写作Agent出初稿，校对Agent改病句。每个Agent可以属于不同租户或者独立服务，LangGraph支持节点函数内部调用其他Agent服务。这样其实就变成了一个简单的“Agent工作流引擎”。我自己跑下来的经验是：先把流程画在纸上，再翻译成StateGraph，顺序不要反。直接写代码的话，很容易漏掉某条边。

5.3 监控、容错与重试

生产环境里Agent一定会失败。你可以给每个执行节点包一层try-except，当某个子任务失败时，把错误信息写入error字段，条件边收到错误后走一个“修复节点”。修复节点的逻辑很简单：让模型根据错误信息重写任务描述，最多重试两次，还不行就把任务标记为失败，直接返回部分结果。实测这个机制能挽回大概三成意外报错——比如模型临时返回空内容，重写任务描述后往往能正常执行。重试要限制次数，否则像网络抖动那种错误会白白消耗大量时间。

6. 写在最后的几点体会

我自己做完这个Demo之后最大的感受是：多智能体并没有想象中那么玄乎，本质就是“把流程拆成节点，让模型在每个节点做一件事”。难的不是LangGraph的API，而是任务拆分的边界感。哪些事该让模型决策，哪些事该用代码判定，这个度需要反复调。建议你第一次跑的时候，故意试几个模糊需求，比如“帮我处理一下这个数据”，看看规划节点怎么拆——它会提醒你，模型不擅长处理信息不完整的请求，所以调度器最好在入口做一个需求澄清的节点，那又是下一个迭代了。

最后分享一个小技巧：如果你在论文或业务项目里要用多智能体，不要一开始就上“多模型多工具多团队”的豪华配置，先跑通一个三节点的最小闭环，然后把你想加的功能逐个贴在边上测试。多智能体系统的问题绝大多数出在数据流上，不在模型能力上。状态设计得清楚，这个系统就成功了一大半。