1. 为什么我要用 langgraph 手搓一个简略版 ManusManus 这类通用 Agent 最吸引人的地方是它能自己拆任务、自己调工具、自己写文件、自己出报告。但真到自己动手时很多人会卡在第一步状态怎么在节点之间传、工具怎么绑定、循环什么时候退出。langgraph 正好解决这个问题它把 Agent 拆成一张状态图每个节点只负责一件事边和条件跳转由框架管。这篇要落地的场景很具体给一个文档让 Agent 自动生成分析计划、逐步执行、动态更新计划、最后输出一份报告。适合已经写过简单 LLM 调用、想进一步理解 Agent 编排的读者。我会给出可复制的节点骨架、状态定义、TaoToken 统一 Key 的 config.toml 示例以及跑通验证的完整步骤。整套代码不依赖复杂框架核心就是 langgraph langchain 几个自定义 tool。先明确一点这不是要复刻 Manus 的全部能力而是抓住它最核心的四个动作——规划、执行、更新、报告。把这四个节点跑通你就有了一个最小可用的 Agent 骨架后面加工具、换模型、接业务都是在这个骨架上长出来的。2. TaoToken 前置统一 Key 与 config.toml 配置在写节点之前先把模型接入这层处理干净。Agent 会频繁调用大模型如果 Key 散落在代码里后面换模型、调参数会很痛苦。我习惯用一个 config.toml 统一管理配合 TaoToken 的兼容接口OpenAI SDK 和 langchain 都能直接读。TaoToken 的 API 地址是 https://taotoken.net/api它兼容 OpenAI 的请求格式所以 langchain 里的 ChatOpenAI 只要改 base_url 和 api_key 就能用。先去控制台创建一个 Key地址在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完在 API Keys 页面复制页面是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。config.toml 这样写[llm] base_url https://taotoken.net/api api_key sk-你的Key model gpt-4o-mini temperature 0.2 max_tokens 4096 [agent] work_dir ./agent/files font_path ./agent/simsun.ttf max_step_retry 3读取配置用一个简单函数避免硬编码import tomllib from langchain_openai import ChatOpenAI def load_config(pathconfig.toml): with open(path, rb) as f: return tomllib.load(f) cfg load_config() llm ChatOpenAI( base_urlcfg[llm][base_url], api_keycfg[llm][api_key], modelcfg[llm][model], temperaturecfg[llm][temperature], )这里有个细节base_url 结尾不要带/v1TaoToken 的兼容层已经处理了路径。如果你用的是其他 SDK把 base_url 填成https://taotoken.net/api即可。模型名按你实际开通的填temperature 建议 0.2 左右Agent 任务需要稳定输出太发散会导致 JSON 解析失败。注意config.toml 不要提交到公开仓库本地用 .gitignore 排除或者改用环境变量注入。3. 可复制的 langgraph 节点与状态图骨架3.1 定义 State 和工具State 是整个图的共享内存节点通过读写它来传递信息。这里用 TypedDict 定义包含用户消息、计划、观察记录、最终报告。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): user_message: str plan: dict messages: Annotated[list, add_messages] observations: Annotated[list, add_messages] final_report: str工具层定义三个基础能力创建文件、替换文本、执行 shell。这三个就够跑通文档分析。from langchain_core.tools import tool import os, subprocess tool def create_file(file_name: str, file_contents: str) - dict: 在工作区创建文件并写入内容。 try: path os.path.join(os.getcwd(), file_name) os.makedirs(os.path.dirname(path), exist_okTrue) with open(path, w, encodingutf-8) as f: f.write(file_contents) return {message: fcreated {path}} except Exception as e: return {error: str(e)} tool def str_replace(file_name: str, old_str: str, new_str: str) - dict: 替换文件中第一处匹配文本。 try: path os.path.join(os.getcwd(), file_name) with open(path, r, encodingutf-8) as f: content f.read() content content.replace(old_str, new_str, 1) with open(path, w, encodingutf-8) as f: f.write(content) return {message: freplaced in {path}} except Exception as e: return {error: str(e)} tool def shell_exec(command: str) - dict: 执行 shell 命令并返回 stdout/stderr。 try: r subprocess.run(command, shellTrue, cwdos.getcwd(), capture_outputTrue, textTrue, checkFalse) return {stdout: r.stdout, stderr: r.stderr} except Exception as e: return {error: str(e)}3.2 四个核心节点规划节点负责把用户输入变成结构化计划。关键是让模型输出严格 JSON解析失败要有兜底。import json from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langgraph.types import Command PLAN_SYSTEM 你是具备自主规划能力的智能体。默认工作语言中文。 输出必须是严格 JSON字段thought(str)、goal(str)、steps(list)。 每个 step 含 title、description、status(pending/completed)。 任务不可行时 steps 返回空数组。 PLAN_CREATE 根据用户消息生成计划。 用户消息{user_message} 只输出 JSON不要其他内容。 def extract_json(text: str) - str: start text.find({) end text.rfind(}) return text[start:end1] if start ! -1 else {} def create_planner_node(state: State): msgs [ SystemMessage(contentPLAN_SYSTEM), HumanMessage(contentPLAN_CREATE.format(user_messagestate[user_message])), ] resp llm.invoke(msgs) plan json.loads(extract_json(resp.content)) return Command(gotoexecute, update{ plan: plan, messages: [AIMessage(contentjson.dumps(plan, ensure_asciiFalse))], })执行节点是核心它找到第一个 pending 步骤绑定工具循环调用直到模型不再请求工具。from langchain_core.messages import ToolMessage EXEC_SYSTEM 你是具备自主能力的 AI 智能体擅长数据处理、分析与可视化。 每次只选择一个工具调用工具失败要换参数重试直到任务完成。 文件读写用文件工具代码先保存为文件再执行。 EXEC_PROMPT 根据用户消息和当前步骤选择最合适的工具。 用户消息{user_message} 当前步骤{step} def execute_node(state: State): plan state[plan] steps plan[steps] current None idx 0 for i, s in enumerate(steps): if s[status] pending: current, idx s, i break if current is None: return Command(gotoreport) msgs state.get(observations, []) [ SystemMessage(contentEXEC_SYSTEM), HumanMessage(contentEXEC_PROMPT.format( user_messagestate[user_message], stepcurrent[description])), ] llm_tools llm.bind_tools([create_file, str_replace, shell_exec]) new_msgs [] tool_map {create_file: create_file, str_replace: str_replace, shell_exec: shell_exec} while True: resp llm_tools.invoke(msgs) msgs.append(resp) new_msgs.append(resp) if not resp.tool_calls: break for tc in resp.tool_calls: result tool_map[tc[name]].invoke(tc[args]) tm ToolMessage(contentstr(result), tool_call_idtc[id]) msgs.append(tm) new_msgs.append(tm) return Command(gotoupdate_planner, update{ plan: plan, messages: new_msgs, observations: new_msgs, })更新节点根据执行结果调整剩余步骤注意它只改未完成的部分。UPDATE_PROMPT 根据上下文更新计划不要更改 goal。 只重新规划未完成步骤已完成步骤保持不变。 输出与输入格式一致的 JSON。 Plan: {plan} Goal: {goal} def update_planner_node(state: State): plan state[plan] msgs state[messages] [ SystemMessage(contentPLAN_SYSTEM), HumanMessage(contentUPDATE_PROMPT.format(planplan, goalplan[goal])), ] for _ in range(3): try: resp llm.invoke(msgs) new_plan json.loads(extract_json(resp.content)) return Command(gotoexecute, update{plan: new_plan}) except Exception as e: msgs.append(HumanMessage(contentfJSON 格式错误{e}请重新输出)) return Command(gotoexecute, update{plan: plan})报告节点汇总所有观察记录生成最终文件。REPORT_SYSTEM 你是报告生成专家根据已有上下文生成分析报告。 报告包含分析背景、数据概述、可视化、结论建议。 图表插入分析过程不单独列出。以文件形式输出。 def report_node(state: State): msgs state.get(observations, []) [SystemMessage(contentREPORT_SYSTEM)] llm_tools llm.bind_tools([create_file, shell_exec]) tool_map {create_file: create_file, shell_exec: shell_exec} while True: resp llm_tools.invoke(msgs) msgs.append(resp) if not resp.tool_calls: break for tc in resp.tool_calls: result tool_map[tc[name]].invoke(tc[args]) msgs.append(ToolMessage(contentstr(result), tool_call_idtc[id])) return {final_report: msgs[-1].content}3.3 组装状态图因为节点用 Command 直接指定跳转中间不需要显式加边但 START 和 END 必须连。from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver def build_graph(): builder StateGraph(State) builder.add_node(create_planner, create_planner_node) builder.add_node(execute, execute_node) builder.add_node(update_planner, update_planner_node) builder.add_node(report, report_node) builder.add_edge(START, create_planner) builder.add_edge(report, END) return builder.compile(checkpointerMemorySaver()) graph build_graph()4. 验证请求与成功结果跑之前先确认工作目录存在把要分析的文档放进去。下面用一个 docx 做示例。import uuid config {configurable: {thread_id: str(uuid.uuid4())}} result graph.invoke( {user_message: 分析当前目录下的 计算机视觉.docx生成一份简单分析报告}, configconfig, ) print(result[final_report])正常跑通时你会看到控制台依次打印规划、执行、更新、报告四个阶段。执行阶段会调用 shell_exec 跑 Python 读取 docx调用 create_file 保存中间结果。最终在./agent/files下生成报告文件。如果只想快速验证模型连通性可以先用模型对话页面发一条消息地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 确认 Key 和模型名没问题再跑完整图。验证成功的标志有三个一是 create_planner 返回的 plan 里 steps 非空二是 execute 阶段日志里出现 tool_name 和 tool_result三是 report 节点结束后 final_report 有内容且文件真实存在。5. 本篇常见错误排查JSON 解析失败模型输出带了 markdown 代码块或多余说明。解决方法是 extract_json 里做首尾大括号截取同时在 prompt 里强调“只输出 JSON”。如果还不行把 temperature 降到 0。工具调用死循环模型反复调用同一个工具但任务没进展。给 while 循环加最大轮次限制比如 10 次超过就强制退出并记录。生产环境一定要加这个保护。文件路径找不到create_file 里用了相对路径但 cwd 不是预期目录。统一用 os.getcwd() 拼接并在启动时打印一次工作目录确认。模型名或 base_url 报 404检查 config.toml 里 base_url 是否误加了/v1TaoToken 的地址是 https://taotoken.net/api 不要多写路径。模型名要和开通的一致。Command 跳转不生效确认节点返回的是 Command 对象而不是普通 dict且 goto 的目标节点名和 add_node 注册的名字完全一致。报告节点没有输出文件模型可能只生成了文本没调工具。在 REPORT_SYSTEM 里明确要求“必须调用 create_file 保存报告”并在循环结束后检查文件是否存在。6. 继续扩展与接入建议这套骨架跑通后扩展方向很清晰。一是加工具比如接数据库查询、接图表生成、接搜索二是把单一模型拆成多模型规划用强模型、执行用快模型能明显降延迟三是把 checkpointer 从内存换成持久化存储支持断点续跑。如果你打算长期跑编码类或 Agent 类任务可以了解下 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它更适合高频调用的场景。接入细节和参数说明在文档里地址是 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到 SDK 兼容问题可以先翻这里。最后提醒一句Agent 的能力上限取决于你给它的工具边界。工具越明确、返回值越结构化模型越不容易跑偏。先把这篇的四个节点和三个工具跑稳再往上叠功能比一上来就堆一堆工具要靠谱得多。