★ FEATURED ARTICLE 1. 为什么“一问一答”撑不起真正的 Agent很多人第一次写 Agent代码长这样把用户输入拼进 prompt调一次模型拿到回复结束。跑 demo 没问题一旦任务需要“先查文件、再改代码、再跑测试、失败了再改”这套请求-响应模式立刻崩掉——因为模型只有一次决策机会它没法根据中间结果调整下一步。AI Agent 的 Runtime Loop主循环解决的正是这件事。它把模型从“回答者”变成“决策者”每一轮模型只做一件事要么直接给用户回复并结束要么发出一个工具调用由运行时执行后把结果写回上下文再进入下一轮。循环持续到模型不再请求工具、或触发终止条件为止。这篇聚焦底层实现机制用配置文件与运行骨架做切入点拆解主循环的调度、状态流转与工具调用链路。你会拿到可复制的 config.toml / settings.json 骨架以及逐步验证动作在本地跑通一个最小可用的 Agent 主循环并观察它的行为。适合已经会调 API、但想把“单次问答”升级成“持续决策系统”的开发者。核心检索词就三个AI Agent、Runtime Loop、底层实现。2. 前置准备用 TaoToken 统一模型入口主循环要跑起来第一件事是让模型调用稳定可用。我习惯把模型访问层单独抽出来通过 TaoToken 统一走 OpenAI 兼容协议这样主循环代码不用关心背后是哪个模型。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 这个地址不加 UTM 参数直接用于代码里的 base_url。你需要先拿到一个 API Key。进入控制台创建密钥https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 密钥只在创建时完整显示一次复制后立刻写进本地环境变量别硬编码进源码。export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api验证密钥是否可用最省事的方式是直接在模型对话页发一条消息https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。如果那边能正常返回说明密钥和网络链路没问题再回到本地写循环。注意主循环会高频调用模型建议在控制台里给这个 Key 设置额度上限避免调试时循环失控把额度跑光。3. 可复制的配置骨架config.toml 与 settings.json主循环的行为几乎都由配置驱动最大轮次、模型、工具白名单、上下文预算、终止策略。把这些从代码里抽出来循环逻辑才能保持干净。先看 config.toml它描述“运行时怎么跑”[agent] name minimal-loop max_turns 12 # 硬性轮次上限防止死循环 model claude-sonnet-4-5 fallback_model gpt-4o-mini # 主模型过载时降级 system_prompt_file ./prompts/system.md [provider] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY # 从环境变量读取不落盘 timeout_seconds 60 [context] max_input_tokens 120000 tool_result_budget 8000 # 单个工具结果裁剪上限 auto_compact_threshold 0.85 # 上下文占用超过 85% 触发摘要压缩 [tools] enabled [read_file, write_file, run_shell, search] concurrency_safe [read_file, search] # 只读工具可并行 max_parallel 10 [loop] stop_on_no_tool_call true max_output_recovery 3 # 输出被截断时的续写次数再看 settings.json它描述“工具怎么被允许调用”相当于权限层{ permissions: { read_file: { allow: true, scope: ./workspace }, write_file: { allow: true, scope: ./workspace, require_confirm: false }, run_shell: { allow: true, deny_patterns: [rm -rf, curl | sh] }, search: { allow: true } }, hooks: { stop: [./hooks/lint_check.sh] } }两个文件的分工要记牢config.toml 决定循环的“节奏”settings.json 决定工具的“边界”。主循环每轮执行工具前都要拿 settings.json 里的权限规则做一次判断不允许就直接把拒绝结果写回上下文让模型知道这条路走不通。4. 主循环骨架状态对象与单轮调度生产级主循环不会把状态散落在局部变量里而是用一个显式 State 对象承载所有循环信息。下面这个骨架可以直接跑我把它拆成“状态定义 单轮执行 循环入口”三段。from dataclasses import dataclass, field from typing import Any, Literal dataclass class LoopState: messages: list[dict] field(default_factorylist) turn_count: int 0 max_output_recovery_count: int 0 transition: str | None None # 记录上一轮为何继续便于调试 dataclass class StepResult: type: Literal[message, tool_call] content: str | None None tool_name: str | None None tool_input: dict | None None单轮调度只做三件事调模型拿决策、判断决策类型、决定继续还是终止。def model_step(state: LoopState, cfg: dict) - StepResult: resp client.chat.completions.create( modelcfg[agent][model], messagesstate.messages, toolsbuild_tool_schemas(cfg), timeoutcfg[provider][timeout_seconds], ) msg resp.choices[0].message if msg.tool_calls: call msg.tool_calls[0] return StepResult( typetool_call, tool_namecall.function.name, tool_inputjson.loads(call.function.arguments), ) return StepResult(typemessage, contentmsg.content)循环入口负责把单轮结果接起来并在每轮结束时构造新的 Statedef run_loop(user_input: str, cfg: dict, settings: dict) - str: state LoopState(messages[{role: user, content: user_input}]) while state.turn_count cfg[agent][max_turns]: state trim_context(state, cfg) # 上下文预算管理 step model_step(state, cfg) if step.type message: state.messages.append({role: assistant, content: step.content}) if run_stop_hooks(state, settings): # 终止前检查 return step.content return step.content # tool_call 分支 if not is_tool_allowed(step.tool_name, settings): state.messages.append({ role: tool, content: fTool not allowed: {step.tool_name}, }) state.turn_count 1 state.transition tool_denied continue result execute_tool(step.tool_name, step.tool_input, cfg) state.messages.append({role: tool, content: result}) state.turn_count 1 state.transition next_turn return Agent stopped: reached max_turns这段骨架里transition字段是关键的可观测性设计。它记录每一轮为什么继续——是正常工具调用、还是权限拒绝、还是压缩重试。调试循环时你只要打印这个字段就能还原整个决策路径不用靠猜。5. 工具调用链路并行分区与结果回流工具执行不是简单 for 循环。只读工具读文件、搜索可以并行有副作用的工具写文件、执行命令必须串行否则会出现“边读边写”的竞态。def partition_tool_calls(calls: list[dict], cfg: dict): safe set(cfg[tools][concurrency_safe]) batches, current [], [] for call in calls: if call[name] in safe: current.append(call) else: if current: batches.append((parallel, current)) current [] batches.append((serial, [call])) if current: batches.append((parallel, current)) return batches执行时按批次走并行批次用线程池串行批次逐个执行def execute_batch(batch_type: str, calls: list[dict], cfg: dict) - list[str]: if batch_type parallel: with ThreadPoolExecutor(max_workerscfg[tools][max_parallel]) as pool: futures [pool.submit(execute_tool, c[name], c[input], cfg) for c in calls] return [f.result() for f in futures] return [execute_tool(c[name], c[input], cfg) for c in calls]结果回流有个容易踩的坑工具返回可能非常长比如一次 grep 返回上万行。必须在写回上下文前裁剪否则下一轮模型调用直接超上下文窗口。def trim_tool_result(result: str, budget: int) - str: if len(result) budget: return result head result[: budget // 2] tail result[-budget // 2 :] return f{head}\n...[truncated {len(result) - budget} chars]...\n{tail}裁剪策略用“头尾保留”而不是简单截断因为工具输出的开头通常是结构信息结尾往往是错误或结论中间才是可丢弃的重复内容。6. 运行验证观察一次完整的主循环配置和骨架就位后跑一个需要多轮工具调用的任务来验证。比如让 Agent“读取 workspace 下的 README.md统计行数然后写一个 summary.txt”。启动脚本python -m agent.run --config ./config.toml --settings ./settings.json \ --input 读取 workspace/README.md统计行数写入 workspace/summary.txt预期你会看到类似这样的轮次日志[turn 1] transitionnext_turn toolread_file args{path:workspace/README.md} [turn 2] transitionnext_turn toolrun_shell args{cmd:wc -l workspace/README.md} [turn 3] transitionnext_turn toolwrite_file args{path:workspace/summary.txt,content:...} [turn 4] transitioncompleted typemessage content已完成summary.txt 已写入四轮里前三轮都是 tool_call第四轮模型不再请求工具直接返回 message循环终止。如果你只看到一轮就结束说明模型没被正确告知有工具可用——检查build_tool_schemas是否把工具定义传进了请求。验证成功的结果有两个硬指标一是workspace/summary.txt确实被创建且内容正确二是日志里transition字段完整记录了每一轮的继续原因。这两个都对上说明主循环的调度、状态流转、工具链路全部打通。7. 本篇常见错排查循环跑满 max_turns 不终止。最常见原因是工具结果写回时 role 用错了。工具结果必须以tool角色、并带上对应的tool_call_id回写否则模型看不到结果会反复请求同一个工具。检查state.messages.append那几行的 role 字段。模型一直返回 message 不调工具。要么工具 schema 没传要么 system prompt 里没说明“需要操作文件时必须调用工具”。在 prompts/system.md 里明确写一句“你只能通过工具读写文件不要凭空编造文件内容”。上下文超限报 prompt_too_long。说明裁剪没生效。检查trim_context是否在每轮模型调用前执行以及tool_result_budget是否设得过大。生产环境里单轮工具结果建议控制在 8000 字符以内。并行工具出现文件读写冲突。说明concurrency_safe白名单配错了把 write_file 这类有副作用的工具也放进了并行批次。只读工具才允许并行写操作一律串行。权限拒绝后模型卡死。工具被 settings.json 拒绝后拒绝信息要作为 tool 结果写回模型才知道换路径。如果直接抛异常中断循环模型永远拿不到反馈。8. 把主循环接进长期编码与 Agent 工作流最小主循环跑通后下一步通常是把它接到真实的编码或 Agent 场景里这时候单次调试的 Key 就不够用了需要更稳定的调用配额和更长的会话管理。如果你打算长期跑编码类 Agent可以看下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合持续性的多轮任务。接入细节和参数说明都在接入文档里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。密钥管理统一在 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 建议给主循环单独建一个 Key方便按项目统计用量。如果你用的是 Claude Code 这类工具做底层验证Anthropic 兼容入口在https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite 配置方式和上面 config.toml 里的 provider 段一致改 base_url 和 api_key_env 即可。主循环真正的难点从来不是那几十行 while而是状态怎么显式化、工具结果怎么裁剪、错误怎么恢复。把这三件事在最小骨架里跑通后面叠加并发、Hook、压缩策略都是在这个骨架上加层不会推倒重来。
阅读完成 · 觉得有帮助？
