企业官网建设流程全解析

1. 100 行编排引擎，卡在「模拟返回」这一层

AgenticOS 那套思路把多智能体编排拆成了 Orchestrator 任务拆解和 research→write→review 的 DAG 依赖调度，但落到代码里，Agent.think 只是打印一行日志再 sleep。要让这套 100 行引擎真正产出内容，我选择走 TaoToken：打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content= 创建 API Key，拿到统一 Base URL https://taotoken.net/api，交给 Codex 去改代码并跑通验证。

多智能体引擎的最小模型其实不复杂。Orchestrator 把“撰写一篇技术文章”拆成调研、写作、审核三个子任务，research 不依赖谁，write 依赖 research，review 依赖 write，调度器按依赖顺序往下走。这个 DAG 逻辑本身是通的，我在 main() 里注册三个 Agent 后，任务状态能稳定返回 completed。问题不在调度，而在每个 Agent 的 think 方法：它没有调用任何模型，只是等待 0.5 秒，然后塞回一句固定模板。这导致 research 输出的“调研结果”和 review 输出的“审核意见”内容几乎一样，多智能体协同变成了三个节点轮流传纸条。

所以这次验证只改一处：把 think 的实现从模拟返回换成对 TaoToken 的真实模型请求，让 Orchestrator 里那套 DAG 依赖判断和 Agent 注册逻辑原样保留。这样既能确认编排引擎可以驱动真实模型，又能把 100 行脚手架真正用起来。

1.1 一个只 sleep 的 Agent，相当于没接模型

原文代码里的 Agent 类，核心逻辑就三件事：打印任务名称、等待半秒、返回“已完成”。用作架构示例可以，但把它放进任何真实场景都会露馅。如果 tasks 换成“撰写一篇关于多智能体操作系统的技术文章”，三个 Agent 最后返回的内容除了前缀不同，正文全是同一句模板。更麻烦的是，Orchestrator 里已经把 context 作为上下文在传了，但 think 函数根本没用它。

async def think(self, task: str, context: dict) -> str: print(f"[{self.role.value}] {self.name} 接收任务: {task[:50]}...") await asyncio.sleep(0.5) return f"[{self.name}] 已完成: {task}"

这一小段就是整篇文章最需要替换的位置。只要把它改成真实模型调用，context 里累积的“已有产出”才能真正参与到后续 Agent 的决策中，DAG 调度也才有意义。

1.2 research→write→review 的 DAG 调度本身是通的

先确认编排引擎里哪些代码值得保留。execute_workflow 每次处理一个子任务时，会先检查 deps 里的每一项是否都已经出现在 context["artifacts"] 里，只有全部满足才执行当前 Agent，并把结果写回 artifacts。这个机制保证了 research 没跑完之前，write 不会被触发，write 没跑完之前，review 不会开始。

missing_deps = [dep for dep in subtask["deps"] if dep not in context["artifacts"]] if missing_deps: print(f"[!] 子任务 {subtask['id']} 依赖 {missing_deps} 未就绪，跳过") continue

这段逻辑没有任何问题，后续改造时我特意不动它。真正要换的只有 Agent.think 的实现方式。

2. 从官网拿 Key，再把 Codex 指到 TaoToken

2.1 创建 API Key，模型 ID 以模型广场为准

先打开 TaoToken 注册并登录。进入控制台创建一个 API Key，这一步拿到的字符串就是后面的 YOUR_API_KEY。模型名称不要凭记忆填，TaoToken 模型广场会列出当前可用的模型 ID，直接复制一个作为 YOUR_MODEL_ID。

注意：落地页和接口是两个地址。注册、建 Key、看模型广场和用量去 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content= ；填进 Codex 和 Python 的 Base URL 只写 https://taotoken.net/api，末尾不补 /v1。

这条链路里，统一接入通道的角色就是 TaoToken。它不改变 Codex 或 Python 的调用方式，只把模型请求集中到一个标准的 OpenAI 兼容入口。对 Codex 来说，它只需要知道 model_provider 的 base_url 和 env_key；对后面的 Python 代码来说，它只需要知道 OpenAI SDK 的 base_url。两处都指向同一个 https://taotoken.net/api，Key 也共用。

