ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Agent-Reach实战:用Python构建CLI AI Agent,搞定工具调用与并发

Agent-Reach实战:用Python构建CLI AI Agent,搞定工具调用与并发 1. 从标题说起Agent-Reach 到底想解决什么问题第一次看到 Agent-Reach 这个名字我脑子里冒出来的第一个念头是这又是一个把 AI Agent 包装成 CLI 工具的项目。但仔细琢磨了一下 Reach 这个词再结合最近圈子里反复被提到的几个关键词——CLI、AI Agent、Python——我大概能猜到它想干的事情让 AI Agent 不再只是待在网页对话框里聊天而是能真正伸手够到本地环境、够到命令行、够到真实的工作流。说白了现在大部分人对 AI Agent 的使用还停留在我打字它回复的阶段。你问它一个问题它给你一段答案然后你复制粘贴到自己的终端里跑。这个过程中间有一道人工的墙Agent 的能力被卡在对话框里出不来。Agent-Reach 这类项目的核心诉求就是把这堵墙拆掉让 Agent 直接通过 CLI 去执行命令、读写文件、调用工具形成一个闭环。我之所以对这个方向感兴趣是因为过去大半年我一直在折腾各种 AI Agent 的落地场景。从最早的简单 Prompt 拼接到后来用 LangChain 搭链再到 LangGraph 做状态机踩过的坑不算少。很多项目看起来架构很漂亮但真到了让 Agent 干活这一步就拉胯——要么是工具调用不稳定要么是上下文管理一团糟要么是并发一上来就崩。所以当我看到 Agent-Reach 这个标题第一反应不是又一个新框架而是它怎么解决 Agent 真正落地干活的问题。这篇文章我打算从几个层面来拆先讲清楚 Agent-Reach 这类 CLI 形态的 AI Agent 的整体设计思路然后深入到核心细节包括工具调用、上下文管理、并发处理这些硬骨头再给出一套可以照着复现的实操流程最后把我自己踩过的坑和排查经验整理出来。不管你是刚接触 AI Agent 的新手还是已经搭过几个项目想优化架构的老手应该都能从里面找到点有用的东西。提示本文讨论的 Agent-Reach 是一个基于 CLI 的 AI Agent 项目形态核心是用 Python 构建、通过命令行交互、能够调用本地工具完成实际任务。文中涉及的具体实现细节部分是基于这类项目的常见实践做的合理推演供参考复现。2. 整体设计思路为什么是 CLI 而不是 Web2.1 CLI 形态的 Agent 到底香在哪里很多人一提到 AI Agent脑子里浮现的就是一个聊天窗口。但你如果真在开发环境里用过 Agent就会发现 Web 界面其实是个累赘。原因很简单开发者的工作流本来就在终端里。你写代码用终端跑测试用终端部署用终端Git 操作也在终端。如果 Agent 非要你切到浏览器去对话那它就没法融入你的工作流只能算个外挂。CLI 形态的 Agent 最大的优势是上下文天然对齐。Agent 运行在你的项目目录下它能直接看到你的文件结构、读取你的配置文件、执行你的构建命令。你不需要把代码复制粘贴给它它自己就能cat出来。这种在场感是 Web 界面给不了的。另一个优势是可组合性。CLI 工具天生就能被管道、脚本、CI/CD 流程调用。你可以让 Agent-Reach 在 Git Hook 里跑可以在 Makefile 里调可以用 cron 定时触发。它就是一个普通的命令行程序遵循 Unix 哲学做好一件事然后能被别的工具组合。还有一个容易被忽略的点资源占用和响应速度。Web 界面要跑前端、要维护 WebSocket 连接、要处理各种 UI 状态。CLI 把这些全砍掉启动快、内存小、没有浏览器兼容性问题。对于我这种经常在服务器上干活的人来说这一点太重要了。2.2 核心架构拆解Python 做胶水工具做手脚Agent-Reach 这类项目的架构我理解下来大概是这么几层最底层是执行层负责真正去跑命令、读写文件、调用 API。这一层用 Python 的subprocess、pathlib、requests这些标准库就能搞定不需要什么花哨的东西。中间是工具层把执行层的能力包装成 Agent 能理解的工具。每个工具有一个名字、一段描述、一组参数定义。Agent 根据用户意图决定调用哪个工具、传什么参数。这一层的设计直接决定了 Agent 的能力边界。上面是推理层也就是大模型所在的位置。它接收用户输入和工具列表输出我要调用哪个工具的决策。这一层通常通过 API 调用远程模型或者本地跑一个小模型。最上面是交互层也就是 CLI 界面。负责接收用户输入、展示 Agent 的思考和执行过程、处理中断和确认。用 Python 来做这件事是很自然的选择。Python 的生态太全了调 API 有 requests 和 httpx处理数据有 pandas跑子进程有 subprocess做 CLI 有 argparse 和 click。而且 Python 的语法门槛低改起来快适合这种需要反复迭代的项目。2.3 为什么不用 Rust 或者 Go最近确实看到不少基于 Rust 语言的 AI Agent的讨论性能好、二进制分发方便。但我的看法是Agent 的瓶颈不在语言性能而在模型推理速度和工具调用的稳定性。你用 Rust 写模型该慢还是慢API 该超时还是超时。Python 的那点性能开销在整个链路里占比微乎其微。反过来Python 的开发效率优势在这个场景下被放大了。你要快速试一个新的工具定义、改一下 Prompt 模板、调一下上下文截断策略Python 改完直接跑Rust 还得编译。对于还在探索阶段的项目这个迭代速度的差距是决定性的。当然如果你的 Agent 需要处理海量并发、需要极低的延迟、需要嵌入到对性能敏感的系统里那 Rust 或者 Go 确实更合适。但对于个人开发者和小团队来说Python 是性价比最高的选择。3. 核心细节解析工具调用、上下文与并发3.1 工具定义Agent 的手长什么样Agent 能不能干活全看工具定义得好不好。我见过太多项目工具定义写得含糊不清模型根本不知道该在什么时候调用结果就是要么不调用要么乱调用。一个好的工具定义我总结下来要满足三个条件名字要动词化且具体。read_file比file_operation好run_shell_command比execute好。模型看到名字就能大致猜到用途。描述要说清楚什么时候用和什么时候不用。光说这个工具用来读文件不够还要说当需要查看文件内容时使用不要用于列出目录。边界越清晰误调用越少。参数要有类型和约束。路径参数要说明是绝对路径还是相对路径字符串参数要说明格式要求。这些约束会体现在 JSON Schema 里模型生成参数时会参考。下面是一个典型的工具定义结构用 Python 字典表示tools [ { name: read_file, description: 读取指定文件的文本内容。当需要查看文件具体内容时使用。不要用于列出目录列出目录请用 list_directory。, parameters: { type: object, properties: { path: { type: string, description: 文件路径相对于当前工作目录 }, max_lines: { type: integer, description: 最多读取的行数默认 200, default: 200 } }, required: [path] } }, { name: run_shell_command, description: 在项目目录下执行 shell 命令并返回输出。用于运行测试、构建、Git 操作等。不要用于需要交互输入的命令。, parameters: { type: object, properties: { command: { type: string, description: 要执行的命令单行不要包含换行符 }, timeout: { type: integer, description: 超时秒数默认 30, default: 30 } }, required: [command] } } ]这里有个细节值得说max_lines和timeout这种参数给个默认值很重要。模型有时候会忘记传可选参数有默认值就不会报错。另外run_shell_command的描述里明确说了不要用于需要交互输入的命令这是为了防止模型调用vim或者top这种会卡住的命令。3.2 上下文管理别让对话历史撑爆窗口Agent 跑久了对话历史会越来越长。每一轮的工具调用结果都往上下文里塞很快就会超出模型的窗口限制。这时候就需要上下文管理策略。我试过几种方案最后觉得最稳的是滑动窗口 摘要压缩的组合。具体做法是保留最近 N 轮完整对话更早的内容压缩成一段摘要。摘要由模型自己生成保留关键决策和结果丢掉冗余的中间过程。def manage_context(messages, max_tokens8000): 管理对话上下文超出限制时压缩早期消息 current_tokens count_tokens(messages) if current_tokens max_tokens: return messages # 保留最近 6 条消息 recent messages[-6:] older messages[:-6] # 把早期消息压缩成摘要 summary_prompt 请用简洁的语言总结以下对话的关键信息和决策\n summary_prompt \n.join([m[content] for m in older if m[role] ! system]) summary call_model(summary_prompt) # 重组上下文 new_messages [messages[0]] # 保留 system prompt new_messages.append({role: system, content: f之前的对话摘要{summary}}) new_messages.extend(recent) return new_messages这个策略的关键在于保留 system prompt。system prompt 里通常包含 Agent 的角色定义和行为约束丢了它 Agent 就会失忆忘记自己是谁、该遵守什么规则。还有一个坑工具调用的结果有时候特别长比如cat一个大文件或者跑测试输出几百行日志。这种内容如果不截断一轮就能把上下文撑爆。我的做法是在工具执行层就做截断超过一定长度只保留头尾中间用省略号代替。注意上下文压缩会丢失信息所以摘要的质量很关键。我建议在摘要 Prompt 里明确要求保留文件路径、命令、错误信息、决策结论这几类内容这些是后续推理最需要的。3.3 并发处理AI Agent 怎么扛住多请求AI Agent 怎么扛并发这个问题最近被问得特别多。我的看法是先搞清楚你的并发场景是什么再谈方案。如果是多个用户同时用同一个 Agent 服务那核心问题是会话隔离。每个用户要有独立的对话历史、独立的工作目录、独立的工具执行环境。这时候用 Python 的asyncio配合每个会话一个Context对象就能搞定。关键是别用全局变量存会话状态否则用户 A 的操作会污染用户 B 的上下文。如果是单个 Agent 要并行执行多个任务比如同时读三个文件、同时跑两个测试那可以用asyncio.gather把多个工具调用并发起来。但要注意不是所有工具都适合并发。写文件的操作如果并发可能会互相覆盖。所以工具定义里最好加一个parallel_safe标记只对读操作开启并发。import asyncio async def execute_tools_parallel(tool_calls): 并发执行多个工具调用只对标记为 parallel_safe 的工具开启并发 parallel_tasks [] serial_tasks [] for call in tool_calls: tool_def get_tool_def(call[name]) if tool_def.get(parallel_safe, False): parallel_tasks.append(execute_tool(call)) else: serial_tasks.append(call) results [] if parallel_tasks: results.extend(await asyncio.gather(*parallel_tasks)) for call in serial_tasks: results.append(await execute_tool(call)) return results如果是高并发场景比如几十上百个请求同时进来那就需要考虑用队列做削峰。把请求丢进asyncio.Queue用固定数量的 worker 去消费。这样既能控制并发度避免把下游 API 打挂又能保证请求不丢失。我实测下来单机用 asyncio 跑 20 到 30 个并发会话是比较稳的再往上就要考虑多进程或者分布式了。但说实话大部分个人项目和小团队场景根本到不了这个量级先把单会话的稳定性做好更重要。4. 实操过程从零搭一个能干的 Agent4.1 环境准备Python 安装与依赖管理先把环境弄干净。我强烈建议用虚拟环境别把系统 Python 搞乱。# 检查 Python 版本建议 3.10 以上 python3 --version # 创建虚拟环境 python3 -m venv agent-env # 激活虚拟环境 source agent-env/bin/activate # Linux/Mac # agent-env\Scripts\activate # Windows # 升级 pip pip install --upgrade pipPython 安装这块如果你是新手去官网下载安装包是最省事的。Windows 上记得勾选Add Python to PATH不然命令行里找不到python命令。Mac 上可以用 Homebrew 装brew install python3.11。Linux 上一般自带版本不够就自己编译或者用包管理器装。依赖方面核心就几个pip install httpx click rich python-dotenvhttpx用来调模型 API支持异步。click做 CLI 参数解析比 argparse 好用。rich做终端输出美化让 Agent 的思考过程看起来清楚点。python-dotenv管理 API Key 这类敏感配置。如果你要用到数据处理再装numpy和pandas。这两个库安装有时候会碰到编译问题Windows 上建议直接下预编译的 wheel 包。pip install numpy一般没问题如果报错就升级 pip 再试。4.2 核心循环Agent 的心跳Agent 的核心就是一个循环接收输入、调用模型、执行工具、把结果喂回模型、再调用模型直到模型不再要求调用工具为止。import json from typing import List, Dict class AgentReach: def __init__(self, model_client, tools, system_prompt, max_iterations10): self.model model_client self.tools tools self.system_prompt system_prompt self.max_iterations max_iterations self.messages [{role: system, content: system_prompt}] async def run(self, user_input: str) - str: self.messages.append({role: user, content: user_input}) for i in range(self.max_iterations): # 调用模型 response await self.model.chat( messagesself.messages, toolsself.tools ) # 没有工具调用直接返回文本 if not response.get(tool_calls): self.messages.append({ role: assistant, content: response[content] }) return response[content] # 有工具调用执行它们 self.messages.append({ role: assistant, content: response.get(content, ), tool_calls: response[tool_calls] }) for call in response[tool_calls]: result await self.execute_tool(call) self.messages.append({ role: tool, tool_call_id: call[id], content: str(result) }) return 达到最大迭代次数任务未完成 async def execute_tool(self, call: Dict) - str: tool_name call[function][name] args json.loads(call[function][arguments]) tool next((t for t in self.tools if t[name] tool_name), None) if not tool: return f错误未知工具 {tool_name} try: return await tool[handler](**args) except Exception as e: return f工具执行失败{str(e)}这个循环有几个关键点。max_iterations是防止 Agent 陷入死循环的保险丝我一般设 10 到 15。工具执行失败时不要把异常直接抛出去而是把错误信息作为工具结果返回给模型让模型自己决定怎么处理。这样 Agent 有一定的自愈能力。4.3 工具实现让 Agent 真的能动手工具的实现要遵循一个原则输入输出都是字符串内部逻辑自己处理。模型只能理解文本所以工具返回的结果要转成文本。import subprocess from pathlib import Path async def read_file(path: str, max_lines: int 200) - str: 读取文件内容 try: file_path Path(path).resolve() # 安全检查限制在工作目录内 if not str(file_path).startswith(str(Path.cwd())): return 错误只能读取当前工作目录下的文件 with open(file_path, r, encodingutf-8) as f: lines f.readlines()[:max_lines] return .join(lines) except FileNotFoundError: return f错误文件 {path} 不存在 except Exception as e: return f读取失败{str(e)} async def run_shell_command(command: str, timeout: int 30) - str: 执行 shell 命令 # 危险命令黑名单 dangerous [rm -rf /, mkfs, dd if, :(){:|:};:] if any(d in command for d in dangerous): return 错误命令被安全策略拦截 try: result subprocess.run( command, shellTrue, capture_outputTrue, textTrue, timeouttimeout, cwdPath.cwd() ) output result.stdout result.stderr # 截断过长输出 if len(output) 5000: output output[:2500] \n...[输出截断]...\n output[-2500:] return output or (命令执行成功无输出) except subprocess.TimeoutExpired: return f错误命令执行超时{timeout}秒 except Exception as e: return f执行失败{str(e)}这里的安全检查很重要。read_file限制在工作目录内防止 Agent 读到系统敏感文件。run_shell_command有危险命令黑名单虽然不能覆盖所有情况但能挡住最明显的误操作。注意shellTrue有命令注入风险。如果 Agent 的参数来自不可信来源建议用shellFalse加参数列表的方式。但在 Agent 场景下命令本身就是模型生成的用shellTrue更灵活。权衡下来加黑名单 限制工作目录是必要的兜底。4.4 CLI 入口把一切串起来最后用 click 做一个 CLI 入口让用户能直接跑起来。import click import asyncio from rich.console import Console from rich.markdown import Markdown console Console() click.command() click.option(--model, defaultgpt-4, help使用的模型) click.option(--workdir, default., help工作目录) click.argument(task, requiredFalse) def main(model, workdir, task): Agent-Reach: 让 AI Agent 在命令行里干活 import os os.chdir(workdir) agent build_agent(model) if task: # 单次任务模式 result asyncio.run(agent.run(task)) console.print(Markdown(result)) else: # 交互模式 console.print([bold green]Agent-Reach 已启动输入 exit 退出[/bold green]) while True: try: user_input console.input([bold blue] [/bold blue]) if user_input.strip().lower() in (exit, quit): break result asyncio.run(agent.run(user_input)) console.print(Markdown(result)) except KeyboardInterrupt: break if __name__ __main__: main()这样就有了两种使用方式agent-reach 帮我看看项目里有哪些 Python 文件直接跑单次任务或者agent-reach进入交互模式连续对话。5. 常见问题与排查技巧实录5.1 模型不调用工具怎么办这是最常见的问题。模型收到任务后直接用自己的知识回答而不是调用工具去查。原因通常是工具描述不够明确或者 system prompt 没有强调优先使用工具。我的解决办法是在 system prompt 里加一段硬性要求你是一个能操作本地环境的 Agent。当任务涉及查看文件、执行命令、获取实时信息时 你必须调用相应的工具不要凭记忆回答。如果你不确定先调用工具确认。另外工具描述里加上当...时使用的触发条件也能提高调用率。5.2 工具调用参数错误模型生成的参数有时候不符合预期比如路径写成绝对路径、命令带换行符。这时候工具执行会失败但错误信息返回给模型后它通常能自己纠正。如果反复错就在工具描述里把参数格式写得更死。我遇到过一个典型情况模型调用run_shell_command时传了多行命令导致 shell 解析出错。后来在描述里明确写单行不要包含换行符问题就少了。5.3 上下文超限对话轮次多了之后API 返回context_length_exceeded错误。这时候就需要前面说的上下文压缩策略。我的经验是与其等到超限再压缩不如每轮结束后主动检查超过阈值就压缩。这样能避免突然报错打断任务。5.4 并发时的会话串扰多个会话同时跑发现 A 的操作影响了 B 的结果。这基本可以确定是用了全局状态。检查一下工具实现里有没有用全局变量存当前目录、当前文件之类的信息。每个会话应该有独立的Context对象所有状态都存在里面。5.5 常见问题速查表问题现象可能原因排查方向解决思路模型不调用工具工具描述模糊检查工具 description补充触发条件和使用场景参数格式错误描述约束不足看模型生成的参数在描述里明确格式要求上下文超限历史消息过长统计 token 数滑动窗口 摘要压缩会话串扰全局状态污染检查全局变量每个会话独立 Context命令执行卡住交互式命令看执行的命令加超时 黑名单输出被截断结果太长看工具返回头尾保留 中间省略API 超时网络或模型慢看请求日志加重试 超时设置5.6 几个我踩过的坑坑一忘了处理工具调用的流式输出。如果模型 API 是流式的工具调用的参数可能分多次返回需要拼接完整后再解析。我一开始没处理导致 JSON 解析失败。坑二工具执行没有超时。有个命令卡住了整个 Agent 就挂在那里。后来给所有工具执行都加了超时超时就返回错误让模型决定下一步。坑三system prompt 太长。我一开始把各种规则都塞进 system prompt结果占了大量 token还让模型抓不住重点。后来精简到只保留核心约束效果反而更好。坑四没有限制工作目录。Agent 跑着跑着跑到系统目录去了读了一堆无关文件。后来强制chdir到指定工作目录并且工具里做路径检查。6. 扩展方向这个项目还能怎么玩Agent-Reach 这类 CLI Agent 的想象空间其实很大。我自己试过几个扩展方向效果还不错。一个是接入定时任务。用 cron 或者 systemd timer 定时触发 Agent让它每天早上检查项目状态、跑一遍测试、生成报告。这就相当于有了一个自动化的项目管家。另一个是接入 Git Hook。在 pre-commit 里调用 Agent 做代码检查在 post-merge 里让它更新文档。这样 Agent 就融入了开发流程不用你主动去叫它。还有一个是多 Agent 协作。一个 Agent 负责规划一个负责执行一个负责审查。它们通过文件或者消息队列通信。这个方向比较复杂但确实能处理更复杂的任务。最后再分享一个小技巧给 Agent 加一个记忆文件。把重要的决策、常用的命令、项目的特殊约定写在一个 markdown 文件里每次启动时读进 system prompt。这样 Agent 就有了长期记忆不用每次从头解释。我在实际使用中发现Agent 的能力上限其实不取决于模型多强而取决于你给它定义了多少好用的工具、设计了多清晰的约束。工具定义得好小模型也能干大事工具定义得烂再强的模型也只能瞎猜。所以与其追新模型不如先把工具层打磨好。
返回列表