从零搭建AI任务调度助手：LangGraph多智能体编排指南

你有没有遇到过这种情况：想让一个模型同时搞定"写脚本""改bug""写使用说明"三件事，结果它顾此失彼，代码里夹着解释文字，文案里又带着代码片段，最后两边都不像样。我一开始也是这么干的，直到我把任务真正拆开，才意识到多智能体的价值不是炫技，而是把一个复杂需求拆给不同"岗位"去并行思考。这篇文章就带你从零做一个简单的AI任务调度助手，基于LangChain生态里的 Agent 和图形编排能力，把"调度者、执行者、质检员"跑通一条完整链路。

整篇文章适合两类人：一类是刚接触 LangChain，想知道 Agent 怎么串起来；另一类是被单一大模型"一个指令干到底"折磨过，想看看多智能体到底怎么落地的新手。我不会把代码堆成一堵墙，而是会把每个设计决策的原因讲清楚，顺手把调试时踩过的坑也记下来。

1. 先聊聊为什么单个大模型干不好"任务调度"

1.1 一个模型身兼数职，指令会互相干扰

你可能会问：我直接给一个大模型写一段很长的提示词，让它"先写代码，再写说明，最后自查"，不也是一种调度吗？表面上可以，实际跑几次就发现问题了。大模型没有真正的"工作记忆"，它只是根据上下文里的所有内容生成下一个 token。当代码任务和文案任务挤在同一段上下文中，模型很容易把两种风格混在一起：写代码时突然冒出一句"这段逻辑很清晰"，写文案时又插入代码块。还有更麻烦的，如果某个子任务执行失败，你要回退重跑，单一大模型往往无法判断"到底该重写哪部分"，只能整段重新生成。

我自己遇到最典型的一个场景，是让模型"写一个Python脚本统计CSV缺失值，并同时生成说明文档"。模型确实给出了完整回复，但脚本里的注释是给非技术人员看的营销文案风格，文档里又贴了一大段源码。分开看好像每个部分都对，合在一起就是没法用。这就是"角色打架"。

1.2 多智能体不是玄学，本质是把流程拆给不同大脑

多智能体的思路其实很像现实中的团队分工：调度者负责接收需求、拆解任务、判断下一步交给谁；执行者各有专长，一个写代码，一个写文案，互不干扰；质检员最后检查产物是否满足要求，不合格就打回重来。每个智能体只关注自己的窄范围，提示词短了，指令冲突少了，上下文也更干净了。

从工程角度看，多智能体的价值在于可控性和可观测性。单一大模型的调用像一个黑盒，你很难知道它内部到底先做了什么；而多智能体系统里，每个节点是什么、输入什么、输出什么，都是你自己定义的。哪怕中间某个 Agent 崩了，你也知道是哪一步崩的，改起来也快得多。

在 LangChain 生态里，实现多智能体一般有两种路线：一种是让一个 Agent 持有"其他 Agent 的工具"，自己做路由；另一种是用 LangGraph 这类编排框架，把每个智能体做成图里的一个节点，节点之间用条件边连接。我这个项目选的是后者，因为条件路由更直观，而且对初学者来说，图结构比隐藏的 Agent 工具链更容易调试。

2. 环境准备与LangChain生态里的现实坑

2.1 安装与版本选择：LangChain不是装一个就完事

很多新手第一次装 LangChain 时会被一堆包搞晕。现在 LangChain 已经不是单个库，而是拆成了很多配套包。如果你只装了langchain，会发现很多 Agent 相关功能根本用不了。我这个项目里实际需要这么几个：

pip install langgraph langchain langchain-openai langchain-community langchain-core

langgraph是负责编排多智能体流程的核心，langchain-openai是让 LangChain 能调用 OpenAI 兼容格式的接口，langchain-community里装着社区维护的一堆工具集成。不装langchain-core可能会缺基础类型，虽然有些包会自动带上，但显式装一遍能省掉不少版本冲突的烦恼。

