1. 为什么我要自己写一个 AI Agent CLI很多人第一次接触 AI Agent脑子里浮现的是「会聊天的机器人」。但真正跑过一轮工具调用之后你会发现聊天只是最表层的能力。Agent 的本质是让大模型不仅能「说」还能「做」——它根据你给的目标自主决定调用哪些工具、分几步完成再拿工具返回的结果继续推理直到目标达成。我这次要带你从零写一个运行在终端里的 AI Agent CLI。你输入一句目标比如「统计当前目录下所有 Python 文件的总行数并生成 summary.md」它会自动拆解步骤、调用文件系统工具、执行命令、写回结果最后向你汇报。整个过程你能在终端里看到每一轮思考与工具调用非常适合理解 Agent 的底层机制。为什么不用现成框架因为 LangChain、LlamaIndex 这类库把循环、工具、记忆都封装好了新手直接上手容易「会用但不懂」。自己写一遍最小闭环你才会真正明白Agent 大模型 工具调用 循环控制。这三件事拆开看都不复杂难点在于把它们串成一个稳定的调度流程。本文面向有 Python 基础、想搞懂 Agent 内部原理的开发者。读完后你能得到一个可运行的 CLI 项目并且知道每一行代码为什么这么写。模型调用部分我用 TaoToken 的统一 Key 接入这样你不需要在多个厂商之间来回切换配置一个 Key 就能跑通 OpenAI 兼容的 Chat Completions 接口。先说清楚整体架构。我们的 CLI Agent 由四部分组成CLI 入口负责解析命令行参数用 typer 实现LLM 客户端封装模型调用走 OpenAI 兼容接口工具系统定义工具 schema 和实际执行函数Agent 循环驱动「模型思考 → 工具调用 → 结果回填 → 再思考」这条链路。四者职责清晰后续换模型、加工具都只动一个文件。项目结构最终长这样ai-agent-cli/ ├── .env # 环境变量放 API Key ├── requirements.txt # 依赖清单 ├── README.md └── agent/ ├── __init__.py ├── cli.py # CLI 入口 ├── llm.py # 大模型客户端 ├── tools.py # 工具 schema 与执行器 └── loop.py # Agent 核心循环选 Python 的理由很直接生态成熟、代码直观而且绝大多数 Agent 框架都以 Python 为第一公民。你把这套代码跑通后再去看那些框架的源码会发现它们的内核和这里几乎一致只是多了规划、记忆、观测这些外围能力。2. TaoToken 统一 Key 接入前的准备与配置在写代码之前先把模型通道准备好。我选择 TaoToken 的原因是它提供 OpenAI 兼容的统一接口一个 Key 就能调用多种模型省去了在代码里维护多套鉴权逻辑的麻烦。对于 Agent 这种需要频繁调用模型的场景统一入口能明显减少配置成本。第一步是拿到 API Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册登录后进入控制台在 API Keys 页面创建一个新的 Key。创建时建议给它起个能识别的名字比如agent-cli-dev方便后续区分用途。Key 只在创建时完整显示一次记得立刻复制保存。拿到 Key 之后你需要确认两件事Base URL 和 Model ID。TaoToken 的 API 入口是 https://taotoken.net/api 这是 OpenAI 兼容协议的根地址代码里配置base_url时用它即可。Model ID 则取决于你想用哪个模型可以在控制台的模型列表里查看也可以直接调用模型对话页面先试一下效果。这里有个容易踩的坑很多人把 Base URL 写成带/v1的完整路径结果请求 404。OpenAI SDK 会自动在base_url后面拼接/chat/completions所以如果你填的是https://taotoken.net/api/v1最终请求会变成https://taotoken.net/api/v1/chat/completions。具体该不该带/v1以你所用通道的文档为准。我实测下来按文档给的根地址配置最稳妥。配置方式我推荐用.env文件加python-dotenv这样 Key 不会硬编码进代码也不会误提交到 Git。在项目根目录创建.envTAOTOKEN_API_KEYsk-your-taotoken-key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELyour-model-id注意变量名我用的是TAOTOKEN_前缀而不是OPENAI_。这样做的好处是语义清晰你一眼就知道这个 Key 来自哪里。后面在llm.py里读取这几个变量即可。如果你更习惯OPENAI_API_KEY这种命名也可以只要代码里对应改掉就行。依赖安装很简单pip install openai typer rich python-dotenv固定到requirements.txtopenai1.40.0 typer0.12.0 rich13.7.0 python-dotenv1.0.0这里解释一下每个依赖的作用。openai是官方 SDK我们用它来发 Chat Completions 请求它天然支持tools参数typer用来构建命令行入口比 argparse 写起来清爽得多rich负责终端里的彩色输出和 Markdown 渲染让 Agent 的思考过程更易读python-dotenv负责加载.env。四个库都是轻量级的没有重型依赖。关于模型选择Agent 场景对模型的指令遵循能力要求比较高因为它需要准确判断「什么时候该调用工具、调用哪个工具、参数怎么填」。建议选一个工具调用支持良好的模型。你可以在 TaoToken 的模型对话页面先手动测一下输入一个需要调用工具的问题看它是否能正确返回tool_calls结构。确认没问题再写进.env。还有一点要提醒Agent 循环会在一轮任务里多次调用模型token 消耗比单轮对话高不少。建议在开发调试阶段用便宜一点的模型等逻辑跑通、prompt 稳定后再换成更强的模型。TaoToken 控制台能看到用量统计方便你评估成本。3. 可复制的 LLM 客户端与工具系统配置这一节是全文的核心我会把llm.py和tools.py两个文件的完整代码给你你可以直接复制。先看 LLM 客户端封装。新建agent/llm.pyimport os from dataclasses import dataclass, field from typing import Any from dotenv import load_dotenv from openai import OpenAI load_dotenv() dataclass class LLMConfig: api_key: str field( default_factorylambda: os.getenv(TAOTOKEN_API_KEY, ) ) base_url: str field( default_factorylambda: os.getenv( TAOTOKEN_BASE_URL, https://taotoken.net/api ) ) model: str field( default_factorylambda: os.getenv(TAOTOKEN_MODEL, ) ) class LLMClient: OpenAI 兼容 Chat Completions 客户端封装。 def __init__(self, config: LLMConfig | None None) - None: self.config config or LLMConfig() if not self.config.api_key: raise ValueError(缺少 TAOTOKEN_API_KEY请检查 .env 文件。) self.client OpenAI( api_keyself.config.api_key, base_urlself.config.base_url, ) def chat( self, messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None None, ) - Any: 发起一次对话请求返回 ChatCompletion 对象。 kwargs: dict[str, Any] { model: self.config.model, messages: messages, temperature: 0.2, } if tools: kwargs[tools] tools kwargs[tool_choice] auto return self.client.chat.completions.create(**kwargs)两个关键点。第一temperature设成 0.2让 Agent 行为更稳定避免它在需要精确调用工具时过度发挥。第二传入tools时用tool_choiceauto表示由模型自行判断本次是否调用工具、调用哪个。这个参数是 Agent 自主性的来源如果写死成某个函数名就退化成固定流程了。再看工具系统。新建agent/tools.py先写执行函数import subprocess from pathlib import Path WORKSPACE_DIR Path.cwd() def read_file(path: str) - str: 读取指定文件的文本内容。 full_path (WORKSPACE_DIR / path).resolve() if not full_path.is_file(): return f错误文件不存在{path} try: if full_path.stat().st_size 256 * 1024: return 错误文件超过 256KB请使用更精确的方式读取。 return full_path.read_text(encodingutf-8, errorsreplace) except Exception as exc: return f读取文件失败{exc} def write_file(path: str, content: str) - str: 把内容写入指定文件必要时自动创建父目录。 full_path (WORKSPACE_DIR / path).resolve() try: full_path.parent.mkdir(parentsTrue, exist_okTrue) full_path.write_text(content, encodingutf-8) return f已成功写入文件{path}共 {len(content)} 个字符。 except Exception as exc: return f写入文件失败{exc} def list_dir(path: str .) - str: 列出目录下的文件和子目录。 full_path (WORKSPACE_DIR / path).resolve() if not full_path.is_dir(): return f错误目录不存在{path} entries [] for item in sorted(full_path.iterdir()): suffix / if item.is_dir() else entries.append(f{item.name}{suffix}) return \n.join(entries) if entries else 空目录 def run_command(command: str) - str: 在终端中执行一条 shell 命令。 answer input(f是否执行命令 [ {command} ] ? (y/N) ) if answer.strip().lower() ! y: return 用户拒绝了命令执行。 try: result subprocess.run( command, shellTrue, capture_outputTrue, textTrue, timeout30, cwdWORKSPACE_DIR, ) output (result.stdout or ) (result.stderr or ) return output.strip() if output.strip() else 命令执行成功无输出 except subprocess.TimeoutExpired: return 命令执行超时30 秒。 except Exception as exc: return f命令执行失败{exc}接着定义 schema。schema 采用 OpenAI tools 参数要求的 JSON Schema 格式description写得越清楚模型越知道什么时候该调用哪个工具TOOLS_SCHEMA [ { type: function, function: { name: read_file, description: 读取工作目录中某个文件的文本内容。, parameters: { type: object, properties: { path: { type: string, description: 相对于工作目录的文件路径, } }, required: [path], }, }, }, { type: function, function: { name: write_file, description: 把文本内容写入工作目录中的某个文件。, parameters: { type: object, properties: { path: { type: string, description: 相对于工作目录的文件路径, }, content: { type: string, description: 要写入文件的完整内容, }, }, required: [path, content], }, }, }, { type: function, function: { name: list_dir, description: 列出工作目录中某个目录下的文件和子目录。, parameters: { type: object, properties: { path: { type: string, description: 相对于工作目录的目录路径默认为当前目录, } }, required: [], }, }, }, { type: function, function: { name: run_command, description: 在终端中执行一条 shell 命令例如统计文件行数等。, parameters: { type: object, properties: { command: { type: string, description: 要执行的 shell 命令, } }, required: [command], }, }, }, ] TOOL_EXECUTORS { read_file: read_file, write_file: write_file, list_dir: list_dir, run_command: run_command, }这里的设计要点是 schema 和执行器分离。schema 是给模型看的「说明书」执行器是真正干活的 Python 函数。两者通过工具名对应起来TOOL_EXECUTORS这个字典就是映射表。以后加新工具只需要写一个执行函数、加一段 schema、在字典里注册一行循环代码完全不用动。注意run_command里我加了人工确认。这是最重要的安全护栏因为 shell 命令能力太强一旦模型判断失误执行了危险操作后果可能很严重。开发阶段务必保留这个确认步骤。4. 核心循环实现与端到端工具调用验证核心循环是整个项目的心脏。新建agent/loop.pyimport json from typing import Any from rich.console import Console from rich.markdown import Markdown from agent.llm import LLMClient from agent.tools import TOOLS_SCHEMA, TOOL_EXECUTORS SYSTEM_PROMPT 你是一个运行在终端里的 AI Agent 助手。 你的能力 1. 你可以调用工具读取文件、写入文件、列出目录、执行 shell 命令。 2. 你会根据用户目标自主规划步骤并在必要时连续调用多个工具。 3. 只有当目标真正完成时才输出最终总结。 要求 - 优先使用工具完成任务而不是凭空猜测文件内容或命令结果。 - 每次调用工具前先说明你要做什么以及为什么。 - 如果工具返回错误请尝试换一种方式解决。 - 当你判断任务已经完成时停止调用工具并清晰地向用户汇报结果。 console Console() class Agent: def __init__(self, llm: LLMClient, max_iterations: int 12) - None: self.llm llm self.max_iterations max_iterations def run(self, user_prompt: str) - str: messages: list[dict[str, Any]] [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_prompt}, ] for step in range(1, self.max_iterations 1): console.print(f[bold cyan] 第 {step} 轮 [/bold cyan]) response self.llm.chat(messages, toolsTOOLS_SCHEMA) message response.choices[0].message if not message.tool_calls: answer message.content or 任务完成无文本输出 console.print(Markdown(answer)) return answer messages.append( { role: assistant, content: message.content, tool_calls: [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in message.tool_calls ], } ) for tc in message.tool_calls: tool_name tc.function.name try: arguments json.loads(tc.function.arguments or {}) except json.JSONDecodeError: tool_result 错误工具参数不是合法 JSON。 else: tool_result self._execute_tool(tool_name, arguments) console.print(f[yellow]调用工具{tool_name}[/yellow]) console.print(f[dim]参数{tc.function.arguments}[/dim]) console.print(f[dim]结果{tool_result}[/dim]) messages.append( { role: tool, tool_call_id: tc.id, content: str(tool_result), } ) return 已达到最大迭代次数Agent 停止了运行。 staticmethod def _execute_tool(name: str, arguments: dict[str, Any]) - str: executor TOOL_EXECUTORS.get(name) if executor is None: return f未知工具{name} try: return str(executor(**arguments)) except TypeError as exc: return f工具参数错误{exc} except Exception as exc: return f工具执行异常{exc}这段代码里有个非常容易踩坑的细节roletool的消息必须携带与tool_calls中一致的tool_call_id。否则 OpenAI 兼容接口会直接报错提示找不到对应的工具调用。我第一次写的时候就是漏了这个字段排查了半天。循环的退出条件有两个模型不再返回tool_calls说明它认为任务完成直接输出文本或者达到max_iterations防止死循环。max_iterations设成 12 是个经验值大多数任务几轮就能完成留点余量应对复杂场景。最后写 CLI 入口新建agent/cli.pyimport typer from rich.console import Console from agent.llm import LLMClient from agent.loop import Agent app typer.Typer(help从零实现的 AI Agent CLI) console Console() app.command() def run( prompt: str typer.Argument(..., help要交给 Agent 完成的目标), ): 执行单次任务。 try: agent Agent(LLMClient()) agent.run(prompt) except Exception as exc: console.print(f[red]运行失败{exc}[/red]) raise typer.Exit(code1) app.command() def repl(): 进入交互式模式连续和 Agent 对话。 agent Agent(LLMClient()) console.print([green]AI Agent CLI 已启动输入 exit 退出。[/green]) while True: try: user_input console.input([bold]你 [/bold]) except (EOFError, KeyboardInterrupt): console.print(\n再见) break if user_input.strip().lower() in {exit, quit, q}: console.print(再见) break if not user_input.strip(): continue console.print([bold]Agent [/bold]) agent.run(user_input) if __name__ __main__: app()agent/__init__.py保持空文件即可。现在验证整条链路。确保虚拟环境已激活在项目根目录执行python -m agent.cli run 统计当前目录下所有 .py 文件的总行数并把结果写入 py_lines.txt你会看到类似下面的执行过程 第 1 轮 调用工具run_command 参数{command: find . -name *.py | xargs wc -l | tail -n 1} 是否执行命令 [ find . -name *.py | xargs wc -l | tail -n 1 ] ? (y/N) y 结果 1231 total 第 2 轮 调用工具write_file 参数{path: py_lines.txt, content: 共 1231 行} 结果已成功写入文件py_lines.txt共 13 个字符。 第 3 轮 已完成任务统计了当前目录下所有 .py 文件的总行数共 1231 行 并已将结果写入 py_lines.txt。看到这个输出说明工具调用、结果回填、循环调度全部跑通了。你也可以进入交互模式连续对话python -m agent.cli repl输入「看看当前目录有哪些文件」Agent 会调用list_dir并返回结果。到这里一个能真正「动手」的 Agent CLI 就完成了。5. 常见报错排查401、local proxy failed 与 reading choices跑通之后我把开发过程中遇到的高频报错整理出来你大概率会碰到其中几个。401 Unauthorized。这是最常见的鉴权错误通常有三个原因。一是 Key 没读到检查.env文件是否在项目根目录、变量名是否和代码里一致。二是 Key 复制时带了空格或换行建议重新复制一次。三是 Base URL 配错比如把https://taotoken.net/api写成了别的路径。排查方法是在LLMClient.__init__里打印一下self.config.base_url和api_key[:8]确认值符合预期。local proxy failed / connection error。这类报错说明请求根本没发出去问题出在网络层。先确认你的网络能正常访问配置的 Base URL可以用curl手动测一下。如果公司网络有出口限制可能需要走内部允许的通道。注意不要使用任何非合规的网络工具这类做法既不稳定也不安全。如果确认网络正常还报错检查base_url是否多了或少了斜杠SDK 拼接路径时对斜杠敏感。Error code: 400 - reading choices。这个报错通常出现在你试图访问response.choices[0]但响应结构不对时。原因可能是模型返回了错误信息而不是正常的 completion比如参数不合法、模型 ID 写错。建议在chat方法里加一层判断先看response有没有choices属性。另一个常见原因是tools参数格式不对比如 schema 里少了type: function这一层接口会直接拒绝。tool_call_id 不匹配。报错信息类似「messages with role tool must be a response to a preceding message with tool_calls」。这就是前面提到的坑回填工具结果时tool_call_id必须和 assistant 消息里的id完全一致。检查你的循环代码确认tc.id被正确传递。OAuth / authentication 相关报错。如果你用的是需要 OAuth 的通道注意 Agent 场景下每次请求都要带有效凭证token 过期会导致循环中途失败。建议在LLMClient里加一个简单的重试逻辑遇到鉴权失败时刷新凭证再试一次。不过用 TaoToken 的统一 Key 方式这个问题基本不会遇到因为 Key 是长期有效的。模型不返回 tool_calls。有时候模型会直接用文本回答而不是调用工具。这通常是 prompt 或模型能力的问题。先确认你选的模型支持工具调用再检查SYSTEM_PROMPT是否明确要求「优先使用工具」。如果还是不行可以在用户消息里加一句「请使用工具完成不要凭空回答」。循环停不下来。如果 Agent 反复调用同一个工具检查工具返回的结果是否清晰。比如read_file返回了错误信息模型可能会不断重试。可以在SYSTEM_PROMPT里加一条「如果同一个工具连续失败两次请停止并汇报问题」。另外max_iterations是最后一道保险别设太大。排查这类问题的通用思路是先看报错发生在哪一层。401 和 OAuth 属于鉴权层local proxy failed 属于网络层reading choices 和 tool_call_id 属于协议层模型不返回 tool_calls 属于 prompt 层。定位到层问题就解决了一半。6. 继续演进从最小闭环到生产级 Agent跑通最小闭环之后你会发现 LangChain、AutoGPT 这些框架本质上都是在这个骨架上叠加能力。下面几个方向可以让你的 CLI 更实用。多轮记忆。现在的repl模式每轮都是独立任务上下文不共享。你可以把messages持久化到 SQLite下次启动时恢复。关键是把 system prompt 和历史的 user/assistant/tool 消息一起存下来注意 tool 消息的tool_call_id也要存否则恢复后会报错。流式输出。把chat方法改成streamTrue让模型的思考过程实时显示。这对长任务体验提升很明显你能看到 Agent 正在「想」什么而不是干等。注意流式模式下tool_calls是分片返回的需要自己拼接比非流式复杂一些。更多工具。网页搜索、HTTP 请求、PDF 解析、代码执行沙箱都是常见扩展。加工具时记住那个原则schema 描述要清楚执行器要做好错误处理。工具返回的错误信息越具体模型越容易自我纠正。子任务规划。引入 Plan-and-Execute 模式让模型先制定完整计划再逐步执行。这适合步骤多、依赖复杂的任务。实现上可以先让模型输出一个 JSON 格式的计划然后循环执行每一项。观测与回放。把每一步的messages序列保存成 JSON 文件出问题时可以完整复现。这对调试 Agent 特别有用因为它的行为有随机性不记录就很难定位。人工中断。监听键盘中断让用户能在任意时刻暂停或否决 Agent 的操作。run_command的确认机制是单点的更完善的做法是在循环层面加一个「每轮询问是否继续」的开关。如果你打算把这个 CLI 用在日常编码里可以考虑接入 TaoToken 的 Coding Plan它在长期编码和 Agent 场景下有更合适的用量方案。配置方式还是那三件套Base URL 用 https://taotoken.net/api Key 从控制台获取Model ID 按需选择。想先体验模型效果的话模型对话页面可以直接试需要管理多个 Key 就去 API Keys 页面接入细节看文档。把这几处地址收藏好后续换模型、加工具时能省不少时间。最后说个实用技巧调试 Agent 时把temperature临时设成 0让行为完全确定方便复现问题。等逻辑稳定后再调回 0.2。另外SYSTEM_PROMPT值得反复打磨它直接决定 Agent 的规划质量和工具使用习惯比调参的收益大得多。
求解)
