Learn Claude Code（一）：从 0 到 1 手写一个 Agent Loop：不到 200 行，跑通模型调工具的全流程

很多人天天聊 Agent但 Agent 核心的循环到底是怎么跑起来的反而不清楚。这篇文章带你手写一个最小可用的 Agent Loop。不依赖 LangChain、不依赖 CrewAI只用 OpenAI SDK 一个 while True跑通模型思考→调用工具→拿到结果→继续推理的完整闭环。01 这篇文章适合谁你知道大模型能调 Function Calling但没亲手搭过完整的 Agent 循环你好奇模型怎么决定什么时候调工具什么时候直接回答你想理解流式输出Stream模式下 Tool Calling 的数据流你想在自己的项目里嵌入一个最简 Agent不想上重型框架不适合谁你要的是生产级 Agent 框架去看 LangGraph、CrewAI你完全不想碰代码02 最终效果是什么运行这个脚本后你会得到一个交互式终端s01 帮我看看当前目录有哪些文件模型收到问题后决定需要执行 Shell 命令输出要运行的命令你会在终端看到$ ls执行命令并拿到结果继续推理基于命令输出给出人类可读的回答流式输出回答内容整个循环完全自动化。你只负责输入问题模型自己决定要不要调工具调什么工具工具返回后怎么继续03 需要准备什么环境要求很低项目说明Python3.9API任意兼容 OpenAI SDK 的大模型 API本文以 DeepSeek 为例依赖openai、python-dotenv安装依赖pipinstallopenai python-dotenv环境变量.env文件DEEPSEEK_API_KEYsk-your-key-hereDEEPSEEK_BASE_URLhttps://api.deepseek.com/v1DEEPSEEK_MODELdeepseek-chat用其他模型也一样换 base_url 和 model 名就行。04 项目结构长什么样整个脚本只有一个文件但我们拆成 5 个模块来理解agent_loop.py ├── ① 自定义输入函数 my_input() # 处理终端输入退格、CtrlC ├── ② 环境与客户端 OpenAI client # API 配置 ├── ③ 工具定义 TOOLS # 告诉模型它能调什么 ├── ④ 工具执行 run_bash() # 真正跑命令的地方 └── ⑤ Agent 循环 agent_loop() # 核心流式推理 自动调工具我们逐个拆开。05 Step 1自定义终端输入为什么需要这个Python 自带的input()函数在终端中不支持退格键Backspace的正确处理——按退格会显示^H而不是删除字符。所以我们自己实现了一个my_input()原理很简单defmy_input(prompt:str)-str:# 把终端设为 raw 模式逐个字符读取# 自己处理回车确认、退格删除、CtrlC 中断# 退出时自动恢复终端设置关键点tty.setraw(fd)让终端进入原始模式每个按键都即时读取遇到\r或\n才结束输入遇到\x7fDEL或\x08BS就删除缓冲区最后一个字符并输出退格序列\b \b用try/finally保证即使出错也能恢复终端这一步不是 Agent 核心但它是让命令行工具好用的工程细节。06 Step 2环境配置与模型客户端这一步很简单但很重要load_dotenv()# 从 .env 加载 API KeyclientOpenAI(api_keyos.getenv(DEEPSEEK_API_KEY),base_urlos.getenv(DEEPSEEK_BASE_URL))MODELos.getenv(DEEPSEEK_MODEL)然后用一个 system prompt 告诉模型它是什么角色SYSTEMfYou are a coding agent at{os.getcwd()}. Use bash to solve tasks. Act, dont explain.这句话很关键——Act, dont explain让模型倾向于直接调工具而不是长篇大论。07 Step 3告诉模型它能用什么工具这一步是 Agent 的能力边界。我们只给模型一个工具——执行 Shell 命令TOOLS[{type:function,function:{name:bash,description:Run a shell command.,parameters:{type:object,properties:{command:{type:string}},required:[command],},},}]一个 Agent 能做多少事取决于你给了它多少工具。这就是 Tool Calling 的起点。08 Step 4安全执行 Shell 命令模型说我要执行这个命令但你不能让它为所欲为。run_bash()做了两件事第一安全过滤# 阻止危险命令dangerous[rm -rf /,sudo ,shutdown ,reboot ,...]ifany(dincommand.lower()fordindangerous):returnError: Dangerous command blocked.甚至分了 Linux 和 Windows 两套危险命令列表。第二执行并返回结果rsubprocess.run(cmd,shellTrue,capture_outputTrue,textTrue,timeout120)out(r.stdoutr.stderr).strip()returnout[:50000]ifoutelseNo output.关键细节timeout120——防止命令卡死输出截断到 50000 字符——防止模型被超长输出冲晕09 Step 5核心——Agent Loop这是整篇文章的灵魂。defagent_loop(messages:list):这个函数只有 50 行左右但它是一个完整的ReAct 循环。循环是怎么跑的整个过程可以拆成四步① 发起推理请求Stream 模式 ↓ ② 流式接收模型输出普通文字 或 Tool Calling 指令 ↓ ③ finish_reason 判断 ├─ tool_calls → 执行工具把结果塞回 messages继续循环 └─ stop → 返回最终答案结束这里重点看一下流式处理 Tool Calling当模型决定调工具时它返回的不是普通 content而是tool_calls增量数据。forchunkinstream:deltachunk.choices[0].delta# 流式输出普通文字ifdelta.content:print(delta.content,end,flushTrue)# 流式累积 Tool Calling 参数ifdelta.tool_calls:# 把多个 chunk 的 tool_calls 片段拼起来tool_call_deltas[idx][function][arguments]tc.function.arguments这就是流式 Function Calling——模型一边想参数你一边收到片段最后拼成完整的 JSON。判断循环方向iffinish_reason!tool_calls:return# 模型说够了结束# 否则执行工具继续循环fortcinassistant_msg.get(tool_calls,[]):argsjson.loads(tc[function][arguments])outputrun_bash(args[command])messages.append({role:tool,tool_call_id:tc[id],content:output})核心逻辑就是只要模型返回finish_reasontool_calls就执行工具、塞回结果、继续循环。10 Step 6把一切串起来主函数做的事情很纯粹history[]whileTrue:querymy_input(s01 )ifquery.strip().lower()in(q,exit,):breakhistory.append({role:user,content:query})agent_loop(history)用户输入问题加入消息历史启动 Agent Loop循环结束后打印模型的最终回答等待下一个问题11 运行和测试启动后你会看到s01 当前目录是什么模型会调用bash工具执行pwd你会在终端看到黄色高亮的$ pwd得到输出后继续推理告诉你当前目录流式输出最终答案你还可以试试s01 帮我创建一个 test.txt 文件里面写入 Hello Agent Loop模型会自动完成echo Hello Agent Loop test.txt你看到的是黄色的$ echo Hello Agent Loop test.txt在执行然后模型告诉你已创建。12 完整代码以下是agent_loop.py的完整代码可以直接复制保存运行importosimportsysimportatexitimportsubprocess# ── 自定义输入函数彻底避免退格问题 ──defmy_input(prompt:str)-str: 替代 builtins.input()自己处理字符回显和退格 不依赖 readline / termios 驱动层配置。 importtermiosimporttty# 不是 TTY 就回退到标准 inputifnotsys.stdin.isatty():returninput(prompt)sys.stdout.write(prompt)sys.stdout.flush()fdsys.stdin.fileno()oldtermios.tcgetattr(fd)buf:list[str][]try:tty.setraw(fd)whileTrue:chsys.stdin.read(1)# 回车 → 结束ifchin(\r,\n):sys.stdout.write(\r\n)break# CtrlC → KeyboardInterruptifch\x03:raiseKeyboardInterrupt# CtrlD → EOFErrorifch\x04:raiseEOFError# 退格DEL (127) 或 BS (8)iford(ch)in(127,8):ifbuf:buf.pop()sys.stdout.write(\b \b)# 常规字符else:buf.append(ch)sys.stdout.write(ch)sys.stdout.flush()finally:termios.tcsetattr(fd,termios.TCSADRAIN,old)return.join(buf)# 确保退出时恢复终端def_cleanup_terminal():try:importtermios fdsys.stdin.fileno()# 什么都不做只是确保异常退出时不会留下 raw 终端exceptException:passatexit.register(_cleanup_terminal)fromopenaiimportOpenAIfromdotenvimportload_dotenv load_dotenv()clientOpenAI(api_keyos.getenv(DEEPSEEK_API_KEY),base_urlos.getenv(DEEPSEEK_BASE_URL))MODELos.getenv(DEEPSEEK_MODEL)SYSTEMfYou are a coding agent at{os.getcwd()}. Use bash to solve tasks. Act, dont explain.TOOLS[{type:function,function:{name:bash,description:Run a shell command.,parameters:{type:object,properties:{command:{type:string}},required:[command],},},}]defrun_bash(command:str)-str:# OS-aware dangerous command patternsdangerous[rm -rf /,sudo ,shutdown,reboot, /dev/sd,dd if,mkfs,:(){,chmod 777 /,poweroff,halt,init 0,init 6,]ifos.nament:dangerous[del /f /s,rd /s /q,format ,diskpart,shutdown,reg delete,net user,taskkill /f,]ifany(dincommand.lower()fordindangerous):returnError: Dangerous command blocked.try:ifos.nament:shellpowershell.exe# Force UTF-8 so Chinese characters dont get garbledcmd($OutputEncoding [Console]::OutputEncoding [System.Text.Encoding]::UTF8; command)else:shell/bin/bashcmdcommand rsubprocess.run(cmd,shellTrue,executableshell,cwdos.getcwd(),capture_outputTrue,textTrue,encodingutf-8,timeout120,)out(r.stdoutr.stderr).strip()returnout[:50000]ifoutelseNo output.exceptsubprocess.TimeoutExpired:returnError: Command timed out (120s).except(FileNotFoundError,OSError)ase:returnfError:{str(e)}defagent_loop(messages:list):importjsonwhileTrue:content_buf[]tool_call_deltas{}# index - accumulated dictstreamclient.chat.completions.create(modelMODEL,messagesmessages,toolsTOOLS,tool_choiceauto,streamTrue,temperature0.2,extra_body{thinking:{type:enabled}},)finish_reasonNoneforchunkinstream:ifnotchunk.choices:continuedeltachunk.choices[0].delta finish_reasonchunk.choices[0].finish_reason# Stream text content token by tokenifdelta.content:content_buf.append(delta.content)print(delta.content,end,flushTrue)# Accumulate tool call deltas (function calling metadata)ifdelta.tool_calls:fortcindelta.tool_calls:idxtc.indexifidxnotintool_call_deltas:tool_call_deltas[idx]{id:,function:{name:,arguments:}}iftc.id:tool_call_deltas[idx][id]tc.idiftc.function:iftc.function.name:tool_call_deltas[idx][function][name]tc.function.nameiftc.function.arguments:tool_call_deltas[idx][function][arguments]tc.function.argumentsprint()# newline after stream# Reconstruct full assistant message from accumulated deltasfull_content.join(content_buf)ifcontent_bufelseNoneassistant_msg{role:assistant,content:full_content}iftool_call_deltas:assistant_msg[tool_calls][{id:v[id],type:function,function:{name:v[function][name],arguments:v[function][arguments],},}for_,vinsorted(tool_call_deltas.items())]messages.append(assistant_msg)iffinish_reason!tool_calls:return# Execute each tool callresults[]fortcinassistant_msg.get(tool_calls,[]):argsjson.loads(tc[function][arguments])commandargs[command]print(f\033[33m${command}\033[0m)outputrun_bash(command)print(output[:200])results.append({role:tool,tool_call_id:tc[id],content:output,})messages.extend(results)if__name____main__:print(s01: Agent Loop)print(输入问题回车发送。输入 q 退出。\n)history[]whileTrue:try:querymy_input(\033[36ms01 \033[0m)except(EOFError,KeyboardInterrupt):breakifquery.strip().lower()in(q,exit,):breakhistory.append({role:user,content:query})agent_loop(history)# Print the models final text responseresponse_contenthistory[-1][content]ifisinstance(response_content,list):forblockinresponse_content:ifgetattr(block,type,None)text:print(block.text)print()保存为agent_loop.py然后在终端执行python agent_loop.py记得在同一个目录下创建.env文件并写入你的 API Key。13 常见问题和排查问题排查方向模型不调工具检查tool_choiceauto是否设置system prompt 是否包含 “use bash”流式输出乱码确认终端编码为 UTF-8Windows 需设置$OutputEncoding代码已处理退格键不正常确保my_input正常生效终端支持 raw 模式API 报错检查.env文件中的 Key 和 Base URL命令超时timeout120可调整或检查命令是否阻塞14 还能怎么优化这个 Agent Loop 是最简原型你可以沿这些方向升级1. 增加更多工具不只是 bash可以加文件读写、网络请求、代码执行沙箱等。2. 持久化对话历史把 messages 存到文件或数据库让 Agent 有长期记忆。3. 支持多轮工具链当前是一轮调一个工具。可以升级成模型连续调多个工具再综合结果回答。4. 更好的输出体验用 rich 库做更漂亮的终端输出或者接入 Web UI。5. 错误重试机制工具执行失败时让模型尝试修正参数重新调用。15 总结这篇文章的核心就一句话Agent Loop 的本质是模型推理 → 判断要不要调工具 → 调工具 → 把结果给模型继续推理 → 循环直到模型说够了。整个实现只有三个关键代码段TOOLS定义——告诉模型它能做什么agent_loop()中的finish_reason判断——决定循环还是结束run_bash()——真正执行工具并返回结果200 行代码一个完整的 Agent 就活了。当然它不是生产级的但它帮你拆掉了 Agent 最神秘的那层窗户纸。下次再有人问Agent 到底怎么循环的你可以说就是一个 while True判断 finish_reason 是不是 tool_calls。