项目教程原地址https://github.com/pigeon-dove/anthony-agent第一章Agent Loop — ReAct 循环的核心整个项目的核心是一个不到 50 行的 while 循环。理解了它后面所有章节都是在这个骨架上叠加功能。什么是 Agent一个普通的 LLM 调用是单轮的你发消息模型回复结束。Agent 不一样。它是一个循环模型不仅能回复文字还能决定我需要调用一个工具来获取信息然后拿到工具结果后继续思考直到它认为任务完成。这就是ReActReasoning Acting模式用户输入 │ ▼ ┌─────────────────────────┐ │ LLM 思考 决策 │◄──────────┐ │ 输出文字 和/或 工具调用 │ │ └────────┬────────────────┘ │ │ │ 有工具调用 │ ├── 否 → 结束返回文字回复 │ └── 是 → 执行工具 → 把结果给 LLM ──┘每一轮循环中模型可以只输出文字任务完成循环结束调用一个或多个工具循环继续把工具结果喂回去让模型继续思考最简 Agent 实现先不考虑流式输出、上下文压缩、会话持久化这些只实现最核心的循环。消息协议OpenAI 的 Chat API 用一个消息列表来维护对话状态每条消息有一个rolemessages[{role:system,content:你是一个编程助手},{role:user,content:当前目录有哪些文件},{role:assistant,content:None,tool_calls:[{id:call_1,type:function,function:{name:bash,arguments:{command: ls}}}]},{role:tool,tool_call_id:call_1,content:main.py\nREADME.md},{role:assistant,content:当前目录有两个文件main.py 和 README.md},]关键点assistant消息的tool_calls字段告诉我们模型想调用哪些工具tool消息通过tool_call_id关联到对应的工具调用携带工具的执行结果模型看到工具结果后会在下一轮决定继续调用工具还是直接回复核心循环importjsonfromopenaiimportAsyncOpenAI clientAsyncOpenAI()# 工具定义告诉模型有哪些工具可用tools[{type:function,function:{name:bash,description:执行 shell 命令,parameters:{type:object,properties:{command:{type:string,description:要执行的命令}},required:[command],},},}]asyncdefexecute_tool(name:str,arguments:dict)-str:执行工具返回结果字符串。ifnamebash:importasyncio procawaitasyncio.create_subprocess_shell(arguments[command],stdoutasyncio.subprocess.PIPE,stderrasyncio.subprocess.STDOUT,)stdout,_awaitproc.communicate()returnstdout.decode()returnf未知工具:{name}asyncdefagent_loop(user_input:str):最简 Agent Loop不断调用 LLM → 执行工具 → 直到模型不再调用工具。messages[{role:system,content:你是一个编程助手可以执行 shell 命令。},{role:user,content:user_input},]whileTrue:# 1. 调用 LLMresponseawaitclient.chat.completions.create(modelgpt-4o,messagesmessages,toolstools,)msgresponse.choices[0].message# 2. 把 assistant 消息加入历史messages.append(msg.model_dump())# 3. 如果有文字输出打印ifmsg.content:print(fAgent:{msg.content})# 4. 如果没有工具调用循环结束ifnotmsg.tool_calls:break# 5. 执行所有工具调用fortcinmsg.tool_calls:argsjson.loads(tc.function.arguments)print(f [调用工具]{tc.function.name}({args}))resultawaitexecute_tool(tc.function.name,args)print(f [工具结果]{result[:200]})# 6. 把工具结果加入历史messages.append({role:tool,tool_call_id:tc.id,content:result,})# 回到 while True 顶部带着工具结果再次调用 LLM这 50 行代码就是一个完整的 Agent。运行效果 await agent_loop(帮我看看当前目录有什么文件然后统计 Python 文件的总行数) [调用工具] bash({command: ls}) [工具结果] main.py README.md utils.py [调用工具] bash({command: wc -l *.py}) [工具结果] 42 main.py 18 utils.py 60 total Agent: 当前目录有 3 个文件其中 2 个 Python 文件总共 60 行代码。模型自主决定了需要两次工具调用先ls看有什么再wc -l统计行数。这就是 ReAct 循环的威力。从简到繁项目中的实际实现上面的最简版本能跑但缺很多东西。项目中的Agent._loop在此基础上增加了1. 流式输出最简版本用client.chat.completions.create()等全部生成完才返回。实际项目中用流式调用模型边生成边输出# 最简版本等全部生成完responseawaitclient.chat.completions.create(...)msgresponse.choices[0].message# 项目实际流式逐 token 输出streamawaitclient.stream_chat(messages,tools)asyncfordeltainstream:ifdelta.content:yieldTextDelta(contentdelta.content)# 实时推给 UImsgstream.message# 流结束后拿到完整消息2. 事件驱动最简版本直接print。项目中 Agent 不直接操作 UI而是通过yield产出事件对象由外层的 Renderer 负责渲染yieldTextDelta(content...)# 文字增量yieldToolCallStart(tool_namebash)# 工具开始yieldToolCallResult(result...)# 工具结果yieldResponseComplete()# 本轮结束这样 Agent 和 UI 完全解耦——同一个 Agent 可以接 TUI、Web、API 等不同界面。3. 工具并行执行最简版本逐个执行工具。项目中普通工具并行执行asyncio.create_task流式工具串行# 并行发起普通工具pending{}fortc,args,is_streaminginparsed:ifnotis_streaming:pending[tc.id]asyncio.create_task(registry.execute(tc.name,args))# 按原始顺序消费结果fortc,args,is_streaminginparsed:ifis_streaming:asyncforeventintool.run_streaming(**args):yieldeventelse:resultawaitpending[tc.id]yieldToolCallResult(resultresult.content)4. 用户中断项目中支持 Esc 中断。中断时需要把已有的部分输出保存下来让模型下一轮能看到ifself._cancelled:contentf{partial}\n[用户中断以上是中断前的部分输出]self._persist({role:tool,tool_call_id:tc.id,content:content})5. 上下文压缩循环开始前检查 token 是否超限超了就压缩旧对话这部分在第四章详细讲。完整 Loop 的伪代码把上面所有增强合在一起项目中的_loop大致结构如下asyncdef_loop(self):whileTrue:# 上下文压缩检查第四章awaitself._try_compact()# 旧工具输出裁剪第四章micro_compact(self._messages)# 流式调用 LLM逐 token yield 事件msgawaitself._stream_llm()# 保存 assistant 消息到历史self._persist(msg)# 没有工具调用 → 结束ifnotmsg.has_tool_calls:break# 执行工具yield 事件保存 tool 消息awaitself._execute_tools(msg)# 回到 while True带着工具结果再调 LLM对比最简版本结构完全一致——仍然是while True 调 LLM 判断工具 执行工具 循环。所有的增强都是在这个骨架上叠加的。小结概念说明ReAct 循环while True: LLM → 有工具→ 执行 → 再 LLM直到模型不再调用工具消息历史list[dict]按role区分 user / assistant / tool工具通过tool_call_id关联事件驱动Agent 只 yield 事件不操作 UI实现解耦并行执行普通工具asyncio.create_task并行流式工具串行
手把手教你实现 Claude Code（第一章：Agent Loop — ReAct 循环的核心）
发布时间：2026/5/29 6:47:03
项目教程原地址https://github.com/pigeon-dove/anthony-agent第一章Agent Loop — ReAct 循环的核心整个项目的核心是一个不到 50 行的 while 循环。理解了它后面所有章节都是在这个骨架上叠加功能。什么是 Agent一个普通的 LLM 调用是单轮的你发消息模型回复结束。Agent 不一样。它是一个循环模型不仅能回复文字还能决定我需要调用一个工具来获取信息然后拿到工具结果后继续思考直到它认为任务完成。这就是ReActReasoning Acting模式用户输入 │ ▼ ┌─────────────────────────┐ │ LLM 思考 决策 │◄──────────┐ │ 输出文字 和/或 工具调用 │ │ └────────┬────────────────┘ │ │ │ 有工具调用 │ ├── 否 → 结束返回文字回复 │ └── 是 → 执行工具 → 把结果给 LLM ──┘每一轮循环中模型可以只输出文字任务完成循环结束调用一个或多个工具循环继续把工具结果喂回去让模型继续思考最简 Agent 实现先不考虑流式输出、上下文压缩、会话持久化这些只实现最核心的循环。消息协议OpenAI 的 Chat API 用一个消息列表来维护对话状态每条消息有一个rolemessages[{role:system,content:你是一个编程助手},{role:user,content:当前目录有哪些文件},{role:assistant,content:None,tool_calls:[{id:call_1,type:function,function:{name:bash,arguments:{command: ls}}}]},{role:tool,tool_call_id:call_1,content:main.py\nREADME.md},{role:assistant,content:当前目录有两个文件main.py 和 README.md},]关键点assistant消息的tool_calls字段告诉我们模型想调用哪些工具tool消息通过tool_call_id关联到对应的工具调用携带工具的执行结果模型看到工具结果后会在下一轮决定继续调用工具还是直接回复核心循环importjsonfromopenaiimportAsyncOpenAI clientAsyncOpenAI()# 工具定义告诉模型有哪些工具可用tools[{type:function,function:{name:bash,description:执行 shell 命令,parameters:{type:object,properties:{command:{type:string,description:要执行的命令}},required:[command],},},}]asyncdefexecute_tool(name:str,arguments:dict)-str:执行工具返回结果字符串。ifnamebash:importasyncio procawaitasyncio.create_subprocess_shell(arguments[command],stdoutasyncio.subprocess.PIPE,stderrasyncio.subprocess.STDOUT,)stdout,_awaitproc.communicate()returnstdout.decode()returnf未知工具:{name}asyncdefagent_loop(user_input:str):最简 Agent Loop不断调用 LLM → 执行工具 → 直到模型不再调用工具。messages[{role:system,content:你是一个编程助手可以执行 shell 命令。},{role:user,content:user_input},]whileTrue:# 1. 调用 LLMresponseawaitclient.chat.completions.create(modelgpt-4o,messagesmessages,toolstools,)msgresponse.choices[0].message# 2. 把 assistant 消息加入历史messages.append(msg.model_dump())# 3. 如果有文字输出打印ifmsg.content:print(fAgent:{msg.content})# 4. 如果没有工具调用循环结束ifnotmsg.tool_calls:break# 5. 执行所有工具调用fortcinmsg.tool_calls:argsjson.loads(tc.function.arguments)print(f [调用工具]{tc.function.name}({args}))resultawaitexecute_tool(tc.function.name,args)print(f [工具结果]{result[:200]})# 6. 把工具结果加入历史messages.append({role:tool,tool_call_id:tc.id,content:result,})# 回到 while True 顶部带着工具结果再次调用 LLM这 50 行代码就是一个完整的 Agent。运行效果 await agent_loop(帮我看看当前目录有什么文件然后统计 Python 文件的总行数) [调用工具] bash({command: ls}) [工具结果] main.py README.md utils.py [调用工具] bash({command: wc -l *.py}) [工具结果] 42 main.py 18 utils.py 60 total Agent: 当前目录有 3 个文件其中 2 个 Python 文件总共 60 行代码。模型自主决定了需要两次工具调用先ls看有什么再wc -l统计行数。这就是 ReAct 循环的威力。从简到繁项目中的实际实现上面的最简版本能跑但缺很多东西。项目中的Agent._loop在此基础上增加了1. 流式输出最简版本用client.chat.completions.create()等全部生成完才返回。实际项目中用流式调用模型边生成边输出# 最简版本等全部生成完responseawaitclient.chat.completions.create(...)msgresponse.choices[0].message# 项目实际流式逐 token 输出streamawaitclient.stream_chat(messages,tools)asyncfordeltainstream:ifdelta.content:yieldTextDelta(contentdelta.content)# 实时推给 UImsgstream.message# 流结束后拿到完整消息2. 事件驱动最简版本直接print。项目中 Agent 不直接操作 UI而是通过yield产出事件对象由外层的 Renderer 负责渲染yieldTextDelta(content...)# 文字增量yieldToolCallStart(tool_namebash)# 工具开始yieldToolCallResult(result...)# 工具结果yieldResponseComplete()# 本轮结束这样 Agent 和 UI 完全解耦——同一个 Agent 可以接 TUI、Web、API 等不同界面。3. 工具并行执行最简版本逐个执行工具。项目中普通工具并行执行asyncio.create_task流式工具串行# 并行发起普通工具pending{}fortc,args,is_streaminginparsed:ifnotis_streaming:pending[tc.id]asyncio.create_task(registry.execute(tc.name,args))# 按原始顺序消费结果fortc,args,is_streaminginparsed:ifis_streaming:asyncforeventintool.run_streaming(**args):yieldeventelse:resultawaitpending[tc.id]yieldToolCallResult(resultresult.content)4. 用户中断项目中支持 Esc 中断。中断时需要把已有的部分输出保存下来让模型下一轮能看到ifself._cancelled:contentf{partial}\n[用户中断以上是中断前的部分输出]self._persist({role:tool,tool_call_id:tc.id,content:content})5. 上下文压缩循环开始前检查 token 是否超限超了就压缩旧对话这部分在第四章详细讲。完整 Loop 的伪代码把上面所有增强合在一起项目中的_loop大致结构如下asyncdef_loop(self):whileTrue:# 上下文压缩检查第四章awaitself._try_compact()# 旧工具输出裁剪第四章micro_compact(self._messages)# 流式调用 LLM逐 token yield 事件msgawaitself._stream_llm()# 保存 assistant 消息到历史self._persist(msg)# 没有工具调用 → 结束ifnotmsg.has_tool_calls:break# 执行工具yield 事件保存 tool 消息awaitself._execute_tools(msg)# 回到 while True带着工具结果再调 LLM对比最简版本结构完全一致——仍然是while True 调 LLM 判断工具 执行工具 循环。所有的增强都是在这个骨架上叠加的。小结概念说明ReAct 循环while True: LLM → 有工具→ 执行 → 再 LLM直到模型不再调用工具消息历史list[dict]按role区分 user / assistant / tool工具通过tool_call_id关联事件驱动Agent 只 yield 事件不操作 UI实现解耦并行执行普通工具asyncio.create_task并行流式工具串行
进行回归预测（含模型描述及部分示例代码）专栏近期有大量优惠 还请多多点一下关注 加油)