版本方面一定要留个心眼：LangGraph 从 0.2 开始 API 有过一些调整，有些教程里写的老写法会直接报错或者被废弃。安装的时候可以在包名后给定版本范围，例如：

pip install "langgraph>=0.2" "langchain>=0.3"

提示：如果后面发现某个函数调用报AttributeError，先别急着怀疑自己代码，很可能是某个依赖被升级到了不兼容的新版本。检查一下pip list里的版本，再对照 langgraph 的变更说明，基本能定位。

2.2 模型接入：随便一个大模型API都能当"大脑"

这个项目里所有智能体本质上都是对同一个大模型 API 的多次调用，只不过每次用的系统提示词不一样。所以我只需要一个能通过OpenAI 兼容接口访问的模型服务就行，不限定具体厂商。用环境变量管理密钥和地址，这样代码换模型时不用改动逻辑部分：

import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm = ChatOpenAI( base_url=os.getenv("LLM_BASE_URL"), api_key=os.getenv("LLM_API_KEY"), model=os.getenv("LLM_MODEL_NAME"), temperature=0.1, )

把LLM_BASE_URL、LLM_API_KEY、LLM_MODEL_NAME写到.env文件里就行。我用的是一个中规模的通用大模型，上下文长度中等，响应速度还不错。如果模型支持结构化输出，调度节点的准确率会高不少；如果模型不支持，后面我会说怎么用提示词约束加解析兜底来处理。

这里有一个跟新手相关的关键点：temperature我设置成 0.1，因为调度节点的任务是对接路由决策，不需要创意，需要确定性。写文案的 Agent 可以把温度调高到 0.4 甚至 0.7。同一份代码里，不同节点使用不同温度完全没问题，只要在定义节点时给不同的 LLM 实例就行。

3. 这个任务调度助手的设计：调度者、执行者、质检员

3.1 状态设计是第一步：先想清楚每个智能体之间传什么

搭多智能体之前千万别直接写代码，第一个要考虑的是状态（State）。多智能体运行相当于一个状态机，每个节点读入状态，执行完毕后又把新状态传给下一个节点。状态里放什么字段，直接决定了每个 Agent 到底能看到多少信息。

我这套助手的核心字段如下：

有些字段你可能看不出用途，但它们在调试时特别重要。比如step_count，如果调度器出了问题，会在节点之间疯狂往返，没有这个计数器，整个程序就跑死了也不自知。我会在调度节点的提示词里强制要求："当 step_count 大于 3 时，必须输出final动作"。

还要考虑一个问题：code_result和doc_result都是直接覆盖还是保留历史？我这个 demo 里选择直接覆盖，因为每次重新生成时，质检员的意见已经通过assignment传给了执行者，不需要保存多轮历史状态。如果你要迭代很多次，最好改成列表类型，保留每一步的产物，方便追溯和对比。

3.2 三条角色链路与回退机制

我设计了三类节点：

• 调度节点（dispatcher）：读用户 query 和已有产出物，输出 JSON，决定下一步是去写代码、写文案，还是直接收尾。
• 执行节点：代码 Agent 负责写脚本并运行验证；文案 Agent 负责写说明文档、使用指南或解释段落。
• 质检节点（inspector）：汇总已完成的产出，判断质量是否达标，不达标就带着反馈意见回到调度节点重新指派。

这个流程是一个典型的"规划-执行-审查"环。第一次跑通的时候，最容易困惑的是：为什么代码 Agent 执行完还要再回到调度节点，而不是直接进质检节点？因为用户的需求可能是"先写代码再写文案"，也可能是"只需要文案"。如果代码 Agent 执行完就直接进质检，那一份没有文案的产出根本没法审查。所以我的图里，执行节点回环到调度节点，由调度节点判断下一个该做什么；当所有子任务都完成，调度节点才会给出final动作，进入质检环节。

质检环节也支持回退：如果质检不通过，它会带着inspect_feedback回到调度节点，调度器把反馈写进新的assignment，再指派对应的执行者改一轮。这里需要留意的是，回退次数必须有限制，否则一个永远改不对的任务会把整个流程拖死。我的做法是把回退次数换算进step_count，超过阈值就让质检员网开一面直接放行。

