自研AI代理的 agent_loop 跑多步工具调用，Key 用 TaoToken-开发者社区

自研AI代理的 agent_loop 跑多步工具调用，Key 用 TaoToken

这篇正文从 Agent/Harness 视角讲自研 AI 代理的agent_loop如何跑通多步工具调用，Key 用 TaoToken。很多读者在把openai.OpenAI的base_url从写死地址改成兼容通道时，容易多写一个/v1，结果chat_with_llm流式文本正常、agent_loop的工具调用决策却出现 404；也有人把ToolRegistry和模型 API 的职责混在一起，以为 TaoToken 会执行read_file、exec_shell。实际 TaoToken 只提供模型 API 的 Key 和 Base URL，工具执行仍然在你自己的进程或沙箱里完成。先到 TaoToken 官网https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_intro创建 Key，API 地址用https://taotoken.net/api（不要带/v1），再把这套配置接入chat_with_llm与agent_loop。

一、原问题与场景：agent_loop 多步工具调用为什么卡在 base_url

类似 OpenClaw 的自研代理，通常会把能力拆成几层：网关/通信层处理用户输入和流式响应，大模型集成层负责对接不同模型，工具调用层注册和执行read_file、exec_shell、网络搜索等能力，工作流/记忆层维护对话历史与多步计划。最开始的 MVP 往往只包含一两个工具，跑通后再逐步增加权限控制、沙箱和审计。

真正进入多步任务后，问题会集中到一个点：模型通道是否足够稳定、是否绑死单一厂商。如果大模型集成层里写死某个厂商的base_url，一旦上游订阅策略或 API 策略变化，整个agent_loop都会受到影响。更麻烦的是，流式输出和工具调用决策可能走了两条不同通道：文本总结用一个地址，tool_calls决策用另一个地址，结果日志分散、Token 消耗不统一、排查超时和断包时很难定位。

所以这里的目标不是重写整个 Agent，而是把模型调用层抽象出来。ToolRegistry仍然负责声明式注册工具、解析参数、执行工具，并在必要时加权限验证和安全沙箱；agent_loop仍然负责把用户任务拆成多步，把工具结果回填给模型，再决定下一步。TaoToken 在这条链路里只做一件事：提供模型 API 的 Key 和 Base URL，让chat_with_llm的流式请求与agent_loop的工具调用决策走同一条兼容通道。它不执行read_file、exec_shell，也不替代你的工具运行时。

二、TaoToken 前置：创建 Key 并确认 Base URL

先到 TaoToken 官网创建 Key，也就是把原来的api_key="your_api_key"换成你在控制台拿到的YOUR_API_KEY。创建入口可以走https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_create_key，创建后在 API Keys 页面复制 Key。Key 不要提交到公开仓库，建议放到环境变量：

export TAOTOKEN_API_KEY="YOUR_API_KEY" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL_ID="MODEL_ID"

这里 Base URL 必须写成https://taotoken.net/api，不要带/v1。原因是openai.OpenAI这个 SDK 在发起chat.completions.create时会自行拼接/chat/completions，如果你把base_url写成https://taotoken.net/api/v1，实际请求可能变成/api/v1/chat/completions，和通道期望的路径不一致，轻则 404，重则流式连接建立失败。模型 ID 则以控制台或接入文档里可见的为准，不要凭记忆写一个不存在的名称。

如果你后续还要把 Claude Code 这类工具也接到同一套模型通道，注意它的配置位置是settings.json，环境变量通常围绕ANTHROPIC_BASE_URL与ANTHROPIC_API_KEY。本篇的自研 Agent 不走settings.json，而是直接改openai.OpenAI的base_url，两者不要混用变量名。

接入文档可以在https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_doc查看，重点确认三件事：Base URL 是否不带/v1、模型 ID 是否可用、请求体是否兼容 OpenAI Chat Completions 格式。对于自研 Agent 来说，这三件事确认完，后面就是工具注册和循环控制。

三、可复制配置：openai.OpenAI、chat_with_llm、ToolRegistry 与 agent_loop

下面给出一份可以直接改 Key 后运行的 Python 示例。它保留openai.OpenAI的调用方式，只把base_url指向 TaoToken，并把流式文本请求与工具调用决策放在同一个client上。工具执行仍然由ToolRegistry完成，TaoToken 不参与read_file和exec_shell的实际执行。

