S01: Agent Loop — 一个循环就够了

问题：

当我们提出一个问题给大模型：“帮我读取我当前目录有哪些文件，并执行start.bat”
模型能输出一条bash命令，但输出完就停了，它不会去跑这条命令，也看不到执行后的结果再继续推理
这些都需要用户自己手动跑，并将结果复制给大模型，让它接着干。等下一个命令出来，再跑下一个命令，再贴回去
每个来回，都由人来做中间层，把人从中解放出来就是，我们接下来要做的

解决方案：

使用一个while True 循环，模型调用工具就继续，不调用工具就停。循环通过检查响应块中内进行判断

工作原理

将方案转换成代码，分步来看看，简单写个伪代码演示一下：
第1步： 把用户问题加入到消息队列中

messages=[{"role":"user","content":query}]

第2步：将消息和工具定义一起发给LLM

response=client.chat.completions.create(model="deepseek-flash",messages=messages,#type: ignoretools=TOOLS,max_tokens=8000)

第3步：追加模型回答，检查它是否调用了工具。没调->结束

messages.append({"role":"assistant","content":response.content})tool_calls=[blockforblockinresponse.contentifblock.type=="tool_use"]ifnottool_calls:return

只有实际存在tool_calls block才会进入执行阶段，因此不会追加空的工具结果消息

第4步：执行模型要求工具，收集结果

results=[]forblockintool_calls:output=run_bash(block.input["command"])results.append({"type":"tool_result","tool_use_id":block.id,"content":output})

第5步：把工具结果作为新新消息追加，回到第2步

messages.append("role":"user","content":results})

组装成一个完整函数：

defagent_loop(messages):whileTrue:response=client.chat.completions.create(model="deepseek-flash",messages=messages,#type: ignoretools=TOOLS,max_tokens=8000)messages.append({"role":"assistant","content":response.content})tool_calls=[blockforblockinresponse.contentifblock.type=="tool_use"]ifnottool_calls:returnresults=[]forblockintool_calls:output=run_bash(block.input["command"])results.append({"type":"tool_result","tool_use_id":block.id,"content":output})messages.append("role":"user","content":results})

三十多行，这就是最小可运行的agent harness内核。它为模型提供持续行动的最小运行框架：模型负责决策，harness负责执行。后面的16个章节也都说在这个循环上叠加的，循环始终不变

试一试

这是整个代码，我把原cc格式改成openai格式

#!/usr/bin/env python3""" s01_agent_loop.py - The Agent Loop (OpenAI format) while True: response = LLM(messages, tools) if response contains no tool_calls: break execute tools append tool results Usage: pip install openai python-dotenv DEEPSEEK_URL=... DEEPSEEK_API_KEY=... MODEL_ID=... python s01_agent_loop/code.py """importjsonimportosimportsubprocesstry:importreadline readline.parse_and_bind('set bind-tty-special-chars off')readline.parse_and_bind('set input-meta on')readline.parse_and_bind('set output-meta on')readline.parse_and_bind('set convert-meta off')exceptImportError:passfromopenaiimportOpenAIfromdotenvimportload_dotenv load_dotenv(override=True)client=OpenAI(base_url=os.getenv("DEEPSEEK_URL"),api_key=os.getenv("DEEPSEEK_API_KEY"))MODEL=os.environ["MODEL_ID"]ENVIRONMENT_PROMPT=("Windows: the bash tool runs through cmd.exe; use cmd.exe syntax, not Unix ""Bash or PowerShell syntax"ifos.name=="nt"else"Unix-like: the bash tool runs the system shell")SYSTEM=(f"You are a coding agent at{os.getcwd()}. Environment:{ENVIRONMENT_PROMPT}. ""Use bash to solve tasks. Act, don't explain.")# -- Tool definition: OpenAI function-calling format --TOOLS=[{"type":"function","function":{"name":"bash","description":"Run a shell command.","parameters":{"type":"object","properties":{"command":{"type":"string"}},"required":["command"],},},}]# -- Tool execution (unchanged) --defrun_bash(command:str)->str:dangerous=["rm -rf /","sudo","shutdown","reboot","> /dev/"]ifany(dincommandfordindangerous):return"Error: Dangerous command blocked"try:r=subprocess.run(command,shell=True,cwd=os.getcwd(),capture_output=True,text=True,errors="replace",timeout=120)out=(r.stdout+r.stderr).strip()returnout[:50000]ifoutelse"(no output)"exceptsubprocess.TimeoutExpired:return"Error: Timeout (120s)"except(FileNotFoundError,OSError)ase:returnf"Error:{e}"# -- The core loop: call tools until the model stops --defagent_loop(messages:list):whileTrue:response=client.chat.completions.create(model=MODEL,messages=messages,tools=TOOLS,max_tokens=8000,)msg=response.choices[0].message# 1) 回填 assistant 消息（如果有 tool_calls 必须一并回填）assistant_msg={"role":"assistant","content":msg.content}ifmsg.tool_calls:assistant_msg["tool_calls"]=[{"id":c.id,"type":"function","function":{"name":c.function.name,"arguments":c.function.arguments,# JSON 字符串，原样回填},}forcinmsg.tool_calls]messages.append(assistant_msg)# 2) 没有工具调用 → 结束ifnotmsg.tool_calls:return# 3) 执行每个工具调用，并把结果以 role="tool" 回填forcallinmsg.tool_calls:try:args=json.loads(call.function.argumentsor"{}")exceptjson.JSONDecodeError:args={}command=args.get("command","")print(f"\033[33m${command}\033[0m")output=run_bash(command)print(output[:200])messages.append({"role":"tool","tool_call_id":call.id,"content":output,})# -- Entry point --if__name__=="__main__":print("s01: Agent Loop")print("Enter a question, press Enter to send. Type q to quit.\n")history=[{"role":"system","content":SYSTEM}]whileTrue:try:query=input("\001\033[36m\002s01 >> \001\033[0m\002")except(EOFError,KeyboardInterrupt):breakifquery.strip().lower()in("q","exit",""):breakhistory.append({"role":"user","content":query})agent_loop(history)# agent_loop 正常返回时，最后一条一定是没有 tool_calls 的 assistant 消息final=history[-1]iffinal.get("role")=="assistant"andfinal.get("content"):print(final["content"])print()

试试这些 prompt：

1. Create a file called hello.py that prints "Hello, World!"
2. List all Python files in this directory
3. What is the current git branch?

观察重点：模型什么时候调用工具（循环继续），什么时候不调用（循环结束）？

代码源码地址：https://github.com/shareAI-lab/learn-claude-code
蓝奏云地址：https://wwbfs.lanzouv.com/iva494dg0h4f
密码:cc39