4. 核心代码逐段拆解：状态图、路由与Agent实现

4.1 基础设施：状态类型、模型初始化与Python执行工具

代码从定义状态类型开始。我使用TypedDict来描述状态结构，这样写节点函数时能获得 IDE 提示，不容易因拼写错误翻车：

from typing import TypedDict class AssistantState(TypedDict): query: str next_action: str assignment: str code_result: str doc_result: str inspect_feedback: str step_count: int final_answer: str

然后是 Python 执行工具。代码 Agent 生成脚本后，不能光停在"我写好了"这一步，最好真跑一遍看看有没有语法错误。我给 Agent 配一个自定义工具，用exec执行传入的代码片段并捕获输出：

import io import contextlib from langchain_core.tools import tool @tool def run_python_snippet(code: str) -> str: """ 在本地进程内执行一段Python代码，返回标准输出与错误信息。 仅用于demo验证短脚本，生产环境务必放到隔离容器里。 """ stdout_buf = io.StringIO() stderr_buf = io.StringIO() try: with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): exec(code, {"__name__": "__main__"}) except Exception as exc: return f"运行异常: {type(exc).__name__}: {exc}\n标准输出:\n{stdout_buf.getvalue()}" return f"标准输出:\n{stdout_buf.getvalue()}\n标准错误:\n{stderr_buf.getvalue()}"

警告：直接在本地exec运行代码是很危险的做法，尤其当代码 Agent 的输出不可信时。这个 demo 只是教学用，我建议你在沙箱容器里跑。即使如此，代码 Agent 的执行结果也要当作外部输出看待，不要盲目信任。

4.2 调度节点的实现（负责拆解与路由）

调度节点本身不需要工具，它就是一次带结构化输出要求的 LLM 调用。我会让模型返回一个 JSON，包含next_action和assignment两个字段。

调度器的系统提示词需要把可选项讲清楚：code表示去写代码，doc表示去写文案，final表示所有产出已完成，准备交给质检。同时还要提醒模型参考已有的code_result和doc_result，避免重复劳动。

from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate DISPATCH_PROMPT = """你是任务调度器，负责拆解用户需求并安排执行者。 可选动作： - code: 让代码Agent去编写并验证Python代码 - doc: 让文案Agent去撰写说明文档或文案 - final: 所有必要产物都已生成，进入质检环节 用户需求：{query} 已有代码产物：{code_result} 已有文案产物：{doc_result} 当前执行计数：{step_count} 规则： 1. 如果用户需要代码而代码产物为空，优先选择code。 2. 如果用户需要说明文档而文案产物为空，选择doc。 3. 如果步骤超过3轮，不管怎样都选择final。 4. 输出JSON格式：{"next_action": "code", "assignment": "给执行者的子任务描述", "reason": "简单说明理由"} 只输出JSON，不要输出其他内容。 """ parser = JsonOutputParser() def dispatch_node(state: AssistantState) -> dict: if state.get("step_count", 0) > 3: return { "next_action": "final", "assignment": "所有任务已经执行过多轮，请质检后直接汇总", "step_count": state.get("step_count", 0) + 1, } prompt = ChatPromptTemplate.from_template(DISPATCH_PROMPT) chain = prompt | llm | parser result = chain.invoke({ "query": state["query"], "code_result": state.get("code_result", "（还没有代码产物）"), "doc_result": state.get("doc_result", "（还没有文案产物）"), "step_count": state.get("step_count", 0), }) return { "next_action": result.get("next_action", "final"), "assignment": result.get("assignment", ""), "step_count": state.get("step_count", 0) + 1, }

这个实现里有一个很容易忽略但很重要的点：我利用step_count作为硬性逃生阀，即使大模型输出的next_action一直不老实，超过 3 轮也会被强行改成final。

4.3 执行者节点：代码Agent和文案Agent

