1. 从线性 todo.md 到 DAG多 Agent 协作的真实痛点manus 这类通用任务智能体很多人第一次拆它的工作流注意力都放在「意图识别 → 任务初始化 → 步骤规划 → 任务执行 → 归纳整理」这条主线上。这条线本身没问题但真正落地到代码里你会发现一个绕不开的坎todo.md 里的任务是线性依赖的。线性依赖意味着什么意味着第 3 步必须等第 2 步彻底跑完才能开始哪怕第 3 步只依赖第 1 步的产物。举个具体例子「日本旅行计划」这个任务被拆成查机票、查酒店、查景点、生成行程表、导出 PDF。其中「生成行程表」依赖机票、酒店、景点三个结果而「导出 PDF」只依赖行程表。线性执行会把查机票、查酒店、查景点串成一条链白白浪费并行窗口更糟的是如果查景点失败了整条链卡死前面已经成功的机票和酒店结果也白跑了。这就是为什么要把 todo.md 升级成 DAG有向无环图。DAG 的核心价值不是「看起来高级」而是三件事第一无依赖的节点可以并行调度缩短总耗时第二某个节点失败时只回溯它的下游不影响无关分支第三节点之间的输入输出关系被显式声明function call 的上下文注入变得可预测。我试过用纯 Python 字典手写依赖关系节点一多就乱成一团后来改成用邻接表 拓扑排序配合 Docker 容器做环境隔离整条链路才稳定下来。这篇就按这个思路把 Agent 节点配置、DAG 依赖声明、function call 注册、Docker 启动与流程验证完整走一遍。适合已经了解 multi-agent 基本概念、想自己动手跑通一条协作链路的开发者也适合正在评估 manus 类产品工程实现细节的技术同学。核心检索词就三个manus 工作流拆解、DAG 编排、function call 注册下面全部围绕它们展开。需要提前说明的是DAG 编排解决的是「调度层」问题function call 解决的是「能力层」问题Docker 解决的是「隔离层」问题。三层分开设计才能让每个 Agent 节点保持单一职责。很多同学一上来就把调度逻辑写进 Agent 内部结果 Agent 既要知道自己干什么还要知道别人干到哪了耦合度爆炸改一个节点牵动全身。下面我会把这三层拆开讲每层都给可复制的配置。2. TaoToken 前置为多 Agent 链路准备模型调用入口多 Agent 协作链路里模型调用是高频动作。意图识别要调一次小模型步骤规划要调一次推理模型search agent 提取网页信息要调多模态模型code agent 生成代码又要调一次。如果每个 Agent 各自维护一套 API Key 和 Base URL配置会散落在十几个文件里排查问题时根本找不到是哪个节点用了哪个模型。我的做法是统一走一个模型调用入口所有 Agent 节点通过环境变量读取同一组配置。这里用 TaoToken 作为统一入口它的 API 地址是 https://taotoken.net/api兼容 OpenAI 风格的接口格式Agent 节点里用 openai SDK 就能直接调不需要为每个模型单独写适配层。先说清楚为什么多 Agent 场景特别需要统一入口。第一模型切换成本低。意图识别用轻量模型步骤规划用推理模型多模态提取用视觉模型如果入口统一切换只是改一个 model 字段。第二Key 管理集中。Docker 容器里通过环境变量注入不用把 Key 写进镜像。第三计费可观测。多 Agent 的 token 消耗本来就高集中入口才能看清哪个节点最费钱。具体操作上你需要先拿到 API Key。访问 https://taotoken.net/api-keys 创建注意这个页面是 deep link创建后复制保存后面 Docker 启动时要注入。如果你还没决定用哪个模型可以先到 https://taotoken.net/models 看看当前可用的模型列表意图识别这类轻任务选便宜的小模型步骤规划选推理能力强的多模态提取选支持图片输入的。这里有个容易踩的坑很多同学把 Key 直接写进 docker-compose.yml 的 environment 字段然后提交到 Git。正确做法是写进 .env 文件docker-compose.yml 里用 ${TAOTOKEN_API_KEY} 引用.env 加进 .gitignore。下面第三节的配置片段会按这个方式来。另外提醒一点多 Agent 链路里每个节点调模型的频率不同建议在统一入口外面包一层薄薄的封装记录每次调用的节点名、模型名、token 消耗。这样跑完一条链路你能清楚看到 search agent 花了多少 tokencode agent 花了多少方便后续优化。这个封装不用复杂一个 Python 装饰器就够后面验证环节会给示例。如果你打算长期跑编码类或 Agent 类任务可以了解下 Coding Plan它更适合高频调用的场景。但本文的重点是工作流拆解模型入口只是前置准备拿到 Key、配好 Base URL、确认模型可用就可以进入下一步了。3. 可复制配置Agent 节点、DAG 依赖与 function call 注册这一节是全文的核心给三份可直接复制的配置Agent 节点定义、DAG 依赖声明、function call 注册表。三份配置放在同一个项目目录下Docker 启动时挂载进去。先看目录结构这样你复制时不会放错位置multi-agent-demo/ ├── docker-compose.yml ├── .env ├── config/ │ ├── agents.yaml │ ├── dag.yaml │ └── functions.json └── runtime/ └── tasks/3.1 Agent 节点配置 agents.yaml每个 Agent 节点声明自己的名称、职责、使用的模型、以及可调用的 function 列表。注意 model 字段的值要和你在模型列表里看到的一致。# config/agents.yaml agents: intent_agent: description: 意图识别与关键词提取 model: gpt-4o-mini base_url: ${TAOTOKEN_BASE_URL} api_key: ${TAOTOKEN_API_KEY} functions: [extract_keywords] planner_agent: description: 步骤规划输出 DAG 节点 model: deepseek-r1 base_url: ${TAOTOKEN_BASE_URL} api_key: ${TAOTOKEN_API_KEY} functions: [write_dag] search_agent: description: 网页搜索与多模态信息提取 model: claude-3.7-sonnet base_url: ${TAOTOKEN_BASE_URL} api_key: ${TAOTOKEN_API_KEY} functions: [web_search, extract_content] code_agent: description: 代码生成与执行 model: claude-3.7-sonnet base_url: ${TAOTOKEN_BASE_URL} api_key: ${TAOTOKEN_API_KEY} functions: [write_file, run_code] summary_agent: description: 结果归纳与输出整理 model: gpt-4o-mini base_url: ${TAOTOKEN_BASE_URL} api_key: ${TAOTOKEN_API_KEY} functions: [collect_artifacts]三件套在这里体现得很清楚Base URL 统一是 ${TAOTOKEN_BASE_URL}Key 统一是 ${TAOTOKEN_API_KEY}Model ID 每个节点按职责不同。这三个值都从环境变量读不硬编码。3.2 DAG 依赖声明 dag.yamlDAG 用邻接表形式声明每个节点列出它的依赖节点。调度器读取后做拓扑排序无依赖的节点并行执行。# config/dag.yaml dag: nodes: - id: intent agent: intent_agent depends_on: [] - id: plan agent: planner_agent depends_on: [intent] - id: search_flight agent: search_agent depends_on: [plan] - id: search_hotel agent: search_agent depends_on: [plan] - id: search_spot agent: search_agent depends_on: [plan] - id: build_itinerary agent: code_agent depends_on: [search_flight, search_hotel, search_spot] - id: export_pdf agent: code_agent depends_on: [build_itinerary] - id: summary agent: summary_agent depends_on: [export_pdf]这份声明里search_flight、search_hotel、search_spot 三个节点都只依赖 plan所以它们会被并行调度。build_itinerary 要等三个搜索节点全部完成。export_pdf 只依赖 build_itinerary。整条链路的最长路径是 intent → plan → search → build → export → summary但三个 search 并行后总耗时比线性执行短很多。3.3 function call 注册表 functions.jsonfunction call 的注册要遵循 OpenAI 的 tools 格式每个 function 声明名称、描述、参数 schema。Agent 节点调用模型时把对应节点的 functions 列表转成 tools 参数传进去。{ functions: [ { name: extract_keywords, description: 从用户输入中提取任务关键词和任务类型, parameters: { type: object, properties: { keywords: {type: array, items: {type: string}}, task_type: {type: string} }, required: [keywords, task_type] } }, { name: write_dag, description: 根据任务关键词生成 DAG 节点列表, parameters: { type: object, properties: { nodes: { type: array, items: { type: object, properties: { id: {type: string}, agent: {type: string}, depends_on: {type: array, items: {type: string}} } } } }, required: [nodes] } }, { name: web_search, description: 调用搜索接口获取网页结果列表, parameters: { type: object, properties: { query: {type: string}, top_k: {type: integer, default: 15} }, required: [query] } }, { name: extract_content, description: 从网页文本和截图中提取有效信息, parameters: { type: object, properties: { url: {type: string}, requirement: {type: string} }, required: [url, requirement] } }, { name: write_file, description: 在任务目录下写入文件, parameters: { type: object, properties: { path: {type: string}, content: {type: string} }, required: [path, content] } }, { name: run_code, description: 执行指定脚本并返回结果, parameters: { type: object, properties: { path: {type: string}, args: {type: array, items: {type: string}} }, required: [path] } }, { name: collect_artifacts, description: 收集任务目录下的所有产物文件, parameters: { type: object, properties: { task_dir: {type: string} }, required: [task_dir] } } ] }3.4 Docker 启动配置 docker-compose.ymlDocker 的作用是给每个任务一个隔离的运行环境。任务开始时创建容器任务结束时清理产物写入挂载的任务目录。# docker-compose.yml version: 3.9 services: agent-runtime: image: python:3.11-slim container_name: multi-agent-runtime working_dir: /app volumes: - ./config:/app/config:ro - ./runtime/tasks:/app/tasks environment: - TAOTOKEN_BASE_URL${TAOTOKEN_BASE_URL} - TAOTOKEN_API_KEY${TAOTOKEN_API_KEY} command: bash -c pip install openai pyyaml requests python /app/config/runner.py restart: no.env 文件内容TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEY你的Key注意 Base URL 用的是 https://taotoken.net/api不带任何多余路径。Key 从 .env 读.env 不进 Git。4. 验证请求跑通一条完整的多 Agent 协作链路配置写完接下来验证。验证分三步先确认模型调用通再确认 DAG 调度通最后确认整条链路产物完整。4.1 单节点模型调用验证先写一个最小脚本确认 Agent 节点能通过统一入口调到模型。这个脚本放在 config/runner.py 里先只测 intent_agent。# config/runner.py import os import yaml from openai import OpenAI def load_agents(): with open(/app/config/agents.yaml) as f: return yaml.safe_load(f)[agents] def test_intent_agent(): agents load_agents() cfg agents[intent_agent] client OpenAI( base_urlcfg[base_url], api_keycfg[api_key] ) resp client.chat.completions.create( modelcfg[model], messages[ {role: system, content: 你是意图识别助手提取关键词和任务类型。}, {role: user, content: 想去日本旅游需要一个旅行计划} ] ) print(resp.choices[0].message.content) if __name__ __main__: test_intent_agent()启动容器docker compose up --build如果配置正确你会看到模型返回类似{keywords: [japan-trip], task_type: travel}的内容。这一步通了说明 Base URL、Key、Model ID 三件套没问题。4.2 DAG 调度验证单节点通了之后加一个拓扑排序的调度器验证 DAG 能正确解析依赖顺序。# config/scheduler.py import yaml from collections import defaultdict, deque def load_dag(): with open(/app/config/dag.yaml) as f: return yaml.safe_load(f)[dag][nodes] def topo_sort(nodes): graph {n[id]: set(n[depends_on]) for n in nodes} indegree {nid: len(deps) for nid, deps in graph.items()} queue deque([nid for nid, d in indegree.items() if d 0]) order [] while queue: nid queue.popleft() order.append(nid) for other, deps in graph.items(): if nid in deps: indegree[other] - 1 if indegree[other] 0: queue.append(other) if len(order) ! len(nodes): raise ValueError(DAG 存在环无法拓扑排序) return order if __name__ __main__: nodes load_dag() print(调度顺序:, topo_sort(nodes))预期输出调度顺序: [intent, plan, search_flight, search_hotel, search_spot, build_itinerary, export_pdf, summary]注意 search 三个节点在 plan 之后连续出现说明它们处于同一层可以并行。如果你的输出里 search 节点被串行排开检查 depends_on 是否写错。4.3 完整链路验证把调度器和 Agent 执行串起来每个节点执行时把产物写入 /app/tasks/{task_id}/ 目录。跑完后检查目录docker compose exec agent-runtime ls -R /app/tasks预期看到/app/tasks/japan-trip/ ├── intent.json ├── dag.yaml ├── search_flight.json ├── search_hotel.json ├── search_spot.json ├── itinerary.html ├── itinerary.pdf └── summary.md每个文件对应一个节点的产物。如果某个文件缺失说明对应节点执行失败看容器日志定位。4.4 并行执行验证为了确认 search 三个节点真的并行可以在每个 search 节点执行时打印时间戳import time print(fsearch_flight start: {time.time()})如果三个节点的 start 时间戳接近说明并行生效。如果相差几秒说明调度器还是串行执行检查是否用了线程池或异步。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth多 Agent 链路跑起来后报错集中在几个地方。这一节按真实报错对照排查。5.1 401 Unauthorized最常见。原因通常是 Key 没注入到容器或者 .env 没被 docker-compose 读取。排查步骤docker compose exec agent-runtime env | grep TAOTOKEN如果输出为空说明环境变量没传进去。检查 docker-compose.yml 的 environment 字段是否引用了 ${TAOTOKEN_API_KEY}以及 .env 文件是否和 docker-compose.yml 在同一目录。还有一种情况是 Key 复制时带了空格或换行。检查 .env 文件cat -A .env如果行尾有$之外的字符说明有隐藏字符重新复制。5.2 local proxy failed这个报错通常出现在容器内网络请求时。原因可能是容器网络配置问题或者 Base URL 写成了 localhost。注意容器内的 localhost 指向容器自己不是宿主机。Base URL 必须是完整的外部地址 https://taotoken.net/api。排查docker compose exec agent-runtime curl -I https://taotoken.net/api如果 curl 不通检查容器 DNS 配置。如果 curl 通但 Python 请求失败检查是否设置了 HTTP_PROXY 之类的环境变量容器内不需要额外代理配置。5.3 reading choices 报错这个报错通常是模型返回格式和代码解析不匹配。比如你期望返回 JSON但模型返回了带 markdown 代码块的文本。排查resp client.chat.completions.create(...) print(resp.choices[0].message.content)先打印原始返回看格式。如果是 markdown 包裹的 JSON加一层清洗import json, re content resp.choices[0].message.content content re.sub(r^json\s*|\s*$, , content.strip()) data json.loads(content)另一个原因是 model 字段写错模型不存在时返回结构异常。对照模型列表确认 Model ID 拼写。5.4 OAuth 相关报错如果你用的是需要 OAuth 的客户端工具报错可能和 token 过期有关。这类工具通常有自己的配置文件比如 Claude Code 的 settings、Cline 的 MCP 配置、Codex 的 auth.json。以 Codex 的 auth.json 为例三件套要写全{ base_url: https://taotoken.net/api, api_key: 你的Key, model: claude-3.7-sonnet }Base URL、Key、Model ID 缺一不可。如果只填了 Key 没填 Base URL会走默认地址导致 401。如果 Model ID 写错会报模型不存在。5.5 DAG 环检测报错如果拓扑排序抛出「DAG 存在环」说明 depends_on 写成了循环依赖。比如 A 依赖 BB 又依赖 A。排查方法把 dag.yaml 的节点画成图检查是否有回边。常见错误是复制节点时忘了改 depends_on。5.6 容器内文件权限问题产物写入 /app/tasks 时如果报 Permission denied检查挂载目录的宿主机权限ls -ld runtime/tasks chmod 777 runtime/tasks生产环境不要用 777这里只是本地验证方便。6. 继续深入从可跑通到可观测链路跑通只是第一步。多 Agent 协作真正难的地方在于可观测性和成本控制。我在实际跑的时候加了一层轻量的调用日志每个 Agent 节点调模型前后各记一条包含节点名、模型名、输入 token、输出 token、耗时。跑完一条链路能清楚看到哪个节点最费钱、哪个节点最慢。具体做法是在 OpenAI client 外面包一层import time, logging def traced_call(node_name, client, **kwargs): start time.time() resp client.chat.completions.create(**kwargs) elapsed time.time() - start usage resp.usage logging.info( fnode{node_name} model{kwargs[model]} fin{usage.prompt_tokens} out{usage.completion_tokens} felapsed{elapsed:.2f}s ) return resp这样每个节点的消耗一目了然。多 Agent 链路的 token 消耗本来就高没有这层日志你根本不知道钱花在哪了。另一个值得做的改进是失败回溯。DAG 的好处是某个节点失败时只重跑它的下游。实现上调度器记录每个节点的状态失败时把下游节点标记为 pending重新入队。这样 search_hotel 失败了不用重跑 search_flight 和 search_spot。如果你想让链路更稳可以在关键节点后面加一个校验 Agent对产物打分低于阈值就回溯到上游节点重跑。这个校验 Agent 本身也是一次 function call注册方式和前面一样。最后说下模型选择。意图识别这类轻任务用便宜的小模型就够步骤规划用推理模型多模态提取用视觉模型。统一入口的好处就是切换只改一个字段。如果你打算长期跑这类链路可以看看 Coding Plan高频调用场景下更划算。需要创建 Key 的话到 API Keys 页面接入细节看接入文档想先试模型效果可以去模型对话页面直接聊几句。