企业官网建设流程全解析

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 基址是 https://taotoken.net/api （这个地址不加 UTM 参数，直接用于代码里的 base_url）。

你需要先拿到一个 API Key。进入控制台创建密钥：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，密钥只在创建时完整显示一次，复制后立刻写进本地环境变量，别硬编码进源码。

export TAOTOKEN_API_KEY="sk-你的密钥" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

验证密钥是否可用，最省事的方式是直接在模型对话页发一条消息：https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。如果那边能正常返回，说明密钥和网络链路没问题，再回到本地写循环。

注意：主循环会高频调用模型，建议在控制台里给这个 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_factory=list) 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( model=cfg["agent"]["model"], messages=state.messages, tools=build_tool_schemas(cfg), timeout=cfg["provider"]["timeout_seconds"], ) msg = resp.choices[0].message if msg.tool_calls: call = msg.tool_calls[0] return StepResult( type="tool_call", tool_name=call.function.name, tool_input=json.loads(call.function.arguments), ) return StepResult(type="message", content=msg.content)

循环入口负责把单轮结果接起来，并在每轮结束时构造新的 State：

def 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": f"Tool 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_workers=cfg["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] transition=next_turn tool=read_file args={"path":"workspace/README.md"} [turn 2] transition=next_turn tool=run_shell args={"cmd":"wc -l workspace/README.md"} [turn 3] transition=next_turn tool=write_file args={"path":"workspace/summary.txt","content":"..."} [turn 4] transition=completed type=message 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 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 。密钥管理统一在 API Keys 页面：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ，建议给主循环单独建一个 Key，方便按项目统计用量。

如果你用的是 Claude Code 这类工具做底层验证，Anthropic 兼容入口在：https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude_code&utm_campaign=rewrite ，配置方式和上面 config.toml 里的 provider 段一致，改 base_url 和 api_key_env 即可。

主循环真正的难点从来不是那几十行 while，而是状态怎么显式化、工具结果怎么裁剪、错误怎么恢复。把这三件事在最小骨架里跑通，后面叠加并发、Hook、压缩策略都是在这个骨架上加层，不会推倒重来。