
1. 从零设计 AI Agent 框架为什么我选择自己写一个 Agent LoopAI Agent 框架能做什么简单说它把大模型的推理能力和外部工具的调用能力串成一个闭环让模型不只是聊天而是能读文件、跑命令、执行代码一步步把任务做完。适合谁适合有 Python 基础、想真正搞懂 Agent 底层运行机制、而不是只会调 LangChain 接口的工程师。我见过太多人一上来就装一堆框架结果连 Agent Loop 里上下文是怎么累积的、工具调用结果怎么回填的都说不清。所以这篇我换个思路从零手写一个极简 Agent 框架核心就是 ReAct 推理循环加 Agent Loop 调度全部代码单文件搞定。模型接入这块我用 TaoToken 统一 Key 和 API 通道省去在多个厂商之间来回切换 base_url 和密钥的麻烦一个 Key 就能打通模型调用。整篇会给出可复制的 config.toml 与 settings.json 骨架演示通过 TaoToken 接入模型并附上本地运行验证 Agent 多轮工具调用的完整步骤。你跟着敲一遍就能得到一个能读文件、写文件、执行 Shell 和 Python 代码的 Agent并且清楚它每一轮循环在干什么。2. TaoToken 前置准备统一 Key 与 API 通道2.1 为什么 Agent 框架需要一个统一入口写 Agent 框架时LLM Call 这一层其实没什么工程变量真正烦人的是每家厂商的 API 细节不一样base_url 不同、鉴权头不同、模型名不同。如果你在 Agent 里硬编码某一家后面想换模型就得改代码。TaoToken 在这里的角色就是一个统一的 API 通道。它兼容 OpenAI SDK 的调用格式你只要把 base_url 指向它用同一个 Key 就能调用不同模型。对 Agent 框架来说这意味着 LLM Provider 这一层可以彻底解耦config 里改个模型名就行。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 注意 API 地址不带 UTM 参数。2.2 获取 Key 与确认接入信息进入控制台创建 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。创建好之后你会拿到一串 sk- 开头的密钥先存到环境变量里别写进代码。export TAOTOKEN_API_KEYsk-你的密钥如果你习惯用配置文件管理可以建一个 config.toml把模型和通道信息集中放进去。下面这个骨架你可以直接复制# config.toml — Agent 框架模型配置骨架 [provider] name taotoken base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY [model] # 主推理模型需支持 tool calls chat_model deepseek-chat # 备用模型主模型不可用时切换 fallback_model gpt-4o-mini temperature 0.3 max_tokens 4096 [agent] max_turns 20 tool_timeout 30对应的 settings.json 骨架方便你在代码里读取{ provider: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY }, model: { chat_model: deepseek-chat, temperature: 0.3 }, agent: { max_turns: 20, tool_timeout: 30 } }注意base_url 一定要写成 https://taotoken.net/api 不要多加路径后缀OpenAI SDK 会自动拼接 /v1/chat/completions。2.3 安装依赖Agent 框架本身只需要两个库openai 用于 LLM Calltomli 用于读 config.tomlPython 3.11 以上自带 tomllib可以跳过。pip install openai tomli到这里前置就完成了。你会发现整个准备过程没有复杂的鉴权流程一个 Key 加一个 base_url 就够这也是我选 TaoToken 做统一通道的原因。3. 可复制配置Agent Loop 与 ReAct 循环实现3.1 整体架构与三大要素先把架构理清楚。一个 Agent 框架在工程上就三部分LLM Call、Tools Call、Context Engineering。LLM Call 负责推理Tools Call 负责执行上下文工程负责把两者串起来。而串起来的那个引擎就是 Agent Loop。Agent Loop 本质是一个 while 循环每一次迭代叫一个 Turn每个 Turn 做三件事调 LLM 推理、解析响应里的 tool_calls、执行工具并把结果回填到上下文。当某一次 Turn 的响应里没有 tool_calls 了说明模型认为任务完成循环退出。上下文用一个 messages 列表承载格式就是 OpenAI chat 格式。它累积系统提示词、用户消息、助手响应和工具结果。这个列表就是 Agent 的短期记忆也是上下文工程的核心操作对象。3.2 Agent Loop 核心代码下面这段是 Agent Loop 的核心我加了详细注释你可以直接复制import json import os from openai import OpenAI MAX_TURNS 20 def agent_loop(user_message: str, messages: list, client: OpenAI, tools: dict) - str: Agent Loopwhile 循环驱动 LLM 推理与工具调用。 流程 1. 将用户消息追加到 messages 2. 调用 LLM 3. 若返回 tool_calls → 逐个执行 → 结果追加到 messages → 继续循环 4. 若无 tool_calls → 退出循环返回文本 5. 安全上限 MAX_TURNS 轮 messages.append({role: user, content: user_message}) tool_schemas [t[schema] for t in tools.values()] for turn in range(1, MAX_TURNS 1): # --- LLM Call --- response client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolstool_schemas, ) choice response.choices[0] assistant_msg choice.message messages.append(assistant_msg.model_dump()) # --- 终止条件无 tool_calls --- if not assistant_msg.tool_calls: return assistant_msg.content or # --- 执行每个 tool_call --- for tool_call in assistant_msg.tool_calls: name tool_call.function.name raw_args tool_call.function.arguments print(f [tool] {name}({raw_args})) try: args json.loads(raw_args) except json.JSONDecodeError: args {} tool_entry tools.get(name) if tool_entry is None: result f[error] unknown tool: {name} else: result tool_entry[function](**args) messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) return [agent] reached maximum turns, stopping.这里有个细节值得说assistant_msg.model_dump() 会把 tool_calls 一起序列化进上下文这是下一轮 LLM 能看懂我上一轮调了什么工具的关键。很多人手写时漏掉这一步导致模型反复调同一个工具。3.3 四个工具的实现工具集我保持极简就四个shell_exec、file_read、file_write、python_exec。这四个覆盖了文件操作、命令执行和代码执行足够验证 Agent Loop 的完整链路。import subprocess import sys import tempfile import os def shell_exec(command: str) - str: 执行 shell 命令并返回 stdout stderr。 try: result subprocess.run( command, shellTrue, capture_outputTrue, textTrue, timeout30, ) output result.stdout if result.stderr: output \n[stderr]\n result.stderr if result.returncode ! 0: output f\n[exit code: {result.returncode}] return output.strip() or (no output) except subprocess.TimeoutExpired: return [error] command timed out after 30s except Exception as e: return f[error] {e} def file_read(path: str) - str: 读取文件内容。 try: with open(path, r, encodingutf-8) as f: return f.read() except Exception as e: return f[error] {e} def file_write(path: str, content: str) - str: 将内容写入文件自动创建父目录。 try: os.makedirs(os.path.dirname(path) or ., exist_okTrue) with open(path, w, encodingutf-8) as f: f.write(content) return fOK — wrote {len(content)} chars to {path} except Exception as e: return f[error] {e} def python_exec(code: str) - str: 在子进程中执行 Python 代码并返回输出。 tmp_path None try: with tempfile.NamedTemporaryFile( modew, suffix.py, deleteFalse, encodingutf-8 ) as tmp: tmp.write(code) tmp_path tmp.name result subprocess.run( [sys.executable, tmp_path], capture_outputTrue, textTrue, timeout30, ) output result.stdout if result.stderr: output \n[stderr]\n result.stderr return output.strip() or (no output) except subprocess.TimeoutExpired: return [error] execution timed out after 30s except Exception as e: return f[error] {e} finally: if tmp_path: try: os.unlink(tmp_path) except OSError: pass3.4 工具注册与 System Prompt工具注册就是一个字典映射name 对应函数和 OpenAI function schema。这样 Agent Loop 拿到 LLM 返回的 tool_call 后能按 name 找到要执行的函数。TOOLS { shell_exec: { function: shell_exec, schema: { type: function, function: { name: shell_exec, description: Execute a shell command and return its output., parameters: { type: object, properties: { command: {type: string, description: The shell command to execute.} }, required: [command], }, }, }, }, file_read: { function: file_read, schema: { type: function, function: { name: file_read, description: Read the contents of a file at the given path., parameters: { type: object, properties: { path: {type: string, description: Absolute or relative file path.} }, required: [path], }, }, }, }, file_write: { function: file_write, schema: { type: function, function: { name: file_write, description: Write content to a file (creates parent directories if needed)., parameters: { type: object, properties: { path: {type: string, description: Absolute or relative file path.}, content: {type: string, description: Content to write.}, }, required: [path, content], }, }, }, }, python_exec: { function: python_exec, schema: { type: function, function: { name: python_exec, description: Execute Python code in a subprocess and return its output., parameters: { type: object, properties: { code: {type: string, description: Python source code to execute.} }, required: [code], }, }, }, }, }System Prompt 要明确告诉模型两件事有哪些工具可用以及什么时候该停下来。SYSTEM_PROMPT You are a helpful AI assistant with access to the following tools: 1. shell_exec — run shell commands 2. file_read — read file contents 3. file_write — write content to a file 4. python_exec — execute Python code Think step by step. Use tools when you need to interact with the file system, run commands, or execute code. When the task is complete, respond directly without calling any tool.3.5 主入口与 TaoToken 客户端初始化最后把入口串起来注意 client 的 base_url 指向 TaoTokendef main(): api_key os.environ.get(TAOTOKEN_API_KEY) if not api_key: print(Error: please set TAOTOKEN_API_KEY environment variable.) sys.exit(1) client OpenAI(api_keyapi_key, base_urlhttps://taotoken.net/api) messages [{role: system, content: SYSTEM_PROMPT}] print(Agent ready. Type your message (or exit to quit, clear to reset).\n) while True: try: user_input input(You ).strip() except (EOFError, KeyboardInterrupt): print(\nBye.) break if not user_input: continue if user_input.lower() exit: print(Bye.) break if user_input.lower() clear: messages.clear() messages.append({role: system, content: SYSTEM_PROMPT}) print((context cleared)\n) continue reply agent_loop(user_input, messages, client, TOOLS) print(f\nAgent {reply}\n) if __name__ __main__: main()4. 验证请求本地运行与多轮工具调用实测4.1 启动与第一轮验证设置好环境变量后直接运行export TAOTOKEN_API_KEYsk-你的密钥 python agent.py先问一个简单问题验证通道是否通You 你好介绍一下你自己如果 TaoToken 通道正常模型会返回一段自我介绍并且不会触发任何工具调用Agent Loop 第一轮就退出。这一步验证的是 LLM Call 链路。4.2 触发多轮工具调用接下来问一个需要工具的问题You 帮我查一下当前目录都有哪些文件你会看到终端打印出类似这样的过程[tool] shell_exec({command: ls -la})模型先推理出要执行 lsAgent Loop 执行后把结果回填模型再基于结果组织语言返回。这是一轮工具调用。再试一个更复杂的验证多轮循环You 帮我统计当前目录下所有 .py 文件的代码行数这个过程会看到 Agent Loop 连续调用多个工具先用 shell_exec 找文件再用 python_exec 写统计脚本并执行最后汇总结果。终端会打印出多次 [tool] 调用说明 while 循环在正常迭代。4.3 验证上下文累积多轮对话后问一个依赖上文的问题You 刚才统计的结果里哪个文件行数最多如果上下文累积正确模型能直接引用上一轮的工具结果回答而不需要重新执行。这一步验证的是 messages 列表作为短期记忆是否生效。4.4 用模型对话快速验证通道如果你想单独验证 TaoToken 通道和模型可用性不想跑整个 Agent可以直接用模型对话页面测试https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。在里面发一条消息确认返回正常再回到 Agent 里排查其他问题。5. 本篇常见错误排查5.1 报错 401 Unauthorized最常见的原因是 API Key 没设置或设置错了。检查环境变量echo $TAOTOKEN_API_KEY如果输出为空说明没导出成功。注意 export 只在当前终端会话有效换个终端就没了。另外确认 Key 没有多余空格复制时容易带上换行。5.2 报错 model not found模型名写错了。config.toml 里的 chat_model 要和 TaoToken 支持的模型名完全一致。如果你不确定有哪些模型可用去控制台看模型列表或者用模型对话页面测试模型名。5.3 工具调用死循环如果 Agent 反复调同一个工具停不下来通常是两个原因一是 assistant_msg.model_dump() 没把 tool_calls 存进上下文模型看不到自己调过什么二是 System Prompt 里没写清楚任务完成时直接回复不要调工具。检查这两处。5.4 工具执行超时shell_exec 和 python_exec 都设了 30 秒超时。如果你的命令确实需要更久改 timeout 参数。但更常见的情况是命令本身有问题卡住了比如等待输入。建议在工具里加个非交互式标志或者把超时调小快速失败。5.5 上下文过长导致报错messages 列表会随着轮次增长长任务可能超出模型上下文窗口。极简版没做上下文压缩你可以先手动用 clear 命令重置。生产环境需要做上下文工程比如滑动窗口、摘要压缩、文件系统外置记忆等。5.6 中文乱码file_read 和 file_write 都指定了 encodingutf-8。如果你在 Windows 上遇到乱码检查系统默认编码必要时在 subprocess.run 里加 encodingutf-8。6. 下一步从极简框架到可用的 Agent 应用跑通上面的代码后你手里就有了一个能用的 Agent 框架。它虽然只有两百多行但 LLM Call、Tools Call、Context Engineering 三部分齐全Agent Loop 的每一轮迭代你都能在终端看到。接下来可以往几个方向扩展。工具层可以加网络搜索、API 调用、MCP 接入但注意工具不是越多越好每个工具的 schema 都会占用上下文。上下文工程层可以做短期记忆和长期记忆的分离把不常用的信息外置到文件系统需要时再读回来。调度层可以引入 Plan-and-Execute让模型先出计划再执行适合复杂长任务。如果你打算长期做编码类 Agent 或者跑自动化任务可以了解下 Coding Plan地址是 https://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 。最后说个我踩过的坑极简框架里 messages 是全局累积的跑长任务时上下文会越来越长模型响应变慢甚至报错。我的做法是给 Agent Loop 加一个轮次计数超过阈值就把早期的工具结果替换成摘要。这个改动不大但能让 Agent 稳定跑更久。