import json import os import subprocess from typing import Any, Callable, Dict, List import openai TAOTOKEN_API_KEY = os.getenv("TAOTOKEN_API_KEY", "YOUR_API_KEY") TAOTOKEN_BASE_URL = "https://taotoken.net/api" # 注意：不要带 /v1 MODEL_ID = os.getenv("TAOTOKEN_MODEL_ID", "MODEL_ID") client = openai.OpenAI( api_key=TAOTOKEN_API_KEY, base_url=TAOTOKEN_BASE_URL, timeout=60.0, ) def chat_with_llm_stream(messages: List[Dict[str, Any]], model: str = MODEL_ID): """流式文本请求，同样走 TaoToken 的兼容通道。""" response = client.chat.completions.create( model=model, messages=messages, stream=True, ) for chunk in response: delta = chunk.choices[0].delta if delta and delta.content: yield delta.content class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict[str, Any]] = {} def register(self, name: str, func: Callable, description: str, parameters: Dict[str, Any]): self._tools[name] = { "function": func, "description": description, "parameters": parameters, } def get_tool_schemas(self) -> List[Dict[str, Any]]: return [ { "type": "function", "function": { "name": name, "description": info["description"], "parameters": info["parameters"], }, } for name, info in self._tools.items() ] def execute(self, name: str, arguments: Dict[str, Any]) -> str: if name not in self._tools: return f"Error: Tool '{name}' not found." try: result = self._tools[name]["function"](**arguments) return str(result) except Exception as e: return f"Error executing {name}: {e}" tool_registry = ToolRegistry() def read_file(file_path: str) -> str: with open(file_path, "r", encoding="utf-8") as f: return f.read() tool_registry.register( name="read_file", func=read_file, description="读取一个文本文件的内容", parameters={ "type": "object", "properties": { "file_path": {"type": "string", "description": "文件路径"} }, "required": ["file_path"], }, ) def exec_shell(command: str, timeout: int = 30) -> str: result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=timeout, ) return f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}\nExit Code: {result.returncode}" tool_registry.register( name="exec_shell", func=exec_shell, description="在受控环境中执行一条 Shell 命令", parameters={ "type": "object", "properties": { "command": {"type": "string", "description": "要执行的命令"}, "timeout": {"type": "integer", "description": "超时时间（秒）"}, }, "required": ["command"], }, ) def agent_loop(user_input: str, max_steps: int = 6) -> str: """多步工具调用主循环：模型决策 -> ToolRegistry 执行 -> 回填 -> 再决策。""" messages: List[Dict[str, Any]] = [ { "role": "system", "content": ( "你是一个可以调用工具的 AI 代理。" "根据用户任务决定是否调用工具；如果需要多步操作，请依次调用工具。" "不要编造工具结果，工具结果会以 role=tool 回填给你。" ), }, {"role": "user", "content": user_input}, ] for step in range(max_steps): response = client.chat.completions.create( model=MODEL_ID, messages=messages, tools=tool_registry.get_tool_schemas(), tool_choice="auto", stream=False, # 工具决策阶段先非流式，便于稳定解析 tool_calls ) msg = response.choices[0].message assistant_msg: Dict[str, Any] = { "role": "assistant", "content": msg.content or "", } if msg.tool_calls: assistant_msg["tool_calls"] = [ { "id": tc.id, "type": "function", "function": { "name": tc.function.name, "arguments": tc.function.arguments, }, } for tc in msg.tool_calls ] messages.append(assistant_msg) if not msg.tool_calls: return msg.content or "" for tool_call in msg.tool_calls: tool_name = tool_call.function.name try: arguments = json.loads(tool_call.function.arguments or "{}") except json.JSONDecodeError as e: tool_result = f"Error: 参数 JSON 解析失败: {e}" else: tool_result = tool_registry.execute(tool_name, arguments) messages.append( { "role": "tool", "tool_call_id": tool_call.id, "content": tool_result, } ) return f"达到最大步数 {max_steps}，仍未得到最终回答。" if __name__ == "__main__": # 最终文本可用流式输出 # for token in chat_with_llm_stream([{"role": "user", "content": "你好"}]): # print(token, end="") print(agent_loop("读取 ./README.md，然后执行 ls -la，最后总结你做了什么。"))

这份配置的关键点有三处。第一，base_url是https://taotoken.net/api，不是https://taotoken.net/api/v1。第二，chat_with_llm_stream和agent_loop内部都使用同一个client，所以流式请求与工具调用决策走同一条兼容通道。第三，ToolRegistry.execute才是真正执行工具的地方，TaoToken 不执行read_file、exec_shell，它只负责模型 API 的 Key 和 Base URL。

如果你希望工具决策也走流式，可以后续把stream=True打开，并聚合chunk.choices[0].delta.tool_calls的增量参数。但建议先用非流式把agent_loop的多步循环跑通，再考虑流式工具调用，否则参数拼接和错误定位会复杂很多。

四、验证请求与成功结果：curl 测试 chat/completions

在跑完整agent_loop之前，先用一个最小请求验证 Key、Base URL 和模型 ID。下面命令中的 API 地址不带 UTM，也不带/v1：

curl -s https://taotoken.net/api/chat/completions \ -H "Authorization: Bearer YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "MODEL_ID", "messages": [{"role": "user", "content": "只回复：通道正常"}], "stream": false }'

如果返回体里能看到choices数组，并且message.content有类似“通道正常”的内容，说明模型通道已经打通。接着运行agent_loop，成功时的日志形态通常如下：