代码 Agent 用 LangChain 的create_react_agent创建，让它能调用run_python_snippet工具来实际执行生成的脚本。系统提示词限定它的职责范围：只写代码、只验证代码，不要写说明文档；代码需要直接能跑，并给出必要注释。

from langchain.agents import create_react_agent from langchain_core.prompts import PromptTemplate CODE_AGENT_TMPL = """你是一名Python工程师。你只负责根据任务描述编写可运行的Python代码， 并用run_python_snippet工具运行验证。不要把任务描述之外的文案混进来。 任务描述： {assignment} 要求： - 代码要简洁，包含必要的注释。 - 先调用run_python_snippet运行一次，即使报错也要把错误信息原样反馈。 - 最终回复时先给完整代码，再附一行执行结果摘要。 """ code_agent_prompt = PromptTemplate.from_template(CODE_AGENT_TMPL) code_agent_runnable = create_react_agent(llm, [run_python_snippet], prompt=code_agent_prompt)

执行节点里的工作就是把这个 Agent 跑起来，并把最终输出塞回状态里：

def code_agent_node(state: AssistantState) -> dict: result = code_agent_runnable.invoke({"assignment": state.get("assignment", "")}) return {"code_result": result["output"]}

文案 Agent 其实不需要工具，我用一个普通的链来生成文案：

DOC_PROMPT = """你是文档撰写员。你不修改代码，只根据已有材料撰写用户能看懂的说明文字。 要求：语气清晰、步骤明确、不要空话。 已有代码产物： {code_result} 任务描述： {assignment} 请输出说明文档正文。 """ def doc_agent_node(state: AssistantState) -> dict: prompt = ChatPromptTemplate.from_template(DOC_PROMPT) chain = prompt | llm result = chain.invoke({ "code_result": state.get("code_result", "（暂无代码产物）"), "assignment": state.get("assignment", ""), }) return {"doc_result": result.content if hasattr(result, "content") else str(result)}

可以看到，文案 Agent 的输入里包含了code_result。这是多智能体协作的关键：上一个角色的输出成为下一个角色的输入上下文。如果把这一步省掉，文案 Agent 就成了"盲写"，写出来的说明和实际代码可能对不上。

4.4 质检节点与最终汇总

质检节点的职责是站在用户视角看整体结果。它对代码和文案做一次通读，输出两个判断字段：pass或fail，以及反馈意见。实现上依然是给 LLM 一个结构化的输出要求。

INSPECT_PROMPT = """你是质量审查员。请检查以下代码和文案是否匹配、是否完整、是否有明显错误。 代码产物： {code_result} 文案产物： {doc_result} 用户需求： {query} 输出JSON，格式： {"verdict": "pass" 或 "fail", "feedback": "你的具体审查意见"} 只输出JSON。 """ def inspect_node(state: AssistantState) -> dict: prompt = ChatPromptTemplate.from_template(INSPECT_PROMPT) chain = prompt | llm | JsonOutputParser() result = chain.invoke({ "code_result": state.get("code_result", ""), "doc_result": state.get("doc_result", ""), "query": state["query"], }) verdict = result.get("verdict", "fail") feedback = result.get("feedback", "") return {"next_action": verdict, "inspect_feedback": feedback}

汇总节点的任务就简单多了，把代码和文案拼成最终答案：

def aggregate_node(state: AssistantState) -> dict: final_answer = ( f"### 代码部分\n\n{state.get('code_result', '')}\n\n" f"### 文案部分\n\n{state.get('doc_result', '')}\n\n" f"### 质检意见\n\n{state.get('inspect_feedback', '')}" ) return {"final_answer": final_answer}

4.5 把图组装起来并跑一个完整例子

现在到了最关键的一步，用 LangGraph 把这些节点连成一张图。有四个节点需要分别处理：dispatch、code_agent、doc_agent、inspect、aggregate。连接规则是：