2.2 ~/.codex/config.toml 只改 model_provider 和 base_url

Codex 的配置文件在 ~/.codex/config.toml，默认会读官方 provider。这里直接加一个 taotoken provider，并把默认模型指过去：

model = "YOUR_MODEL_ID" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"

保存后把 Key 放进环境变量。bash 里执行：

export TAOTOKEN_API_KEY=YOUR_API_KEY export TAOTOKEN_MODEL=YOUR_MODEL_ID

这样 Codex 每次启动都会从 TAOTOKEN_API_KEY 读凭据，从 TAOTOKEN_MODEL 读模型。不要在这里写死某个具体模型名，因为模型 ID 随时可能更新，以模型广场当前列表为准。后面 Python 代码里也用同一个环境变量，这样换模型时只需要改 export，不用动两个文件。

2.3 先用 taotoken CLI 或 codex 命令验证连通性

不想立刻改代码的话，可以先跑一条命令验证模型 ID 是否可用。如果电脑上有 Node.js 环境，可以用 TaoToken 的 CLI 做一次极简对话：

npm install -g @taotoken/taotoken taotoken cc -k YOUR_API_KEY -u https://taotoken.net/api -m YOUR_MODEL_ID

这里 Base URL 原样填的是 https://taotoken.net/api，没有把官网落地页地址塞进工具。这条命令能返回模型应答，就说明 Key、模型 ID、接口路径三者都正确。验证通过后，再打开 Codex 跑codex exec "1+1"，如果返回 2，说明 config.toml 也接通了。

3. 用 Codex 把 Agent.think 换成真实模型调用

3.1 给 Codex 的任务书（包含禁止加 /v1）

在项目目录执行codex exec，把下面这段作为任务描述：

请修改 agents.py：把 Agent.think 方法里的模拟返回改成真实模型调用。 要求： 1. 使用 openai 库的 AsyncOpenAI，base_url 固定为 https://taotoken.net/api，末尾不要加 /v1。 2. api_key 从环境变量 TAOTOKEN_API_KEY 读取，模型 ID 从 TAOTOKEN_MODEL 读取，不要写死模型名。 3. 保持 async def think(self, task, context) 签名不变，返回值必须是 str。 4. 调用成功后返回 completion.choices[0].message.content.strip()。

Codex 会先读文件，再重写 think，最后检查语法。如果它把 base_url 写成 https://taotoken.net/api/v1，你需要在 review 阶段明确驳回，因为 TaoToken 的兼容端点是 /api，不是 /api/v1。这是最容易改错的一处，比模型名错误更隐蔽。

3.2 改造后的 Agent 类和 keep 不变的 Orchestrator

改造后核心差异在于：Agent 持有 AsyncOpenAI 客户端，think 把任务描述、已有产出、角色身份一起拼进 prompt，再返回真实模型输出。下面是一份可直接替换的完整 agents.py，把 Orchestrator 的依赖检查、main() 入口都包含进去：

