尧图网站建设 
1. 为什么 Planner、Executor、Tool Registry 是 Agent 的骨架很多人第一次写 AI Agent会把所有逻辑塞进一个函数拼 prompt、调模型、解析返回、执行动作、再拼 prompt。跑通一个 demo 没问题但只要任务稍微复杂一点代码就会变成一团乱麻。我试过在一个 300 行的run_agent()里同时处理检索、代码分析、答案校验结果每次加一个新工具都要改五六个地方调试时根本不知道是哪一步出了问题。Planner、Executor、Tool Registry 这三个模块本质上就是把「想」和「做」拆开。Planner 只负责一件事根据用户任务产出一个步骤列表每一步说明要调用哪个工具、传什么参数。Executor 只负责另一件事拿着这个步骤列表一步步执行把工具返回的结果塞进上下文最后生成答案。Tool Registry 则是中间的契约层所有工具在这里注册声明自己的名字、描述、输入输出结构Executor 通过统一接口调用不需要知道每个工具内部怎么实现。这套结构能解决三个实际问题。第一是可观测每一步执行都有记录前端 Trace 能清楚看到 Planner 拆了几步、每步调了什么工具、返回了什么。第二是可扩展新增一个工具只需要在 Registry 里注册Planner 的白名单里加上名字Executor 完全不用改。第三是安全边界模型只能从白名单里选工具不能凭空捏造一个delete_all_files然后被执行。适合谁看如果你已经能调通大模型 API写过简单的 function calling但每次加功能都觉得很乱那这篇就是给你准备的。我会给出可复制的接口定义、注册表配置、一次端到端任务的验证步骤以及用统一 Key/API 通道接入模型调用的方式确保 Planner 和 Executor 链路能跑通、能观测。核心检索词先明确Planner 负责任务拆解与步骤编排Executor 负责按计划调用工具并处理返回Tool Registry 负责工具注册、参数校验与动态发现。这三个模块合起来就是 Agent 的骨架。下面从工程落地角度一步步把它搭出来。2. 前置准备用 TaoToken 统一模型调用通道在写 Planner 之前得先解决模型调用的问题。Planner 需要调 LLM 来生成计划Executor 里的draft_grounded_answer、summarize_context也需要调 LLM。如果每个模块各自配置 API Key、各自处理 base_url代码会重复且难维护。我的做法是抽一个统一的 provider 层所有模型调用都走同一个通道。这里用 TaoToken 作为统一入口。它的 API 地址是https://taotoken.net/api兼容 OpenAI 的接口格式所以你可以直接用openai这个 SDK只需要改base_url和api_key。这样做的好处是Planner、Executor、Summarizer 共用一套配置换模型时只改一个地方。先拿 Key。打开https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite创建一个 API Key复制出来。注意不要把它硬编码进代码放到环境变量里。export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在项目里建一个llm_provider.py封装两个最常用的方法chat()用于普通对话chat_json()用于要求模型返回 JSON 的场景Planner 会用到。# backend/app/services/llm_provider.py import os import json from openai import OpenAI _client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) DEFAULT_MODEL os.environ.get(AGENT_MODEL, gpt-4o-mini) def chat(messages, modelDEFAULT_MODEL, temperature0.2): resp _client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, ) return resp.choices[0].message.content def chat_json(messages, modelDEFAULT_MODEL, temperature0.0): resp _client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, response_format{type: json_object}, ) return json.loads(resp.choices[0].message.content)这里有几个工程细节值得说。第一response_format{type: json_object}能让模型更稳定地返回 JSONPlanner 解析计划时不容易崩。第二temperature在 Planner 里设成 0因为计划需要确定性在draft_grounded_answer里可以设成 0.2 到 0.3让答案自然一点。第三模型名通过环境变量控制方便切换。如果你还没决定用哪个模型可以先在模型对话页面https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite里试几个看看哪个在 JSON 输出和指令遵循上更稳。Planner 对模型的指令遵循能力要求比较高选一个能稳定输出合法 JSON 的模型很重要。配置好之后先写一个最小验证脚本确认通道是通的# scripts/check_llm.py from backend.app.services.llm_provider import chat print(chat([{role: user, content: 只回复两个字通了}]))跑一下python scripts/check_llm.py如果输出「通了」说明模型通道没问题。这一步别跳过后面 Planner 报错时你得先排除是不是 Key 或 base_url 的问题。3. Tool Registry工具注册、参数校验与动态发现Tool Registry 是整个 Agent 的契约中心。它的职责有三块注册工具、校验参数、让 Executor 能按名字动态发现工具。先定义数据结构。# backend/app/services/agent_tools.py from dataclasses import dataclass from typing import Any, Callable, Dict ToolRunner Callable[[Dict[str, Any]], Dict[str, Any]] dataclass(frozenTrue) class AgentTool: name: str description: str input_schema: Dict[str, str] output_schema: Dict[str, str] run: ToolRunner每个工具通过统一格式返回包含四个字段status、data、observation、error。这样 Executor 不需要了解工具内部实现只看status决定下一步看data取结构化结果看observation给模型当上下文。def ok(dataNone, observation): return {status: ok, data: data or {}, observation: observation, error: None} def fail(error, observation): return {status: error, data: {}, observation: observation, error: str(error)}接下来是注册表本身。用一个字典存工具提供register、get_agent_tool、list_tools三个方法。_TOOLS: Dict[str, AgentTool] {} def register(tool: AgentTool) - None: if tool.name in _TOOLS: raise ValueError(ftool already registered: {tool.name}) _TOOLS[tool.name] tool def get_agent_tool(name: str) - AgentTool: if name not in _TOOLS: raise KeyError(funknown tool: {name}) return _TOOLS[name] def list_tools() - list[AgentTool]: return list(_TOOLS.values())参数校验放在 Executor 调用之前或者放在工具内部。我倾向于在 Registry 层提供一个validate_args函数根据input_schema检查必填字段和类型。def validate_args(tool: AgentTool, args: Dict[str, Any]) - None: for key, typ in tool.input_schema.items(): if key not in args: raise ValueError(fmissing arg: {key}) expected {str: str, int: int, float: float, dict: dict, list: list}.get(typ) if expected and not isinstance(args[key], expected): raise TypeError(farg {key} expects {typ}, got {type(args[key]).__name__})现在注册几个核心工具。以retrieve_course_context为例它从课程资料里检索相关 chunk。def _run_retrieve_course_context(args): query args[query] top_k int(args.get(top_k, 5)) chunks vector_store.search(query, top_ktop_k) return ok( data{chunks: chunks}, observationf检索到 {len(chunks)} 个相关片段, ) register(AgentTool( nameretrieve_course_context, description从上传的课程资料中检索与 query 相关的文本片段, input_schema{query: str, top_k: int}, output_schema{chunks: list}, run_run_retrieve_course_context, ))再注册read_project_file注意安全边界禁止绝对路径禁止逃逸项目根目录限制文件扩展名和大小。import os from pathlib import Path PROJECT_ROOT Path(os.environ.get(PROJECT_ROOT, .)).resolve() ALLOWED_EXT {.py, .java, .js, .ts, .vue, .md, .txt, .json, .yml, .yaml} MAX_FILE_SIZE 200_000 def _run_read_project_file(args): rel args[path] target (PROJECT_ROOT / rel).resolve() if not str(target).startswith(str(PROJECT_ROOT)): return fail(path escapes project root) if target.suffix not in ALLOWED_EXT: return fail(fextension not allowed: {target.suffix}) if target.stat().st_size MAX_FILE_SIZE: return fail(file too large) content target.read_text(encodingutf-8, errorsignore) return ok(data{path: rel, content: content}, observationf读取 {rel}{len(content)} 字符)list_project_files要排除.git、node_modules、dist、build、target、venv这些目录避免把无关文件暴露给模型。EXCLUDE_DIRS {.git, node_modules, dist, build, target, venv, __pycache__} def _run_list_project_files(args): limit int(args.get(limit, 50)) results [] for root, dirs, files in os.walk(PROJECT_ROOT): dirs[:] [d for d in dirs if d not in EXCLUDE_DIRS] for f in files: if Path(f).suffix in ALLOWED_EXT: results.append(str(Path(root, f).relative_to(PROJECT_ROOT))) if len(results) limit: return ok(data{files: results}, observationf列出 {len(results)} 个文件) return ok(data{files: results}, observationf列出 {len(results)} 个文件)analyze_code_structure用正则做简单结构分析提取类、方法、依赖。它支持 Java、Python、JavaScript、TypeScript、Vue 的简单结构。import re PATTERNS { py: [r^class\s(\w), r^\s*def\s(\w)], java: [rclass\s(\w), r(?:public|private|protected)\s\w\s(\w)\s*\(], js: [rclass\s(\w), rfunction\s(\w), rconst\s(\w)\s*\s*\(], } def _run_analyze_code_structure(args): path args.get(path) content args.get(content) if path: target (PROJECT_ROOT / path).resolve() content target.read_text(encodingutf-8, errorsignore) ext target.suffix.lstrip(.) else: ext args.get(ext, py) symbols [] for pat in PATTERNS.get(ext, []): symbols.extend(re.findall(pat, content, re.MULTILINE)) return ok(data{symbols: symbols}, observationf提取到 {len(symbols)} 个符号)map_reduce_learning_plan用ThreadPoolExecutor并行执行多个检索 worker再 reduce 成学习计划。这是 Planner 里比较重的一个工具适合「帮我制定学习路线」这类任务。from concurrent.futures import ThreadPoolExecutor, as_completed def _run_map_reduce_learning_plan(args): task args[task] queries args.get(worker_queries) or [task] results [] with ThreadPoolExecutor(max_workers4) as pool: futures {pool.submit(vector_store.search, q, 3): q for q in queries} for fut in as_completed(futures): results.extend(fut.result()) return ok(data{chunks: results}, observationf并行检索 {len(queries)} 路合并 {len(results)} 个片段)注册完这些工具后list_tools()应该能返回完整列表。你可以写个脚本打印出来确认每个工具的name、description、input_schema都对。# scripts/list_tools.py from backend.app.services import agent_tools for t in agent_tools.list_tools(): print(t.name, |, t.description, |, t.input_schema)这一步的输出就是后面 Planner 白名单的来源。Planner 只能从这个列表里选工具不能凭空造。4. Planner 与 Executor可复制的配置与端到端验证Planner 的核心是生成AgentPlanStep列表。先定义步骤结构。# backend/app/services/agent_planner.py from dataclasses import dataclass from typing import Any, Dict, Optional dataclass(frozenTrue) class AgentPlanStep: action: str tool_name: Optional[str] None args: Optional[Dict[str, Any]] NonePlanner 的白名单工具名从 Registry 动态获取这样新增工具不用改 Planner 代码。from backend.app.services import agent_tools ALLOWED_TOOLS [ retrieve_course_context, list_project_files, read_project_file, analyze_code_structure, propose_learning_steps, map_reduce_learning_plan, retrieve_related_code, summarize_context, draft_grounded_answer, verify_grounding, ]LLM Planner 的 prompt 要求模型返回 JSON格式如下{ steps: [ {action: retrieve, tool_name: retrieve_course_context, args: {query: Servlet 生命周期, top_k: 5}}, {action: draft, tool_name: draft_grounded_answer, args: {task: 解释 Servlet 生命周期}}, {action: verify, tool_name: verify_grounding, args: {}} ] }解析时要做防御如果模型返回的工具名不在白名单里直接丢弃这一步或者替换成 fallback。def parse_plan(raw: dict) - list[AgentPlanStep]: steps [] for item in raw.get(steps, []): tool item.get(tool_name) if tool and tool not in ALLOWED_TOOLS: continue steps.append(AgentPlanStep( actionitem.get(action, unknown), tool_nametool, argsitem.get(args) or {}, )) return steps当 LLM Planner 不可用时走 fallback plan。普通任务的 fallback 是load_memory→retrieve_course_context→draft_grounded_answer→verify_grounding→finalize。代码解释任务把retrieve_course_context换成retrieve_related_code。Map-Reduce 任务用map_reduce_learning_plan。FALLBACK_PLANS { general_rag: [load_memory, retrieve_course_context, draft_grounded_answer, verify_grounding, finalize], code_explanation: [load_memory, retrieve_related_code, draft_grounded_answer, verify_grounding, finalize], map_reduce: [load_memory, map_reduce_learning_plan, draft_grounded_answer, verify_grounding, finalize], } def fallback_plan(task_type: str) - list[AgentPlanStep]: names FALLBACK_PLANS.get(task_type, FALLBACK_PLANS[general_rag]) return [AgentPlanStep(actionn, tool_namen, args{}) for n in names]Executor 的职责是遍历 plan为每个 step 准备 tool_args调用get_agent_tool(step.tool_name).run(tool_args)更新 context持久化 AgentStep 和 AgentToolCall自动压缩上下文生成答案校验答案写回 AgentRun。# backend/app/services/agent_executor.py from backend.app.services import agent_tools CONTEXT_COMPRESSION_CHAR_BUDGET 2000 def execute_plan(plan, context, run_id): for step in plan: tool agent_tools.get_agent_tool(step.tool_name) args build_args(step, context) agent_tools.validate_args(tool, args) result tool.run(args) context update_context(context, step, result) persist_step(run_id, step, result) persist_tool_call(run_id, step.tool_name, args, result) if total_chars(context[chunks]) CONTEXT_COMPRESSION_CHAR_BUDGET: summary agent_tools.get_agent_tool(summarize_context).run({chunks: context[chunks]}) context[summaries].append(summary[data]) context[chunks] [] return contextcontext 是 Agent run 的工作内存结构如下context { memory: memory, chunks: [], summaries: [], answer: , citations: [], verification: {passed: False, issues: [not_verified], grounding: needs_revision}, }持久化时_persist_step保存action、status、observation、input_json、output_json、error_message_persist_tool_call保存tool_name、status、input_json、output_json、error_message。_tool_trace_output对工具输出做裁剪避免前端 Trace 展示过大内容。现在做一次端到端验证。假设用户任务是「解释 Servlet 生命周期并给出代码示例」。先分类任务类型为code_explanation然后生成计划。# scripts/run_agent.py from backend.app.services.agent_planner import plan_for_task from backend.app.services.agent_executor import execute_plan task 解释 Servlet 生命周期并给出代码示例 plan plan_for_task(task, task_typecode_explanation) print(plan:, [(s.action, s.tool_name) for s in plan]) context { memory: {}, chunks: [], summaries: [], answer: , citations: [], verification: {passed: False, issues: [not_verified], grounding: needs_revision}, } result execute_plan(plan, context, run_idtest-run-001) print(answer:, result[answer][:200]) print(verification:, result[verification])预期输出plan 里能看到retrieve_related_code、draft_grounded_answer、verify_grounding等步骤answer 里有一段基于检索内容的解释verification 的passed为 True 或给出具体 issues。如果 Planner 走的是 LLM 路径你可以在plan_for_task里加日志打印模型返回的原始 JSON方便排查。如果走 fallback日志里会显示fallback_plan used。验证成功后把 run_id 对应的 AgentStep 和 AgentToolCall 查出来确认每一步都有记录。这一步是后面接前端 Trace 的基础。5. 常见报错排查401、local proxy failed、reading choices、OAuth接入过程中最容易卡在几个固定报错上。下面按真实报错逐个排查。401 Unauthorized。最常见的原因是 API Key 没设对或者环境变量没生效。先确认echo $TAOTOKEN_API_KEY有输出且以sk-开头。如果是在 IDE 里跑检查运行配置有没有继承 shell 环境变量。还有一种情况是 Key 复制时带了空格或换行用python -c import os; print(repr(os.environ[TAOTOKEN_API_KEY]))看一下真实值。确认无误后用最小脚本重试。local proxy failed / connection error。这类报错通常是 base_url 写错或者网络层有额外配置。先确认TAOTOKEN_BASE_URL是https://taotoken.net/api不要多加/v1或结尾斜杠。然后用curl直接测curl -s https://taotoken.net/api/models \ -H Authorization: Bearer $TAOTOKEN_API_KEY | head -c 300如果 curl 通而 Python 不通检查是不是有全局代理环境变量干扰比如HTTP_PROXY、HTTPS_PROXY。在脚本里临时清掉再试import os os.environ.pop(HTTP_PROXY, None) os.environ.pop(HTTPS_PROXY, None)reading choices of undefined。这个报错说明返回体里没有choices字段通常是模型名写错或者接口返回了错误对象。在chat()里加一层防御resp _client.chat.completions.create(...) if not getattr(resp, choices, None): raise RuntimeError(fno choices in response: {resp})然后打印完整 resp看error字段说了什么。常见原因是模型 ID 拼错比如把gpt-4o-mini写成gpt-4-mini。在模型对话页面确认可用模型 ID再填到AGENT_MODEL环境变量里。OAuth / auth.json 相关报错。如果你用的是 Codex 或 Claude Code 这类工具它们会读~/.codex/auth.json或类似配置文件。报 OAuth 错误时先确认配置文件里的base_url和api_key字段。以 Codex 的auth.json为例三件套要写全{ base_url: https://taotoken.net/api, api_key: sk-你的key, model: gpt-4o-mini }Base URL、Key、Model ID 三个缺一不可。如果只填了 Key 没填 base_url工具会去连默认地址导致认证失败。Claude Code 的配置类似在 settings 里指定ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY模型 ID 单独指定。Cline MCP 的配置也是同样逻辑在 MCP server 配置里写全这三项。还有一个容易忽略的点Planner 返回的 JSON 解析失败。如果模型输出带了 markdown 代码块标记json.loads会崩。加一个清洗函数def clean_json(text: str) - str: text text.strip() if text.startswith(): text text.split(\n, 1)[1] text text.rsplit(, 1)[0] return text.strip()排查顺序建议先确认 Key 和 base_url再确认模型 ID再看返回体结构最后看 JSON 解析。大部分问题在前两步就能定位。6. 把链路接稳从验证到长期运行Planner、Executor、Tool Registry 三件套跑通之后下一步是让它稳定运行。几个实用建议。第一给每个工具加超时和重试。retrieve_course_context这类检索工具如果卡住整个 run 都会挂。用concurrent.futures包一层超时from concurrent.futures import ThreadPoolExecutor, TimeoutError def run_with_timeout(tool, args, timeout10): with ThreadPoolExecutor(max_workers1) as pool: fut pool.submit(tool.run, args) try: return fut.result(timeouttimeout) except TimeoutError: return fail(tool timeout)第二把 Planner 的 LLM 调用和 Executor 的工具调用分开记日志。Planner 日志记原始 prompt 和返回 JSONExecutor 日志记每步的 tool_name、args、status、耗时。这样出问题时能快速定位是规划错了还是执行错了。第三上下文压缩的阈值CONTEXT_COMPRESSION_CHAR_BUDGET 2000可以按模型上下文窗口调整。如果用的是长上下文模型可以调到 8000如果模型窗口小调到 1000。压缩时保留 citation 来源不要只留摘要。第四长期编码或 Agent 任务建议用 Coding Plan 来管理调用配额和模型切换地址是https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite。它适合需要持续跑 Agent、频繁调模型的场景比每次手动换 Key 省事。第五接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有完整的接口说明和示例。遇到不确定的参数先查文档再改代码。最后一步验证把run_agent.py跑通后把 run_id 对应的 AgentStep 和 AgentToolCall 导出成 JSON检查每一步的status是否都是okobservation是否有内容error_message是否为空。如果某一步status是error看error_message定位是参数问题还是工具内部问题。这一步做完你的 Agent 骨架就算真正立起来了。 从零开始做一个AI Agent（十一）Planner、Executor 与 Tool Registry
1. 为什么 Planner、Executor、Tool Registry 是 Agent 的骨架很多人第一次写 AI Agent会把所有逻辑塞进一个函数拼 prompt、调模型、解析返回、执行动作、再拼 prompt。跑通一个 demo 没问题但只要任务稍微复杂一点代码就会变成一团乱麻。我试过在一个 300 行的run_agent()里同时处理检索、代码分析、答案校验结果每次加一个新工具都要改五六个地方调试时根本不知道是哪一步出了问题。Planner、Executor、Tool Registry 这三个模块本质上就是把「想」和「做」拆开。Planner 只负责一件事根据用户任务产出一个步骤列表每一步说明要调用哪个工具、传什么参数。Executor 只负责另一件事拿着这个步骤列表一步步执行把工具返回的结果塞进上下文最后生成答案。Tool Registry 则是中间的契约层所有工具在这里注册声明自己的名字、描述、输入输出结构Executor 通过统一接口调用不需要知道每个工具内部怎么实现。这套结构能解决三个实际问题。第一是可观测每一步执行都有记录前端 Trace 能清楚看到 Planner 拆了几步、每步调了什么工具、返回了什么。第二是可扩展新增一个工具只需要在 Registry 里注册Planner 的白名单里加上名字Executor 完全不用改。第三是安全边界模型只能从白名单里选工具不能凭空捏造一个delete_all_files然后被执行。适合谁看如果你已经能调通大模型 API写过简单的 function calling但每次加功能都觉得很乱那这篇就是给你准备的。我会给出可复制的接口定义、注册表配置、一次端到端任务的验证步骤以及用统一 Key/API 通道接入模型调用的方式确保 Planner 和 Executor 链路能跑通、能观测。核心检索词先明确Planner 负责任务拆解与步骤编排Executor 负责按计划调用工具并处理返回Tool Registry 负责工具注册、参数校验与动态发现。这三个模块合起来就是 Agent 的骨架。下面从工程落地角度一步步把它搭出来。2. 前置准备用 TaoToken 统一模型调用通道在写 Planner 之前得先解决模型调用的问题。Planner 需要调 LLM 来生成计划Executor 里的draft_grounded_answer、summarize_context也需要调 LLM。如果每个模块各自配置 API Key、各自处理 base_url代码会重复且难维护。我的做法是抽一个统一的 provider 层所有模型调用都走同一个通道。这里用 TaoToken 作为统一入口。它的 API 地址是https://taotoken.net/api兼容 OpenAI 的接口格式所以你可以直接用openai这个 SDK只需要改base_url和api_key。这样做的好处是Planner、Executor、Summarizer 共用一套配置换模型时只改一个地方。先拿 Key。打开https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite创建一个 API Key复制出来。注意不要把它硬编码进代码放到环境变量里。export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在项目里建一个llm_provider.py封装两个最常用的方法chat()用于普通对话chat_json()用于要求模型返回 JSON 的场景Planner 会用到。# backend/app/services/llm_provider.py import os import json from openai import OpenAI _client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) DEFAULT_MODEL os.environ.get(AGENT_MODEL, gpt-4o-mini) def chat(messages, modelDEFAULT_MODEL, temperature0.2): resp _client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, ) return resp.choices[0].message.content def chat_json(messages, modelDEFAULT_MODEL, temperature0.0): resp _client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, response_format{type: json_object}, ) return json.loads(resp.choices[0].message.content)这里有几个工程细节值得说。第一response_format{type: json_object}能让模型更稳定地返回 JSONPlanner 解析计划时不容易崩。第二temperature在 Planner 里设成 0因为计划需要确定性在draft_grounded_answer里可以设成 0.2 到 0.3让答案自然一点。第三模型名通过环境变量控制方便切换。如果你还没决定用哪个模型可以先在模型对话页面https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite里试几个看看哪个在 JSON 输出和指令遵循上更稳。Planner 对模型的指令遵循能力要求比较高选一个能稳定输出合法 JSON 的模型很重要。配置好之后先写一个最小验证脚本确认通道是通的# scripts/check_llm.py from backend.app.services.llm_provider import chat print(chat([{role: user, content: 只回复两个字通了}]))跑一下python scripts/check_llm.py如果输出「通了」说明模型通道没问题。这一步别跳过后面 Planner 报错时你得先排除是不是 Key 或 base_url 的问题。3. Tool Registry工具注册、参数校验与动态发现Tool Registry 是整个 Agent 的契约中心。它的职责有三块注册工具、校验参数、让 Executor 能按名字动态发现工具。先定义数据结构。# backend/app/services/agent_tools.py from dataclasses import dataclass from typing import Any, Callable, Dict ToolRunner Callable[[Dict[str, Any]], Dict[str, Any]] dataclass(frozenTrue) class AgentTool: name: str description: str input_schema: Dict[str, str] output_schema: Dict[str, str] run: ToolRunner每个工具通过统一格式返回包含四个字段status、data、observation、error。这样 Executor 不需要了解工具内部实现只看status决定下一步看data取结构化结果看observation给模型当上下文。def ok(dataNone, observation): return {status: ok, data: data or {}, observation: observation, error: None} def fail(error, observation): return {status: error, data: {}, observation: observation, error: str(error)}接下来是注册表本身。用一个字典存工具提供register、get_agent_tool、list_tools三个方法。_TOOLS: Dict[str, AgentTool] {} def register(tool: AgentTool) - None: if tool.name in _TOOLS: raise ValueError(ftool already registered: {tool.name}) _TOOLS[tool.name] tool def get_agent_tool(name: str) - AgentTool: if name not in _TOOLS: raise KeyError(funknown tool: {name}) return _TOOLS[name] def list_tools() - list[AgentTool]: return list(_TOOLS.values())参数校验放在 Executor 调用之前或者放在工具内部。我倾向于在 Registry 层提供一个validate_args函数根据input_schema检查必填字段和类型。def validate_args(tool: AgentTool, args: Dict[str, Any]) - None: for key, typ in tool.input_schema.items(): if key not in args: raise ValueError(fmissing arg: {key}) expected {str: str, int: int, float: float, dict: dict, list: list}.get(typ) if expected and not isinstance(args[key], expected): raise TypeError(farg {key} expects {typ}, got {type(args[key]).__name__})现在注册几个核心工具。以retrieve_course_context为例它从课程资料里检索相关 chunk。def _run_retrieve_course_context(args): query args[query] top_k int(args.get(top_k, 5)) chunks vector_store.search(query, top_ktop_k) return ok( data{chunks: chunks}, observationf检索到 {len(chunks)} 个相关片段, ) register(AgentTool( nameretrieve_course_context, description从上传的课程资料中检索与 query 相关的文本片段, input_schema{query: str, top_k: int}, output_schema{chunks: list}, run_run_retrieve_course_context, ))再注册read_project_file注意安全边界禁止绝对路径禁止逃逸项目根目录限制文件扩展名和大小。import os from pathlib import Path PROJECT_ROOT Path(os.environ.get(PROJECT_ROOT, .)).resolve() ALLOWED_EXT {.py, .java, .js, .ts, .vue, .md, .txt, .json, .yml, .yaml} MAX_FILE_SIZE 200_000 def _run_read_project_file(args): rel args[path] target (PROJECT_ROOT / rel).resolve() if not str(target).startswith(str(PROJECT_ROOT)): return fail(path escapes project root) if target.suffix not in ALLOWED_EXT: return fail(fextension not allowed: {target.suffix}) if target.stat().st_size MAX_FILE_SIZE: return fail(file too large) content target.read_text(encodingutf-8, errorsignore) return ok(data{path: rel, content: content}, observationf读取 {rel}{len(content)} 字符)list_project_files要排除.git、node_modules、dist、build、target、venv这些目录避免把无关文件暴露给模型。EXCLUDE_DIRS {.git, node_modules, dist, build, target, venv, __pycache__} def _run_list_project_files(args): limit int(args.get(limit, 50)) results [] for root, dirs, files in os.walk(PROJECT_ROOT): dirs[:] [d for d in dirs if d not in EXCLUDE_DIRS] for f in files: if Path(f).suffix in ALLOWED_EXT: results.append(str(Path(root, f).relative_to(PROJECT_ROOT))) if len(results) limit: return ok(data{files: results}, observationf列出 {len(results)} 个文件) return ok(data{files: results}, observationf列出 {len(results)} 个文件)analyze_code_structure用正则做简单结构分析提取类、方法、依赖。它支持 Java、Python、JavaScript、TypeScript、Vue 的简单结构。import re PATTERNS { py: [r^class\s(\w), r^\s*def\s(\w)], java: [rclass\s(\w), r(?:public|private|protected)\s\w\s(\w)\s*\(], js: [rclass\s(\w), rfunction\s(\w), rconst\s(\w)\s*\s*\(], } def _run_analyze_code_structure(args): path args.get(path) content args.get(content) if path: target (PROJECT_ROOT / path).resolve() content target.read_text(encodingutf-8, errorsignore) ext target.suffix.lstrip(.) else: ext args.get(ext, py) symbols [] for pat in PATTERNS.get(ext, []): symbols.extend(re.findall(pat, content, re.MULTILINE)) return ok(data{symbols: symbols}, observationf提取到 {len(symbols)} 个符号)map_reduce_learning_plan用ThreadPoolExecutor并行执行多个检索 worker再 reduce 成学习计划。这是 Planner 里比较重的一个工具适合「帮我制定学习路线」这类任务。from concurrent.futures import ThreadPoolExecutor, as_completed def _run_map_reduce_learning_plan(args): task args[task] queries args.get(worker_queries) or [task] results [] with ThreadPoolExecutor(max_workers4) as pool: futures {pool.submit(vector_store.search, q, 3): q for q in queries} for fut in as_completed(futures): results.extend(fut.result()) return ok(data{chunks: results}, observationf并行检索 {len(queries)} 路合并 {len(results)} 个片段)注册完这些工具后list_tools()应该能返回完整列表。你可以写个脚本打印出来确认每个工具的name、description、input_schema都对。# scripts/list_tools.py from backend.app.services import agent_tools for t in agent_tools.list_tools(): print(t.name, |, t.description, |, t.input_schema)这一步的输出就是后面 Planner 白名单的来源。Planner 只能从这个列表里选工具不能凭空造。4. Planner 与 Executor可复制的配置与端到端验证Planner 的核心是生成AgentPlanStep列表。先定义步骤结构。# backend/app/services/agent_planner.py from dataclasses import dataclass from typing import Any, Dict, Optional dataclass(frozenTrue) class AgentPlanStep: action: str tool_name: Optional[str] None args: Optional[Dict[str, Any]] NonePlanner 的白名单工具名从 Registry 动态获取这样新增工具不用改 Planner 代码。from backend.app.services import agent_tools ALLOWED_TOOLS [ retrieve_course_context, list_project_files, read_project_file, analyze_code_structure, propose_learning_steps, map_reduce_learning_plan, retrieve_related_code, summarize_context, draft_grounded_answer, verify_grounding, ]LLM Planner 的 prompt 要求模型返回 JSON格式如下{ steps: [ {action: retrieve, tool_name: retrieve_course_context, args: {query: Servlet 生命周期, top_k: 5}}, {action: draft, tool_name: draft_grounded_answer, args: {task: 解释 Servlet 生命周期}}, {action: verify, tool_name: verify_grounding, args: {}} ] }解析时要做防御如果模型返回的工具名不在白名单里直接丢弃这一步或者替换成 fallback。def parse_plan(raw: dict) - list[AgentPlanStep]: steps [] for item in raw.get(steps, []): tool item.get(tool_name) if tool and tool not in ALLOWED_TOOLS: continue steps.append(AgentPlanStep( actionitem.get(action, unknown), tool_nametool, argsitem.get(args) or {}, )) return steps当 LLM Planner 不可用时走 fallback plan。普通任务的 fallback 是load_memory→retrieve_course_context→draft_grounded_answer→verify_grounding→finalize。代码解释任务把retrieve_course_context换成retrieve_related_code。Map-Reduce 任务用map_reduce_learning_plan。FALLBACK_PLANS { general_rag: [load_memory, retrieve_course_context, draft_grounded_answer, verify_grounding, finalize], code_explanation: [load_memory, retrieve_related_code, draft_grounded_answer, verify_grounding, finalize], map_reduce: [load_memory, map_reduce_learning_plan, draft_grounded_answer, verify_grounding, finalize], } def fallback_plan(task_type: str) - list[AgentPlanStep]: names FALLBACK_PLANS.get(task_type, FALLBACK_PLANS[general_rag]) return [AgentPlanStep(actionn, tool_namen, args{}) for n in names]Executor 的职责是遍历 plan为每个 step 准备 tool_args调用get_agent_tool(step.tool_name).run(tool_args)更新 context持久化 AgentStep 和 AgentToolCall自动压缩上下文生成答案校验答案写回 AgentRun。# backend/app/services/agent_executor.py from backend.app.services import agent_tools CONTEXT_COMPRESSION_CHAR_BUDGET 2000 def execute_plan(plan, context, run_id): for step in plan: tool agent_tools.get_agent_tool(step.tool_name) args build_args(step, context) agent_tools.validate_args(tool, args) result tool.run(args) context update_context(context, step, result) persist_step(run_id, step, result) persist_tool_call(run_id, step.tool_name, args, result) if total_chars(context[chunks]) CONTEXT_COMPRESSION_CHAR_BUDGET: summary agent_tools.get_agent_tool(summarize_context).run({chunks: context[chunks]}) context[summaries].append(summary[data]) context[chunks] [] return contextcontext 是 Agent run 的工作内存结构如下context { memory: memory, chunks: [], summaries: [], answer: , citations: [], verification: {passed: False, issues: [not_verified], grounding: needs_revision}, }持久化时_persist_step保存action、status、observation、input_json、output_json、error_message_persist_tool_call保存tool_name、status、input_json、output_json、error_message。_tool_trace_output对工具输出做裁剪避免前端 Trace 展示过大内容。现在做一次端到端验证。假设用户任务是「解释 Servlet 生命周期并给出代码示例」。先分类任务类型为code_explanation然后生成计划。# scripts/run_agent.py from backend.app.services.agent_planner import plan_for_task from backend.app.services.agent_executor import execute_plan task 解释 Servlet 生命周期并给出代码示例 plan plan_for_task(task, task_typecode_explanation) print(plan:, [(s.action, s.tool_name) for s in plan]) context { memory: {}, chunks: [], summaries: [], answer: , citations: [], verification: {passed: False, issues: [not_verified], grounding: needs_revision}, } result execute_plan(plan, context, run_idtest-run-001) print(answer:, result[answer][:200]) print(verification:, result[verification])预期输出plan 里能看到retrieve_related_code、draft_grounded_answer、verify_grounding等步骤answer 里有一段基于检索内容的解释verification 的passed为 True 或给出具体 issues。如果 Planner 走的是 LLM 路径你可以在plan_for_task里加日志打印模型返回的原始 JSON方便排查。如果走 fallback日志里会显示fallback_plan used。验证成功后把 run_id 对应的 AgentStep 和 AgentToolCall 查出来确认每一步都有记录。这一步是后面接前端 Trace 的基础。5. 常见报错排查401、local proxy failed、reading choices、OAuth接入过程中最容易卡在几个固定报错上。下面按真实报错逐个排查。401 Unauthorized。最常见的原因是 API Key 没设对或者环境变量没生效。先确认echo $TAOTOKEN_API_KEY有输出且以sk-开头。如果是在 IDE 里跑检查运行配置有没有继承 shell 环境变量。还有一种情况是 Key 复制时带了空格或换行用python -c import os; print(repr(os.environ[TAOTOKEN_API_KEY]))看一下真实值。确认无误后用最小脚本重试。local proxy failed / connection error。这类报错通常是 base_url 写错或者网络层有额外配置。先确认TAOTOKEN_BASE_URL是https://taotoken.net/api不要多加/v1或结尾斜杠。然后用curl直接测curl -s https://taotoken.net/api/models \ -H Authorization: Bearer $TAOTOKEN_API_KEY | head -c 300如果 curl 通而 Python 不通检查是不是有全局代理环境变量干扰比如HTTP_PROXY、HTTPS_PROXY。在脚本里临时清掉再试import os os.environ.pop(HTTP_PROXY, None) os.environ.pop(HTTPS_PROXY, None)reading choices of undefined。这个报错说明返回体里没有choices字段通常是模型名写错或者接口返回了错误对象。在chat()里加一层防御resp _client.chat.completions.create(...) if not getattr(resp, choices, None): raise RuntimeError(fno choices in response: {resp})然后打印完整 resp看error字段说了什么。常见原因是模型 ID 拼错比如把gpt-4o-mini写成gpt-4-mini。在模型对话页面确认可用模型 ID再填到AGENT_MODEL环境变量里。OAuth / auth.json 相关报错。如果你用的是 Codex 或 Claude Code 这类工具它们会读~/.codex/auth.json或类似配置文件。报 OAuth 错误时先确认配置文件里的base_url和api_key字段。以 Codex 的auth.json为例三件套要写全{ base_url: https://taotoken.net/api, api_key: sk-你的key, model: gpt-4o-mini }Base URL、Key、Model ID 三个缺一不可。如果只填了 Key 没填 base_url工具会去连默认地址导致认证失败。Claude Code 的配置类似在 settings 里指定ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY模型 ID 单独指定。Cline MCP 的配置也是同样逻辑在 MCP server 配置里写全这三项。还有一个容易忽略的点Planner 返回的 JSON 解析失败。如果模型输出带了 markdown 代码块标记json.loads会崩。加一个清洗函数def clean_json(text: str) - str: text text.strip() if text.startswith(): text text.split(\n, 1)[1] text text.rsplit(, 1)[0] return text.strip()排查顺序建议先确认 Key 和 base_url再确认模型 ID再看返回体结构最后看 JSON 解析。大部分问题在前两步就能定位。6. 把链路接稳从验证到长期运行Planner、Executor、Tool Registry 三件套跑通之后下一步是让它稳定运行。几个实用建议。第一给每个工具加超时和重试。retrieve_course_context这类检索工具如果卡住整个 run 都会挂。用concurrent.futures包一层超时from concurrent.futures import ThreadPoolExecutor, TimeoutError def run_with_timeout(tool, args, timeout10): with ThreadPoolExecutor(max_workers1) as pool: fut pool.submit(tool.run, args) try: return fut.result(timeouttimeout) except TimeoutError: return fail(tool timeout)第二把 Planner 的 LLM 调用和 Executor 的工具调用分开记日志。Planner 日志记原始 prompt 和返回 JSONExecutor 日志记每步的 tool_name、args、status、耗时。这样出问题时能快速定位是规划错了还是执行错了。第三上下文压缩的阈值CONTEXT_COMPRESSION_CHAR_BUDGET 2000可以按模型上下文窗口调整。如果用的是长上下文模型可以调到 8000如果模型窗口小调到 1000。压缩时保留 citation 来源不要只留摘要。第四长期编码或 Agent 任务建议用 Coding Plan 来管理调用配额和模型切换地址是https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite。它适合需要持续跑 Agent、频繁调模型的场景比每次手动换 Key 省事。第五接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有完整的接口说明和示例。遇到不确定的参数先查文档再改代码。最后一步验证把run_agent.py跑通后把 run_id 对应的 AgentStep 和 AgentToolCall 导出成 JSON检查每一步的status是否都是okobservation是否有内容error_message是否为空。如果某一步status是error看error_message定位是参数问题还是工具内部问题。这一步做完你的 Agent 骨架就算真正立起来了。