• dispatch根据next_action无条件选择去code_agent、doc_agent还是inspect。
• code_agent结束后回到dispatch，让它继续判断是否需要doc_agent。
• doc_agent结束后同样回到dispatch。
• inspect根据next_action判断是回dispatch继续修改，还是去aggregate收尾。

from langgraph.graph import StateGraph, END graph = StateGraph(AssistantState) graph.add_node("dispatch", dispatch_node) graph.add_node("code_agent", code_agent_node) graph.add_node("doc_agent", doc_agent_node) graph.add_node("inspect", inspect_node) graph.add_node("aggregate", aggregate_node) graph.set_entry_point("dispatch") graph.add_conditional_edges( "dispatch", lambda state: state.get("next_action", "final"), { "code": "code_agent", "doc": "doc_agent", "final": "inspect", }, ) graph.add_edge("code_agent", "dispatch") graph.add_edge("doc_agent", "dispatch") graph.add_conditional_edges( "inspect", lambda state: "aggregate" if state.get("next_action") == "pass" else "dispatch", { "aggregate": "aggregate", "dispatch": "dispatch", }, ) graph.add_edge("aggregate", END) app = graph.compile()

然后就可以用一句话来测试整个调度助手：

result = app.invoke({ "query": "帮我写一个Python脚本，读取data.csv并统计每列缺失值数量，然后用中文写一段脚本使用说明。", "next_action": "", "assignment": "", "code_result": "", "doc_result": "", "inspect_feedback": "", "step_count": 0, "final_answer": "", }) print(result["final_answer"])

第一次跑通的时候，你会看到 LangGraph 打印出一串节点执行日志，比如dispatch -> code_agent -> dispatch -> doc_agent -> dispatch -> inspect -> aggregate，这个顺序正是调度器自己规划出来的。这种感觉挺奇妙的——你并没有在代码里写死流程，而是让模型在几个选项之间做选择，最终走出一条合理的执行路径。

5. 第一次跑往往不会顺利：我踩过的三个实际问题

5.1 路由结果不够"结构化"带来的解析危机

第一次搭建时，我让调度器直接输出文本，然后在代码里用if "code" in output这种办法判断下一步。听起来很省事，实际上一团糟。模型可能输出"我们应该先 code，然后 doc"，也可能输出"CODE"，大小写和格式都不稳定，判断逻辑越写越脏。

后来我改成用JsonOutputParser，明确要求只输出 JSON。结果在小模型上又出现一个问题：模型不遵守指令，输出的 JSON 前面多了一句解释性文字，或者直接把引号写弯了，解析器照样报错。解决思路有两个：一是换用支持response_format={"type": "json_object"}的模型，在 API 层面强制结构输出；二是做好幂等的解析兜底，用正则把第一个{到最后一个}之间的内容截出来再解析。

我现在的做法是写一个小的解析函数，如果直接解析失败，就尝试提取 JSON 片段：

import json import re def safe_parse_json(text: str) -> dict: text = text.strip() try: return json.loads(text) except json.JSONDecodeError: match = re.search(r"\{.*\}", text, re.S) if match: return json.loads(match.group(0)) raise

把safe_parse_json接到dispatch_node里之后，路由解析基本稳定了。这里想提醒所有新手：不要让提示词依赖模型自觉，要在代码层面做防御。

5.2 会无限循环的调度器：递归限制来得太迟

LangGraph 默认对递归深度有限制，超过限制会抛RecursionLimit。我一开始没加step_count硬保护，结果有一次模型反复给code_agent派任务，明明代码已经生成了，它还是说"还需要进一步验证"，于是code_agent执行完回到dispatch，又来一轮，直到递归上限被触发，整个调用链崩掉。

这个问题在日志里看起来非常奇怪：code_agent和dispatch两个节点互相跳，像死机了一样。我当时第一反应是检查图结构，怀疑边连错了，排查了半天才发现纯粹是调度器的决策没被约束。

解决方案就是在dispatch_node里加硬编码的轮次判断。我把它放在所有 LLM 调用之前，如果轮次超限，直接返回final。这个改动很小，效果却立竿见影。后来我又把阈值从 3 改成 4，因为有些合规流程确实需要"代码→文案→质检→修改"三个来回，给定更大余量后减少误杀。

