30行代码构建智能体核心循环:从零实现AI自主执行Bash命令

📅 2026/8/21 11:07:45
30行代码构建智能体核心循环:从零实现AI自主执行Bash命令
这次我们来看一个面向开发者的智能体入门项目核心是理解智能体如何通过“思考-行动”循环来自主完成任务。项目标题点明了关键一个智能体的核心循环其内核代码可能只有30行核心模式就是while True循环配合Bash命令执行。这听起来很抽象但它的价值在于提供了一个极简、可运行的起点让你能亲手搭建一个能“自己跑命令接着干”的智能体。对于想入门智能体开发的开发者来说最大的障碍往往不是理论而是不知道如何动手。这个项目直接切入实践用最少的代码展示智能体如何与环境比如你的终端交互如何解析任务如何执行命令并根据结果决定下一步行动。它不依赖复杂的框架而是聚焦于最本质的循环逻辑。本文将带你从零开始理解这个核心循环并基于 Claude Code或类似的大模型代码解释器搭建一个可运行的智能体原型验证其从接收任务到执行命令的完整流程。1. 核心能力速览能力项说明项目类型智能体开发入门实践 / 核心循环原型核心思想实现智能体的“感知-思考-行动”循环重点是行动执行通过 Bash与结果反馈技术栈Python主循环、Bash命令执行、大模型 API如 Claude Code、DeepSeek 等用于推理硬件门槛无特殊要求普通开发机即可。主要消耗在于调用大模型 API 的 Token 费用。启动方式命令行直接运行 Python 脚本关键依赖1. Python 3.8 环境2. 对应大模型的 API Key 和 SDK3. 操作系统支持 BashLinux/macOS 原生Windows 需 Git Bash 或 WSL是否支持 API是本项目本身需要调用大模型的 API如 Anthropic Claude、DeepSeek 等。是否支持批量任务是核心循环逻辑天然支持处理任务队列只需在外层封装即可。适合场景1. 学习智能体基础工作原理2. 快速验证智能体处理简单命令行任务的想法3. 作为更复杂智能体项目如自动化运维、代码生成助手的起点2. 适用场景与使用边界这个极简智能体原型最适合以下几类开发者智能体开发初学者希望绕过复杂框架直接理解智能体最核心的循环与控制流。任务自动化探索者有一些重复性的命令行操作如文件整理、日志分析、简单部署想尝试用自然语言驱动。原型验证者有一个关于智能体应用的创意需要快速构建一个可交互的“概念验证”PoC来演示核心流程。它能解决什么问题概念具象化将“智能体”这个抽象概念转化为一段可运行、可观察的代码。环境交互验证证明智能体可以通过生成并执行 Bash 命令来操作本地文件系统、运行程序、查询系统状态。任务分解演示对于“帮我找出当前目录下最大的5个文件”这类任务智能体可以分解为ls -l,sort,head等命令的组合并执行。它不适合什么场景生产环境直接使用这是一个教学原型缺乏错误恢复、安全沙箱、权限控制、成本监控等生产级功能。复杂图形界面操作核心是 Bash 命令无法直接操作 GUI 应用。高精度或高风险操作如直接操作数据库、删除关键文件、执行rm -rf /等危险命令。绝对禁止在未加严格限制的情况下让智能体执行此类操作。安全与合规边界权限最小化务必在沙箱环境或严格限制的目录下运行避免智能体拥有过高系统权限。命令过滤必须对智能体生成的命令进行安全检查过滤掉rm、dd、chmod、wget从不明源下载等高风险指令。监督介入原型阶段建议采用“人类在环”模式即智能体每执行一个命令或一系列命令前需经人工确认。数据隐私发送给大模型 API 的上下文可能包含系统信息注意避免泄露敏感数据。3. 环境准备与前置条件在开始编写和运行智能体之前需要确保你的开发环境满足以下条件操作系统推荐 Linux 或 macOS因为它们原生支持完整的 Bash 环境。Windows 用户可以使用Git Bash或WSL (Windows Subsystem for Linux)来获得兼容的 Bash 环境。Python 环境需要 Python 3.8 或更高版本。建议使用venv或conda创建独立的虚拟环境。Bash 终端确保你有一个可用的终端并且bash命令可以正常执行。大模型 API 访问权限这是智能体的“大脑”。你需要准备以下之一Anthropic Claude API Key如果你打算使用 Claude Code 模型。DeepSeek API Key或其他支持代码生成与推理的模型 API。OpenAI API Key使用gpt-4或gpt-4o的代码解释器能力。网络连接能够稳定访问你所选大模型的 API 端点。文本编辑器或 IDE如 VSCode、PyCharm 等用于编写代码。通用检查清单# 1. 检查 Python 版本 python3 --version # 2. 检查 Bash 版本 bash --version # 3. 创建一个新的项目目录并进入 mkdir simple_agent cd simple_agent # 4. 创建并激活 Python 虚拟环境 (以 venv 为例) python3 -m venv .venv # Linux/macOS source .venv/bin/activate # Windows (Git Bash) source .venv/Scripts/activate # 激活后命令行提示符前应显示 (.venv)4. 安装依赖与项目初始化智能体循环本身不复杂但需要安装与大模型 API 交互的 SDK。这里以使用Anthropic Claude API为例。安装必要的 Python 包# 确保虚拟环境已激活 pip install anthropic # Claude官方SDK pip install openai # 如果你打算用OpenAI的模型 # 其他工具包按需安装 pip install requests设置 API Key 环境变量为了安全不要将 API Key 硬编码在代码中。# Linux/macOS export ANTHROPIC_API_KEY你的-claude-api-key # Windows (PowerShell) $env:ANTHROPIC_API_KEY你的-claude-api-key你也可以将环境变量配置在.env文件中使用python-dotenv包加载。验证 API 连通性可选 创建一个简单的测试脚本test_api.pyimport anthropic import os client anthropic.Anthropic(api_keyos.environ.get(ANTHROPIC_API_KEY)) # 发送一个简单的消息测试 message client.messages.create( modelclaude-3-5-sonnet-20241022, # 或你使用的模型 max_tokens100, messages[{role: user, content: Hello, world!}] ) print(message.content[0].text)运行python test_api.py如果能看到模型回复说明 API 配置成功。5. 构建核心智能体循环现在我们来编写那“30行”左右的核心循环。这个循环的伪代码如下初始化对话历史 while True: 1. 将当前任务、历史、系统状态组合成提示词 (Prompt) 2. 调用大模型请求生成下一步的“思考”和“要执行的命令” 3. 解析模型响应提取命令 4. 如果命令是退出指令则结束循环 5. 否则在安全的上下文中执行该 Bash 命令 6. 捕获命令执行后的输出和错误 7. 将命令和结果追加到对话历史中作为下一轮循环的上下文以下是具体的 Python 实现simple_agent.pyimport subprocess import os import anthropic import sys class SimpleBashAgent: def __init__(self, api_keyNone, modelclaude-3-5-sonnet-20241022): 初始化智能体设置API客户端和模型。 self.api_key api_key or os.environ.get(ANTHROPIC_API_KEY) if not self.api_key: raise ValueError(请设置 ANTHROPIC_API_KEY 环境变量或传入 api_key 参数。) self.client anthropic.Anthropic(api_keyself.api_key) self.model model self.conversation_history [] # 保存对话历史 def execute_bash_command(self, command): 安全地执行 Bash 命令并返回输出。 try: # 注意这里没有做命令过滤生产环境必须添加 result subprocess.run( command, shellTrue, capture_outputTrue, textTrue, cwdos.getcwd(), # 在当前工作目录执行 timeout30 # 设置超时防止卡死 ) output result.stdout if result.stderr: output f\n[STDERR]: {result.stderr} return output.strip(), result.returncode except subprocess.TimeoutExpired: return 命令执行超时30秒。, -1 except Exception as e: return f命令执行出错: {str(e)}, -1 def run_agent_loop(self, initial_task): 运行智能体的主循环。 system_prompt 你是一个擅长使用 Bash 命令完成任务的助手。用户会给你一个任务你需要一步步思考并生成对应的 Bash 命令来执行。 你只能生成一条可执行的 Bash 命令。在输出命令前先简要说明你的思考。 命令执行后你会看到命令的输出结果。根据结果你再决定下一步是生成新命令还是报告任务完成。 如果你认为任务已完成就生成一条特殊的命令AGENT_STOP。 输出格式严格遵循 思考你的思考过程 命令要执行的 bash 命令 或 AGENT_STOP self.conversation_history.append({role: user, content: initial_task}) print(f任务: {initial_task}) print(- * 50) step 0 while True: step 1 print(f\n[步骤 {step}]) # 1. 构建提示词 prompt_messages [{role: system, content: system_prompt}] for msg in self.conversation_history[-6:]: # 只保留最近几轮历史防止上下文过长 prompt_messages.append(msg) # 2. 调用大模型 try: response self.client.messages.create( modelself.model, max_tokens500, messagesprompt_messages ) response_text response.content[0].text except Exception as e: print(f调用模型 API 失败: {e}) break print(fAI 响应:\n{response_text}) # 3. 解析响应提取命令 command_to_run None lines response_text.split(\n) for line in lines: if line.startswith(命令): command_to_run line.replace(命令, ).strip() break if not command_to_run: print(未能在响应中找到命令。) self.conversation_history.append({role: assistant, content: response_text}) continue # 4. 检查退出条件 if command_to_run AGENT_STOP: print(智能体认为任务已完成。) break # 5. 执行命令 print(f执行命令: {command_to_run}) output, return_code self.execute_bash_command(command_to_run) print(f命令输出:\n{output}) print(f返回码: {return_code}) # 6. 保存历史 self.conversation_history.append({role: assistant, content: response_text}) # 将命令执行结果作为下一轮的“用户”输入 result_message f上一条命令 {command_to_run} 的执行结果\n{output} self.conversation_history.append({role: user, content: result_message}) # 简单防循环机制 if step 15: print(步骤过多为防止无限循环自动停止。) break print(\n *50) print(智能体循环结束。) if __name__ __main__: # 示例让智能体统计当前目录的文件信息 initial_task 请帮我统计当前工作目录下有多少个文件和子目录并列出最大的3个文件。 agent SimpleBashAgent() agent.run_agent_loop(initial_task)6. 功能测试与效果验证现在让我们运行这个智能体看它如何完成任务。测试 1基础文件系统操作测试目的验证智能体能否理解任务并生成正确的ls、du、find等命令。操作步骤确保在项目目录simple_agent下。运行脚本python simple_agent.py。观察控制台输出。预期结果与过程分析 智能体接收任务后可能会产生类似以下的思考与行动序列任务请帮我统计当前工作目录下有多少个文件和子目录并列出最大的3个文件。 -------------------------------------------------- [步骤 1] AI 响应 思考首先我需要查看当前目录下有什么。使用 ls -la 可以列出所有文件和目录并显示详细信息。 命令ls -la 执行命令ls -la 命令输出 total 24 drwxr-xr-x 4 user staff 128 Mar 20 10:00 . drwxr-xr-x 5 user staff 160 Mar 20 09:55 .. -rw-r--r-- 1 user staff 1234 Mar 20 10:00 simple_agent.py -rw-r--r-- 1 user staff 567 Mar 20 09:58 test_api.py drwxr-xr-x 2 user staff 64 Mar 20 10:01 data ... 返回码0智能体看到ls的输出后会将其作为上下文输入下一轮。[步骤 2] AI 响应 思考现在我需要统计文件和目录的数量。ls -la 的输出中以 - 开头的是文件以 d 开头的是目录。我可以使用 grep 和 wc 来计数。 命令ls -la | grep ^- | wc -l 执行命令ls -la | grep ^- | wc -l 命令输出 8 返回码0[步骤 3] AI 响应 思考统计目录数。 命令ls -la | grep ^d | wc -l 执行命令ls -la | grep ^d | wc -l 命令输出 3 返回码0[步骤 4] AI 响应 思考现在需要找出最大的3个文件。可以使用 du 命令查看文件大小并按大小排序。 命令du -ah . | grep -v /\. | sort -rh | head -4 执行命令du -ah . | grep -v /\. | head -4 命令输出 2.5M ./data/large_file.log 1.2M ./simple_agent.py 876K ./venv/lib/python3.9/site-packages/some_package.xyz ... 返回码0[步骤 5] AI 响应 思考我已经获取了文件数、目录数和最大的几个文件信息。任务完成。 命令AGENT_STOP 智能体认为任务已完成。判断成功标准智能体通过多轮交互最终生成了AGENT_STOP命令并且中间生成的命令逻辑正确成功获取了所需信息。测试 2依赖前序结果的复杂任务测试目的验证智能体能否利用历史上下文执行需要多步且有依赖关系的任务。输入任务“请先创建一个名为test_output的目录然后在其中生成一个包含当前日期时间的文本文件。”预期过程智能体应首先生成mkdir test_output。执行成功后在下一轮它需要结合“已在test_output目录下”的上下文生成创建文件的命令如date test_output/current_time.txt或echo $(date) test_output/current_time.txt。最后生成AGENT_STOP。失败排查模型未理解上下文如果第二步它仍然在根目录创建文件说明历史上下文传递可能有问题或者模型未充分关注历史。可以尝试优化提示词强调“基于上一步的结果”。命令执行失败如果mkdir失败目录已存在智能体应能根据错误输出File exists调整策略例如跳过或询问。我们的简单原型可能处理不了这指出了改进方向。测试 3错误处理与边界测试目的观察智能体在遇到错误命令时的行为。可以手动模拟在execute_bash_command函数执行后如果返回码非0观察智能体下一轮的思考。一个健壮的智能体应该尝试分析错误并修正命令。我们的基础版本可能只会将错误信息作为上下文模型可能会尝试提出新方案也可能陷入困惑。7. 接口 API 与批量任务集成当前原型是交互式循环。我们可以很容易地将其改造成一个可编程的API 服务或批量任务处理器。7.1 封装为函数供外部调用将核心逻辑封装接受一个任务字符串返回执行过程日志和最终结果。def run_agent_task(task_description, max_steps10): 执行单个任务返回执行日志。 agent SimpleBashAgent() logs [] # 这里需要重写 run_agent_loop 以收集日志而不是直接打印 # ... 重写逻辑将 print 改为 logs.append(...) # 最终返回 logs 和最终状态完成/失败/超时 return logs7.2 构建简单的 Flask API 服务创建一个agent_api.py文件提供 HTTP 接口。from flask import Flask, request, jsonify import threading import queue app Flask(__name__) task_queue queue.Queue() results {} app.route(/submit_task, methods[POST]) def submit_task(): 提交一个任务返回任务ID。 data request.json task data.get(task) if not task: return jsonify({error: No task provided}), 400 task_id ftask_{len(results)} task_queue.put((task_id, task)) results[task_id] {status: pending, logs: []} return jsonify({task_id: task_id}) def worker(): 后台工作线程从队列中取任务执行。 while True: task_id, task task_queue.get() results[task_id][status] running # 调用智能体执行任务并捕获日志 # logs run_agent_task(task) # results[task_id][logs] logs # results[task_id][status] completed task_queue.task_done() # 启动后台线程 threading.Thread(targetworker, daemonTrue).start() app.route(/get_result/task_id, methods[GET]) def get_result(task_id): 根据任务ID获取执行结果。 if task_id not in results: return jsonify({error: Task not found}), 404 return jsonify(results[task_id]) if __name__ __main__: app.run(host127.0.0.1, port5000, debugFalse)启动服务python agent_api.py。然后可以使用curl或 Pythonrequests库提交任务。curl -X POST http://127.0.0.1:5000/submit_task \ -H Content-Type: application/json \ -d {task: 统计 /tmp 目录下文件数量}7.3 批量任务处理批量任务的核心是维护一个任务队列。上面的 Flask 示例已经包含了队列的雏形。在实际应用中你需要考虑并发控制同时运行多少个智能体实例状态持久化任务结果需要保存到数据库或文件。错误重试任务失败后是否重试、重试几次。资源隔离每个任务应在独立的临时目录中执行避免相互干扰。8. 资源占用与性能观察这个原型项目的资源消耗主要分为两部分大模型 API 调用成本消耗 API Token产生费用。每次循环交互都会消耗输入和输出的 Token。延迟网络往返时间RTT加上模型推理时间是循环中主要的耗时环节。一次简单的任务可能需要 3-10 秒的 API 调用时间。观察方法监控 API 调用次数和 Token 使用量。大多数云服务商的控制台提供用量监控。本地 Python 进程与 Bash 子进程CPU/内存占用极低几乎可以忽略不计。磁盘 I/O取决于智能体执行的命令。如果命令涉及大量文件读写则会产生 I/O。观察方法使用系统工具如top、htopLinux/macOS或任务管理器Windows。性能优化方向提示词优化精简system_prompt和conversation_history减少不必要的 Token 消耗。缓存策略对于相同的子任务可以缓存模型的响应。异步调用如果处理批量任务可以使用异步 HTTP 客户端来并发调用 API减少等待时间。本地模型如果条件允许使用本地部署的小型代码模型如 DeepSeek Coder 的本地版本可以消除网络延迟和 API 成本但需要 GPU 资源。9. 常见问题与排查方法在开发和运行此类智能体时你会遇到一些典型问题。问题现象可能原因排查方式解决方案启动时报错ModuleNotFoundError: No module named anthropicPython 依赖未安装或虚拟环境未激活。检查当前终端前缀是否有(.venv)运行pip list | grep anthropic。激活虚拟环境source .venv/bin/activate然后安装pip install anthropic。运行时报错ValueError: 请设置 ANTHROPIC_API_KEY...API Key 未正确设置。在终端执行echo $ANTHROPIC_API_KEY(Linux/macOS) 或echo %ANTHROPIC_API_KEY%(Windows CMD) 检查。正确设置环境变量或修改代码直接传入api_key参数。模型 API 调用返回 401 或 403 错误API Key 无效、过期或没有对应模型的权限。查看 SDK 返回的错误信息。去对应平台检查 API Key 状态和额度。更换有效的 API Key或检查是否订阅了对应模型如 Claude Code。智能体陷入无限循环不断生成无意义的命令提示词设计有缺陷或模型无法从上下文中理解任务已结束。查看循环中的“思考”部分看模型是否在重复逻辑。1. 强化system_prompt中关于AGENT_STOP的说明。2. 在代码中设置最大步数限制我们代码中已有step 15。3. 增加对重复命令的检测。Bash 命令执行失败返回command not found智能体生成的命令在当前环境中不存在如 Windows 下生成ls但未装 Git Bash。检查命令输出中的错误信息。1. 在system_prompt中明确环境限制如“你运行在 Git Bash 中”。2. 在execute_bash_command中增加命令白名单过滤。执行危险命令如rm -rf /没有对生成的命令进行安全检查。这是严重的安全漏洞必须在execute_bash_command函数中添加命令黑名单或白名单逻辑。例如pythonbrdangerous_keywords [‘rm -rf’, ‘dd’, ‘mkfs’, ‘:(){:|:};:’] # 黑洞命令等brfor kw in dangerous_keywords:br if kw in command:br return f“拒绝执行危险命令: {command}”, -1br任务涉及文件操作但智能体找不到文件工作目录cwd设置不正确。打印os.getcwd()查看当前目录。在execute_bash_command中可以根据任务需要动态改变cwd参数或让智能体在命令中使用绝对路径。API 调用速度慢任务执行时间长网络延迟或模型响应慢。使用time命令测量单次 API 调用耗时。1. 考虑使用响应更快的模型如 Claude Haiku。2. 实现超时和重试机制。3. 对于复杂任务尝试让模型一次性生成多步计划减少交互轮数。10. 最佳实践与使用建议基于这个原型进行开发或实验时请遵循以下建议安全第一沙箱运行永远不要在拥有重要数据的生产环境或个人主目录直接运行。使用 Docker 容器或虚拟机创建一个隔离的沙箱环境。在代码中实现严格的命令过滤器只允许执行明确安全的命令集。从简到繁逐步迭代先用“列出文件”这种无害命令测试循环是否跑通。再测试“创建文件/目录”等有副作用的操作。最后再尝试更复杂的、需要多步推理的任务。优化提示词Prompt Engineeringsystem_prompt是智能体的“宪法”定义它的角色、能力和约束。多花时间打磨它。明确输出格式如“思考... 命令...”便于代码解析。在提示词中注入环境信息如操作系统、可用工具列表。完善日志与监控不要只依赖print将完整的交互历史用户消息、模型响应、命令、输出、返回码记录到文件或数据库中。这对于调试和后续分析至关重要。设计任务边界明确告诉智能体任务的边界。例如“你的操作范围仅限于/home/agent_workspace目录”。对于开放式任务如“优化我的网站”智能体很可能无法完成。应拆解为具体、可验证的子任务如“检查Nginx配置语法”、“查找访问日志中的错误”。成本控制在 API 平台设置用量告警和月度限额。在代码中估算 Token 消耗对于长上下文任务要格外小心。这个30行的智能体循环就像一台发动机的核心部件。它本身不能直接上路但为你理解智能体如何“动起来”提供了最清晰的蓝图。你的下一步可以是为它装上“安全护栏”命令过滤、增加“导航系统”更复杂的任务规划、连接“更多工具”不止Bash还能调用Web API、数据库最终打造出一个真正能解决实际问题的智能体应用。建议从改造这个核心循环开始亲手添加每一个功能你会对智能体开发有更深刻的理解。