learn-claude-code/s01_agent_loop at main · shareAI-lab/learn-claude-code
Agent归根到底就是一个循环
- 阅读query
- thinking……
- do something……/stop
- observation
Agent通过工具调用进行外部交互
While True的停止条件实际上就是
stop_reason == "tool_use"：执行，返回工具结果，继续循环stop_reason != "tool_use"：退出循环
Antropic SDK
和Openai风格的SDK不太一样，似乎大家喜欢用A社的做工具调用
1#!/usr/bin/env python3
2
3import os
4from dotenv import load_dotenv
5from anthropic import Anthropic
6
7load_dotenv(override=True)
8
9client = Anthropic(
10 base_url=os.getenv("ANTHROPIC_BASE_URL"),
11 api_key=os.getenv("ANTHROPIC_API_KEY"),
12)
13
14MODEL = os.getenv("MODEL_ID", "claude-3-5-sonnet-latest")
15
16TOOLS = [
17 {
18 "name": "bash",
19 "description": "Run a shell command.",
20 "input_schema": {
21 "type": "object",
22 "properties": {
23 "command": {"type": "string"}
24 },
25 "required": ["command"],
26 },
27 }
28]
29
30response = client.messages.create(
31 model=MODEL,
32 max_tokens=1024,
33 messages=[
34 {
35 "role": "user",
36 "content": (
37 "请先用一句话说明你将要做什么，"
38 "然后调用 bash 工具查看当前目录下的文件列表。"
39 ),
40 }
41 ],
42 tools=TOOLS
43)
44
45print(response)
Tool
工具定义，其实就是定义一个函数
1{
2 "name": "bash",
3 "description": "Run a shell command.",
4 "input_schema": {
5 "type": "object",
6 "properties": {
7 "command": {"type": "string"}
8 },
9 "required": ["command"],
10 },
11}
相当于注册了一个函数bash(command:str)
- name：工具名，模型调用工具必须名字一致
- description：告诉模型工具的说明
- input_schema：工具输入参数定义
定义了一个command参数（required，必须存在）
参数值是字符串，且传参方式是是字典对象
例如：
{"command": "ls -la"}
Return
上述代码的返回是：
1Message(
2 id='ec45b38f-7416-4b74-b501-e03448af1633',
3 container=None,
4 content=[
5 ThinkingBlock(
6 signature='ec45b38f-7416-4b74-b501-e03448af1633',
7 thinking="....",
8 type='thinking'
9 ),
10 TextBlock(
11 citations=None,
12 text='我将使用 bash 工具查看当前目录下的文件列表。',
13 type='text'
14 ),
15 ToolUseBlock(
16 id='call_00_9mk6ZlmzqJtCPY0Epd165520',
17 caller=None,
18 input={
19 'command': 'ls -la'
20 },
21 name='bash',
22 type='tool_use'
23 )
24 ],
25 model='deepseek-v4-pro',
26 role='assistant',
27 stop_details=None,
28 stop_reason='tool_use',
29 stop_sequence=None,
30 type='message',
31 usage=Usage(
32 cache_creation=None,
33 cache_creation_input_tokens=0,
34 cache_read_input_tokens=256,
35 inference_geo=None,
36 input_tokens=33,
37 output_tokens=86,
38 output_tokens_details=None,
39 server_tool_use=None,
40 service_tier='standard'
41 )
42)
content：按Block进行切分，有专门的ToolUseBlock方便解析stop_reason：这里是tool_use，说明需要调用工具
Code
完整代码：
1#!/usr/bin/env python3
2
3import os
4from dotenv import load_dotenv
5from anthropic import Anthropic
6import subprocess
7
8load_dotenv(override=True)
9
10client = Anthropic(
11 base_url=os.getenv("ANTHROPIC_BASE_URL"),
12 api_key=os.getenv("ANTHROPIC_API_KEY"),
13)
14
15MODEL = os.getenv("MODEL_ID", "claude-3-5-sonnet-latest")
16
17TOOLS = [
18 {
19 "name": "bash",
20 "description": "Run a shell command.",
21 "input_schema": {
22 "type": "object",
23 "properties": {
24 "command": {"type": "string"}
25 },
26 "required": ["command"],
27 },
28 }
29]
30
31
32
33
34SYSTEM = f"You are a coding agent at {os.getcwd()}. "
35
36# ── Tool execution ────────────────────────────────────────
37def run_bash(command: str) -> str:
38 dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
39 if any(d in command for d in dangerous):
40 return "Error: Dangerous command blocked"
41 try:
42 r = subprocess.run(command, shell=True, cwd=os.getcwd(),
43 capture_output=True, text=True, timeout=120)
44 out = (r.stdout + r.stderr).strip()
45 return out[:50000] if out else "(no output)"
46 except subprocess.TimeoutExpired:
47 return "Error: Timeout (120s)"
48 except (FileNotFoundError, OSError) as e:
49 return f"Error: {e}"
50
51def agent_loop(history):
52 while True:
53
54 response = client.messages.create(
55 model=MODEL,
56 max_tokens=1024,
57 messages=history,
58 tools=TOOLS,
59 system=SYSTEM,
60 )
61
62 # 保存模型记录
63 history.append({"role": "assistant", "content": response.content})
64
65 # 如果不再需要工具调用
66 if response.stop_reason != "tool_use":
67 return
68
69 tool_results = []
70 for block in response.content:
71 if block.type == 'tool_use':
72 print(f"Tool call: {block.name} with input {block.input}")
73
74 run_bash_result = run_bash(block.input['command'])
75 tool_results.append({
76 "type": "tool_result",
77 "tool_use_id": block.id, # 对应同block的工具调用id
78 "content": run_bash_result
79 })
80
81 # 将工具调用结果添加到历史记录中，供模型后续生成使用
82 history.append({"role": "user", "content": tool_results})
83
84
85
86
87if __name__ == "__main__":
88
89 history = []
90
91 # 这里是多轮对话的循环
92 while True:
93 query = input("Enter your query: ")
94
95 if query.lower() in ["exit", "quit"]:
96 print("Exiting the agent loop.")
97 break
98
99 history.append({"role": "user", "content": query})
100 agent_loop(history) # Agent内部循环，完成工具调用和响应生成
101
102 final_output = history[-1]["content"]
103 print(final_output)
Claude Code
以上是一个AgentLoop最小实现
但对于Claude Code的实现也值得去看一下
循环结构
CC并不是等待一个响应完成，检测最后的stop_reason
而是根据当前流式输出内容，如果出现ToolUseBlock
设置needsFollowUp = True，表示需要下一轮循环
这里只是说明如何判断Loop中断
stop_reason事实上也有其他逻辑使用，暂时不提
同时为了性能，工具调用也会在流式时进行
但是会做concurrency-safe的检测，决定并行or独占
同时针对费用超限、结构化输出失败等做了非常多保护
退出
CC事实上对blocking limit、prompt too long、model error、abort、hook stop、max turns、token budget continuation、reactive compact retry 等场景，都准备了退出、恢复机制
以上代码只有模型不调用工具就结束的逻辑