1. 为什么我要用 langgraph 手搓一个简略版 Manus
Manus 这类通用 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_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，创建完在 API Keys 页面复制，页面是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。
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(path="config.toml"): with open(path, "rb") as f: return tomllib.load(f) cfg = load_config() llm = ChatOpenAI( base_url=cfg["llm"]["base_url"], api_key=cfg["llm"]["api_key"], model=cfg["llm"]["model"], temperature=cfg["llm"]["temperature"], )这里有个细节：base_url 结尾不要带/v1，TaoToken 的兼容层已经处理了路径。如果你用的是其他 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_ok=True) with open(path, "w", encoding="utf-8") as f: f.write(file_contents) return {"message": f"created {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", encoding="utf-8") as f: content = f.read() content = content.replace(old_str, new_str, 1) with open(path, "w", encoding="utf-8") as f: f.write(content) return {"message": f"replaced 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, shell=True, cwd=os.getcwd(), capture_output=True, text=True, check=False) 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:end+1] if start != -1 else "{}" def create_planner_node(state: State): msgs = [ SystemMessage(content=PLAN_SYSTEM), HumanMessage(content=PLAN_CREATE.format(user_message=state["user_message"])), ] resp = llm.invoke(msgs) plan = json.loads(extract_json(resp.content)) return Command(goto="execute", update={ "plan": plan, "messages": [AIMessage(content=json.dumps(plan, ensure_ascii=False))], })执行节点是核心，它找到第一个 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(goto="report") msgs = state.get("observations", []) + [ SystemMessage(content=EXEC_SYSTEM), HumanMessage(content=EXEC_PROMPT.format( user_message=state["user_message"], step=current["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(content=str(result), tool_call_id=tc["id"]) msgs.append(tm) new_msgs.append(tm) return Command(goto="update_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(content=PLAN_SYSTEM), HumanMessage(content=UPDATE_PROMPT.format(plan=plan, goal=plan["goal"])), ] for _ in range(3): try: resp = llm.invoke(msgs) new_plan = json.loads(extract_json(resp.content)) return Command(goto="execute", update={"plan": new_plan}) except Exception as e: msgs.append(HumanMessage(content=f"JSON 格式错误：{e}，请重新输出")) return Command(goto="execute", update={"plan": plan})报告节点汇总所有观察记录，生成最终文件。
REPORT_SYSTEM = """你是报告生成专家，根据已有上下文生成分析报告。 报告包含分析背景、数据概述、可视化、结论建议。 图表插入分析过程，不单独列出。以文件形式输出。""" def report_node(state: State): msgs = state.get("observations", []) + [SystemMessage(content=REPORT_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(content=str(result), tool_call_id=tc["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(checkpointer=MemorySaver()) graph = build_graph()4. 验证请求与成功结果
跑之前先确认工作目录存在，把要分析的文档放进去。下面用一个 docx 做示例。
import uuid config = {"configurable": {"thread_id": str(uuid.uuid4())}} result = graph.invoke( {"user_message": "分析当前目录下的 计算机视觉.docx，生成一份简单分析报告"}, config=config, ) print(result["final_report"])正常跑通时，你会看到控制台依次打印规划、执行、更新、报告四个阶段。执行阶段会调用 shell_exec 跑 Python 读取 docx，调用 create_file 保存中间结果。最终在./agent/files下生成报告文件。
如果只想快速验证模型连通性，可以先用模型对话页面发一条消息，地址是 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite ，确认 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 是否误加了/v1，TaoToken 的地址是 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_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ，它更适合高频调用的场景。接入细节和参数说明在文档里，地址是 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，遇到 SDK 兼容问题可以先翻这里。
最后提醒一句：Agent 的能力上限取决于你给它的工具边界。工具越明确、返回值越结构化，模型越不容易跑偏。先把这篇的四个节点和三个工具跑稳，再往上叠功能，比一上来就堆一堆工具要靠谱得多。