step 1: tool_call read_file {"file_path": "./README.md"} tool result: ... step 2: tool_call exec_shell {"command": "ls -la", "timeout": 30} tool result: STDOUT: ... STDERR: ... Exit Code: 0 final: 我先读取了 README.md，然后执行了 ls -la，目录中包含 ...

这里要区分两个成功指标。第一，模型返回了正确的tool_calls，并且工具名和参数能通过json.loads解析。第二，ToolRegistry真正执行了工具，并把结果以role=tool回填给模型，模型在后续步骤中基于工具结果继续决策。TaoToken 侧你能统一看到模型 Token 消耗，但工具执行日志仍然在你自己的 Agent 进程里。两者配合起来，才算是多步工具调用跑通。

如果只想先验证模型对话是否正常，也可以用模型对话页面发一条消息，入口是https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_chat。这能快速排除 Key 和模型 ID 的问题，再回到代码里排查ToolRegistry和agent_loop。

五、本篇常见错排查：base_url 带 /v1、tool_calls 解析失败

第一个高频错误是base_url多写了/v1。例如写成https://taotoken.net/api/v1，然后openai.OpenAI再拼/chat/completions，请求路径就会和预期不一致。排查时先打印实际请求地址，或者直接用 curl 请求https://taotoken.net/api/chat/completions。如果 curl 正常、SDK 不正常，优先检查base_url是否被环境变量覆盖。

第二个错误是 Key 用错。有人把其他平台的 Key 填进api_key，然后看到 401 或权限错误。本篇的 Key 应该从 TaoToken 创建，Key 值为YOUR_API_KEY的位置需要替换成真实 Key。不要在代码里硬编码后提交到仓库，也不要把 Key 写进前端。

第三个错误是tool_calls参数解析失败。模型返回的tool_call.function.arguments是字符串，不是字典，必须json.loads后再传给ToolRegistry.execute。如果参数为空，可能是"{}"，也要兜底处理。若 JSON 解析失败，把原始字符串和工具名打日志，通常能看出是模型格式问题还是提示词约束不够。

第四个错误是role=tool消息缺少tool_call_id。agent_loop里每条工具结果都必须带对应的tool_call_id，否则下一轮模型请求会报消息序列不合法。顺序也要注意：先是assistant消息，其中包含tool_calls；然后是多个tool消息，分别对应每个工具调用；最后才是下一轮模型请求。

第五个错误是流式模式下直接解析tool_calls。流式返回的delta.tool_calls是分片的，参数可能跨多个 chunk 到达。如果你在流式循环里直接json.loads，很容易半截失败。建议工具决策阶段先用stream=False，最终回答阶段再用chat_with_llm_stream输出文本。这样既不牺牲最终输出的流式体验，又能让多步循环更稳定。

第六个错误是工具执行越权。exec_shell和read_file不应该无限制执行。至少要做工作目录限制、命令白名单、超时控制和沙箱隔离。TaoToken 不负责这些安全边界，它只提供模型 API 的 Key 和 Base URL，工具执行的安全责任在 Agent 自身。把这一点分清楚，后续排查问题时就不会把工具报错误判成模型通道报错。

第七个错误是循环不退出。agent_loop一定要有max_steps，并在模型不返回tool_calls时结束。如果模型反复调用同一个工具，可以在消息里加入“同一工具相同参数不要重复调用”的约束，或者对工具结果做摘要后再回填，避免上下文无限膨胀。

第八个错误是模型 ID 不存在或已下线。控制台能看到哪些模型 ID，就用哪些模型 ID。不要把示例里的MODEL_ID原样运行到生产。可以在配置里加一个启动检查，先发一条最小chat/completions请求，通过后再进入agent_loop。

六、语义一致 CTA：固定模型通道后继续跑 Agent

自研 AI 代理的agent_loop要跑多步工具调用，最稳妥的做法是把模型通道和工具执行彻底分开：模型通道统一走 TaoToken 的 Key 和 Base URL，工具执行留在自己的ToolRegistry和沙箱里。这样换模型、看 Token 消耗、定位流式断包和tool_calls问题时，都有统一的入口。

如果你还在接入阶段，先去 TaoToken 官网创建 Key，然后到 API Keys 页面复制YOUR_API_KEY，再对照接入文档确认base_url为https://taotoken.net/api、模型 ID 可用。创建 Key 入口：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_create_key；API Keys 入口：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_api_keys；接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_doc。

如果你已经能让agent_loop跑通一两步，接下来通常是长期运行、多任务并发和 Agent 工具链扩展。这个阶段可以关注 Coding Plan，入口是https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_coding_plan。验证模型通道是否正常，则可以用模型对话页面https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=agent_loop_chat。把base_url固定到https://taotoken.net/api后，chat_with_llm的流式请求和agent_loop的工具调用决策就能在同一个通道里持续跑下去，而read_file、exec_shell这些工具仍然由你自己的 Agent 安全地执行。