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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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。在项目根目录创建.env:
TAOTOKEN_API_KEY=sk-your-taotoken-key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=your-model-id注意变量名我用的是TAOTOKEN_前缀,而不是OPENAI_。这样做的好处是语义清晰,你一眼就知道这个 Key 来自哪里。后面在llm.py里读取这几个变量即可。如果你更习惯OPENAI_API_KEY这种命名,也可以,只要代码里对应改掉就行。
依赖安装很简单:
pip install openai typer rich python-dotenv固定到requirements.txt:
openai>=1.40.0 typer>=0.12.0 rich>=13.7.0 python-dotenv>=1.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.py:
import 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_factory=lambda: os.getenv("TAOTOKEN_API_KEY", "") ) base_url: str = field( default_factory=lambda: os.getenv( "TAOTOKEN_BASE_URL", "https://taotoken.net/api" ) ) model: str = field( default_factory=lambda: 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_key=self.config.api_key, base_url=self.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_choice="auto",表示由模型自行判断本次是否调用工具、调用哪个。这个参数是 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(encoding="utf-8", errors="replace") 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(parents=True, exist_ok=True) full_path.write_text(content, encoding="utf-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, shell=True, capture_output=True, text=True, timeout=30, cwd=WORKSPACE_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.py:
import 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, tools=TOOLS_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}"这段代码里有个非常容易踩坑的细节:role=tool的消息必须携带与tool_calls中一致的tool_call_id。否则 OpenAI 兼容接口会直接报错,提示找不到对应的工具调用。我第一次写的时候就是漏了这个字段,排查了半天。
循环的退出条件有两个:模型不再返回tool_calls,说明它认为任务完成,直接输出文本;或者达到max_iterations,防止死循环。max_iterations设成 12 是个经验值,大多数任务几轮就能完成,留点余量应对复杂场景。
最后写 CLI 入口,新建agent/cli.py:
import 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(code=1) @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方法改成stream=True,让模型的思考过程实时显示。这对长任务体验提升很明显,你能看到 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 的规划质量和工具使用习惯,比调参的收益大得多。