import os import json import asyncio from dataclasses import dataclass, field from enum import Enum from openai import AsyncOpenAI class AgentRole(Enum): ORCHESTRATOR = "orchestrator" RESEARCHER = "researcher" WRITER = "writer" REVIEWER = "reviewer" PUBLISHER = "publisher" @dataclass class Agent: name: str role: AgentRole skills: list[str] = field(default_factory=list) max_retries: int = 3 client: AsyncOpenAI = field(init=False) def __post_init__(self): self.client = AsyncOpenAI( base_url="https://taotoken.net/api", api_key=os.getenv("TAOTOKEN_API_KEY", "YOUR_API_KEY"), ) async def run(self, task: str, context: dict) -> dict: for attempt in range(self.max_retries): try: result = await self.think(task, context) return {"agent": self.name, "status": "success", "result": result} except Exception as exc: if attempt == self.max_retries - 1: return {"agent": self.name, "status": "failed", "error": str(exc)} await asyncio.sleep(0.5) async def think(self, task: str, context: dict) -> str: print(f"[{self.role.value}] {self.name} 接收任务: {task[:50]}...") completion = await self.client.chat.completions.create( model=os.getenv("TAOTOKEN_MODEL", "YOUR_MODEL_ID"), messages=[ {"role": "system", "content": f"你是多智能体编排中的{self.name}，角色是{self.role.value}。"}, {"role": "user", "content": f"上级任务：{context.get('user_task', '')}\n已有产出：{json.dumps(context.get('artifacts', {}), ensure_ascii=False)}\n本次任务：{task}\n请直接输出最终结果。"}, ], temperature=0.3, ) return completion.choices[0].message.content.strip() @dataclass class Orchestrator: name: str agents: dict = field(default_factory=dict) def register_agent(self, agent: Agent): self.agents[agent.role.value] = agent def decompose_task(self, task: str): return [ {"id": "research", "description": f"调研任务：{task}", "assignee": "researcher", "deps": []}, {"id": "write", "description": f"撰写任务：{task}", "assignee": "writer", "deps": ["research"]}, {"id": "review", "description": f"审核任务：{task}", "assignee": "reviewer", "deps": ["write"]}, ] async def execute_workflow(self, user_task: str) -> dict: print(f"\n[Orchestrator] 接收用户任务: {user_task}") subtasks = self.decompose_task(user_task) context = {"user_task": user_task, "artifacts": {}} for subtask in subtasks: missing_deps = [dep for dep in subtask["deps"] if dep not in context["artifacts"]] if missing_deps: print(f"[!] 子任务 {subtask['id']} 依赖 {missing_deps} 未就绪，跳过") continue agent = self.agents.get(subtask["assignee"]) if agent is None: print(f"[!] 找不到 Agent: {subtask['assignee']}") continue result = await agent.run(subtask["description"], context) context["artifacts"][subtask["id"]] = result print(f"[✓] {subtask['id']} 完成") return {"status": "completed", "artifacts": context["artifacts"]} async def main(): orchestrator = Orchestrator(name="主编排器") orchestrator.register_agent(Agent(name="调研助手", role=AgentRole.RESEARCHER, skills=["web_search", "data_analysis"])) orchestrator.register_agent(Agent(name="写作助手", role=AgentRole.WRITER, skills=["content_generation", "markdown"])) orchestrator.register_agent(Agent(name="审核专员", role=AgentRole.REVIEWER, skills=["quality_check", "fact_verify"])) result = await orchestrator.execute_workflow("撰写一篇关于多智能体操作系统的技术文章") print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": asyncio.run(main())

注意几个细节。第一，client 在__post_init__里初始化，避免每次 think 都重新建立连接。第二，上下文里把 context 中的已有产出用 json.dumps 传给模型，write 阶段能看到 research 的真实内容，review 阶段能看到 write 的真实内容，DAG 的依赖才有价值。第三，retry 逻辑保留，模型偶发网络错误时不会让整个编排直接崩掉。

4. 运行 main()，确认编排状态为 completed

4.1 预期输出：research、write、review 依次完成

设置好环境变量后运行python main.py。如果一切正常，控制台会按这个顺序输出：Orchestrator 拆任务，research 完成，write 完成，review 完成，最后 JSON 里 status 为 completed。

[Orchestrator] 接收用户任务: 撰写一篇关于多智能体操作系统的技术文章 [Orchestrator] 开始任务拆解... [researcher] 调研助手 接收任务: 调研任务: 撰写... [✓] research 完成 [writer] 写作助手 接收任务: 撰写任务: 撰写... [✓] write 完成 [reviewer] 审核专员 接收任务: 审核任务: 撰写... [✓] review 完成 === 最终执行结果 === { "status": "completed", "artifacts": { "research": {"agent": "调研助手", "status": "success", "result": "基于 MCP 和 A2A 的调研摘要..."}, "write": {"agent": "写作助手", "status": "success", "result": "多智能体操作系统架构初稿..."}, "review": {"agent": "审核专员", "status": "success", "result": "核对事实并补充参考文献..."} } }

这次输出的 artifacts 不再是固定字符串。research 的结果是模型生成的调研摘要，write 的结果是在该摘要基础上展开的初稿，review 的结果是针对初稿的核对意见。三份内容有接续关系，说明 context 传递和依赖调度确实生效了。status 保持 completed，说明原编排引擎的核心逻辑没被破坏。

4.2 去控制台对一下这次真实调用