5.3 ReAct Agent的中间步骤被截断，结果缺胳膊少腿

代码 Agent 用的是create_react_agent，它内部会执行多轮"思考-行动-观察"循环。有一次它生成了很长的代码，我拿到code_result后发现结尾是断的，明显是被截断了。排查下来，不是上下文窗口不够，而是max_iterations或者输出长度限制卡住了。

create_react_agent支持通过max_iterations控制思考轮数，默认值在有些版本里偏小。当模型需要先思考、再跑代码、再看结果、再修正的时候，轮数根本不够。我给这个 Agent 显式设置了max_iterations=5，并且把handle_parsing_errors=True打开，让解析异常不至于中断整个 Agent：

code_agent_runnable = create_react_agent( llm, [run_python_snippet], prompt=code_agent_prompt, max_iterations=5, handle_parsing_errors=True, )

还有一个容易被忽略的点：代码生成 Agent 返回内容可能包含 Markdown 代码块标记，比如python ...。传给文案 Agent 没问题，但如果你直接拿它去喂给exec，就会因为多余的标记字符而报错。我的建议是：在run_python_snippet工具内部先做一层处理，把代码块标记剥掉，清理后再执行。

6. 这个Demo还能怎么长进：从玩具到能用的调度助手

6.1 参数和提示词层面的性价比优化

如果想让这套调度助手的产出更稳定，第一步不是堆更多节点，而是调模型参数。我把调度器和质检员的温度都保持 0.1，文案 Agent 的温度设在 0.4 左右。原因前面说过：路由和审查需要确定性，文案生成需要一点变化性。

提示词方面也有优化空间。DISPATCH_PROMPT 里我是把code_result和doc_result的完整内容塞进去的，任务复杂后这会变得很长。你可以先用一个 LLM 调用把已有产出做"压缩摘要"，再交给调度器判断，能显著降低 token 消耗。这个摘要智能体本质上是多智能体架构里的"信息预处理节点"，加进去不需要改动整体图结构。

另外，质检节点现在只输出pass/fail和一句反馈，略显粗糙。可以升级为按"代码可运行性、文档匹配度、完整性"三个维度打分，并给出具体修复建议。模型在结构化打分任务上的表现通常比笼统的"判断是否通过"好不少，因为每个维度都有明确检查目标。

6.2 扩展方向：记忆、人工确认、多渠道输入

我目前这个 demo 是无状态的，每次调用都是独立任务，不会记住上一次运行的历史。如果想让助手具备对话记忆，StateGraph 编译时启用持久化机制，把状态存到后端，用户再次访问时能接着上一轮继续。这一步对任务调度的实际体验提升很明显，比如用户说"上次那个脚本再改成支持Excel"，调度器能结合历史上下文理解"上次那个"是谁。

更实际的扩展是引入人工确认节点。比如代码 Agent 生成了删除数据库数据的脚本，这种高风险任务不应该让审查 Agent 独自拍板，而是应该中断流程，等人类确认后再继续。LangGraph 提供了中断和恢复的机制，可以在关键节点挂起。这样整个调度系统就从"全自动"变成了"人在回路"，更适合真正的工作流。

还可以继续加新角色：资料搜索 Agent、表格数据分析 Agent、翻译 Agent，等等。每加一个执行者，只需要在调度器的动作列表里加一个枚举值，再加一个对应的执行节点和一条条件边，其余部分完全不用动。这种可扩展性正是把任务调度拆成多智能体架构带来的最大红利。

我个人最后想多提醒一句：多智能体系统确实能解决单模型"角色混乱"的问题，但它不是银弹。节点越多、回退次数越多，调用成本就越高，延迟也越长。做的时候不要为了"多智"而多智，先把用户需求拆出必要角色，能用两个 Agent 就别加第三个，保持图结构简单，才能一眼看出问题出在哪。