验证“真的接了模型”而不是本地模拟，有一个最简单的办法：回到 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content= 的控制台看用量。刚才这次运行应该产生了三条请求记录，分别对应当前代码里 research、write、review 三次 chat completions 调用。如果只看到一条或零条，说明部分 Agent 的 think 仍然没走模型，或者依赖检查把子任务跳过了。

这一步也把 Key 从“创建成功”推进到“真实消费”。多智能体编排跑通后，你真正需要盯的指标不是代码状态，而是模型请求次数和 token 用量。编排越复杂，Agent 数量越多，这个数字增长越快，提前在控制台确认路径，比最后排查账单要省事得多。

5. 实测最容易翻车的三个细节：Base URL、artifacts、模型 ID

5.1 Base URL 末尾别补 /v1

OpenAI SDK 会自动在 base_url 后面拼 /chat/completions。如果 base_url 写成 https://taotoken.net/api/v1，实际请求会变成 https://taotoken.net/api/v1/chat/completions，返回 404。Codex 的 config.toml 和 agents.py 里的 AsyncOpenAI 都只填 https://taotoken.net/api。这个错误很隐蔽，因为登录官网不受影响，只有发请求时才报错。看到 404 先检查这里，而不是去怀疑 Key 失效。

5.2 artifacts 的 key 与 deps 要保持一致

如果某个子任务把结果写进 context["artifacts"]["research_result"]，而后续 write 的 deps 里写的是 research，missing_deps 会一直包含 research，控制台打印“依赖未就绪，跳过”，write 和 review 都不会执行。上面示例里把子任务 id、artifacts 的 key、deps 里的引用统一成 research / write / review。自定义新任务时，保持这三个值一致即可。

5.3 模型 ID 不要写死，从模型广场复制

代码里用os.getenv("TAOTOKEN_MODEL", "YOUR_MODEL_ID")，而不是把某个具体模型名写死在 messages 或 config 里。模型 ID 以模型广场当前列表为准，写死一个不存在的 ID，编排跑到第一个 Agent 就报 404。改成环境变量后，换模型只需改 export，不用动 Python 代码，也不用让 Codex 再改一遍。

6. 跑通之后，往关键路径调度和 MCP 工具扩展

6.1 并行子任务：image_gen 与 write 同时跑

当前 execute_workflow 是顺序 for 循环。实际上 write 只依赖 research，如果再加一个 image_gen，它同样只依赖 research，完全可以和 write 并行。用 asyncio.gather 把同一层子任务一起调度，就能复现原文提到的关键路径调度思路：

tasks = [] for subtask in parallel_group: agent = self.agents[subtask["assignee"]] tasks.append(agent.run(subtask["description"], context)) parallel_results = await asyncio.gather(*tasks) for subtask, result in zip(parallel_group, parallel_results): context["artifacts"][subtask["id"]] = result

这样能把“串行写文章”升级成“调研完成后，写作和配图同时进行”。TaoToken 的 Key 和 Base URL 不用变，只是多消耗一次并发请求。

6.2 真实内容生成后，接 MCP 工具会更实用

当 research 真的返回模型生成的文本后，你可以再给 Agent 挂一个工具：把调研结果保存成 markdown 文件，或者调用搜索类 MCP Server 补充资料。TaoToken 提供的统一接入通道不限制这些工具调用方式，模型请求仍然走 https://taotoken.net/api，只是 Agent.think 的返回值会被后续工具进一步处理。

这里要特别说明：工具若需要访问数据库或生产环境，仍应由你在本地执行，把执行结果贴回对话，不要让 Agent 直接连生产库。多智能体编排负责的是任务拆解和内容流转，不是替你去操作生产系统。

6.3 把这次调用记在控制台里

验证到这里，可以回到 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content= 看这次 Codex 和 main.py 一共产生了多少次请求。多智能体编排的 Key、模型 ID、用量记录都落在同一个控制台里，后续无论怎么改 Agent 和调度，这套接入配置都不需要再动。

再进一步，同一把 Key 可以直接在 模型对话 里试不同模型 ID 的输出风格；日常写代码多可以看 Coding Plan；Key 的创建和回收在 控制台 API Keys。如果之后切到 Claude Code，环境变量对照见 接入文档。