本博客最终要达成的目标用一个 LLM API 完成普通对话。让模型输出结构化 JSON。定义一个工具函数例如 search、calculator、get_time。会解析模型的 tool call / function call。会执行工具并把工具结果喂回模型。会给 agent loop 加最大步数、超时和错误处理agent的发展chatbox用户输入↓LLM大模型↓生成回复Workflow工作流固定步骤 AI能力开发者提前规定流程agentAI决定流程reactreasonact模式下的Agent loop用户目标 ↓ 思考Reason ↓ 选择工具Act ↓ 观察结果Observe ↓ 继续思考 ↓ 直到完成任务Multi-Agent多智能体用户目标 ↓ Manager Agent ↓ 任务拆分 ↓ 多个Agent协作 ↓ 汇总结果agent 的基本循环observe - think - act - observedef agent_loop(user_input: str): Agent 主循环 messages [{role: user, content: user_input}] while True: # 1. 调用模型 response client.messages.create( modelclaude-3-sonnet-20240229, # 或 deepseek-chat systemSYSTEM_PROMPT, messagesmessages, toolsTOOLS, max_tokens8000 ) # 2. 检查是否需要调用工具 if response.stop_reason tool_use: # 提取工具调用信息 tool_use next(block for block in response.content if block.type tool_use) tool_name tool_use.name tool_input tool_use.input if tool_name bash: # 执行命令 result run_bash(tool_input[command]) # 把结果返回给模型 messages.append({role: assistant, content: response.content}) messages.append({ role: user, content: [{ type: tool_result, tool_use_id: tool_use.id, content: result }] }) else: # 没有工具调用输出最终答案 final_answer response.content[0].text print(fAgent: {final_answer}) break「我的场景为什么需要 agent而不是普通 workflow」比如我要做ai任务规划助手我需要助手帮我理解我的需求、自主分析并且拆解任务、调用外部工具执行子任务、根据执行的结果动态调整后续步骤workflow用户输入 → 步骤1查天气→ 步骤2查酒店→ 步骤3生成建议→ 输出无法处理用户意图不明确的情况、无法根据中间结果临时决定做什么Function CallingFunction calling是让大模型能输出结构化指令的一种方式。模型不会自己执行代码它只输出一个指令JSON 格式你的程序收到后去调用真实的函数然后把结果发回给模型可以结合上面的agent_loop代码看#模型和程序之间交互的过程 我们定义好工具函数名 参数 JSON Schema 把工具定义和用户问题一起发给模型 模型返回 tool_calls包含函数名和参数值 我们的程序解析这个 JSON调用真正的函数 把函数执行结果包装成 tool 消息再发给模型 模型看到结果后要么继续调工具要么生成最终回答function calling和普通api调用的区别普通 API 调用是程序员决定什么时候调什么函数比如if user_input 天气: call_weather_api()。Function Calling 是模型决定什么时候调什么函数程序只负责执行和转发结果。这本质上是控制权的转移从开发者写死的规则变成了模型的动态决策现在自己做一个会调用工具的聊天循环第一版输入你好 模型给出回答不涉及loopfrom http.client import responses from openai import OpenAI #使用openai下的deepseek api client OpenAI( api_keysk-, base_urlhttps://api.deepseek.com/v1,#DeepSeek 在互联网上的一个 门牌号 。当你在代码里调用 client.chat.completions.create(...) 时openai 库就知道要带着你的问题和 API Key到这个地址去找 DeepSeek 的服务器 ) MODEL_NAME deepseek-chat response client.chat.completions.create( modeldeepseek-chat, #整个 Agent Loop 本质上就是在不断维护 messages messages[{role:user,content: 你好}] ) #print(type(response)) print(response)效果模型输出ChatCompletion(id0e4ac917-f4af-49bf-b7d9-a1c98f407340, choices[Choice(finish_reasonstop, index0, logprobsNone, messageChatCompletionMessage(content你好很高兴见到你有什么可以帮你的吗, refusalNone, roleassistant, annotationsNone, audioNone, function_callNone, tool_callsNone))], created1779965294, modeldeepseek-v4-flash, objectchat.completion, service_tierNone, system_fingerprintfp_8b330d02d0_prod0820_fp8_kvcache_20260402, usageCompletionUsage(completion_tokens12, prompt_tokens5, total_tokens17, completion_tokens_detailsNone, prompt_tokens_detailsPromptTokensDetails(audio_tokensNone, cached_tokens0), prompt_cache_hit_tokens0, prompt_cache_miss_tokens5)让ai整理一下格式可以看到response里面有一个choices数组第二版让模型输出结构化 JSON实现“Chatbot” → “Agent Runtime”不再把模型当“聊天对象”而是开始把它当“可编程执行器”第一版模型输出你好有什么可以帮助你的吗程序无法执行期望模型输出然后程序就能parseexecuteloop这才是 Agent代码作如下修改system设定输出格式、角色、规则、示例等SYSTEM_PROMPT 你是一个回答用户问题的助手你的回答必须使用JSON,不要返回自然语言 response client.chat.completions.create( modeldeepseek-chat, #整个 Agent Loop 本质上就是在不断维护 messages messages[ {role: system, content: SYSTEM_PROMPT}, {role:user,content: 你的回答是JSON吗} ], )结果如下已经发现对于ai目前的回答真正有用的内容在response.choices[0].message.content现在整个content回答都在引号里本质还是string要把 JSON 字符串解析成 Python 对象后面才能使用比如基于这个值做下一步决策比如判断对错、继续追问、调用其他函数需要使用到json.loads()函数用法如下import json # 1. JSON 字符串 → Python 字典 json_str {name: 小王, age: 25} data json.loads(json_str) print(data[name]) # 输出小王 # 2. JSON 字符串 → Python 列表 json_str [1, 2, 3] data json.loads(json_str) print(data[0]) # 输出1第三版加入Tool CallingAgent 和普通 chatbot 的真正分界线25 * 48 等于多少普通chatbot1200agent{action: calculator,input: 25 * 48}system_promptAgeng PolicycalculatortoolresponseAgent Brain(LLM)json.loadsstructured parsingif reply_dict[action] ...Tool Routing/Dispatchingcalculator(...)Tool Executionfrom http.client import responses import json from openai import OpenAI #使用openai下的deepseek api client OpenAI( api_keys, base_urlhttps://api.deepseek.com/v1,#DeepSeek 在互联网上的一个 门牌号 。当你在代码里调用 client.chat.completions.create(...) 时openai 库就知道要带着你的问题和 API Key到这个地址去找 DeepSeek 的服务器 ) MODEL_NAME deepseek-chat SYSTEM_PROMPT (你是一个AI Agent你可以使用以下工具: calculator:用于数学计算 当需要调用工具时。你必须返回 { action:工具名, input:输入内容 } 如果不需要工具返回 { action:final_answer, answer:... } ) #自定义一个工具 def calculator(expression): return eval(expression) response client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: SYSTEM_PROMPT}, {role:user,content: 24 * 2} ], ) #print(response) reply_str response.choices[0].message.content #print(reply_str) reply_dict json.loads(reply_str) #print(reply_dict) if reply_dict[action] calculator: answer calculator(reply_dict[input]) print(f计算结果为{answer}) else: answer reply_dict[answer]第四版Tool Result Feedback第三版我们已经完成LLM↓结构化 action↓程序 routing↓工具执行现在距离真正的Agent Loop只差最后一步把模型调用工具的结果喂给模型因为工具结果只是48用户真正想看到的是根据计算结果24*2的结果是.....模型分析第四版流程user: 24*2 assistant: { action:calculator, input:24*2 } tool: 48 assistant: { action:final_answer, answer:48 }from http.client import responses import json import os from dotenv import load_dotenv from openai import OpenAI, api_key load_dotenv() #使用openai下的deepseek api client OpenAI( api_key os.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com/v1,#DeepSeek 在互联网上的一个 门牌号 。当你在代码里调用 client.chat.completions.create(...) 时openai 库就知道要带着你的问题和 API Key到这个地址去找 DeepSeek 的服务器 ) MODEL_NAME deepseek-chat SYSTEM_PROMPT (你是一个AI Agent 你是一个 AI Agent。 你只能输出合法 JSON。 禁止输出任何解释文字。 禁止输出 markdown。 你的回答必须能被 Python json.loads() 直接解析。你可以使用以下工具: calculator:用于数学计算 当需要调用工具时。你必须返回 { action:工具名, input:输入内容 } 如果不需要工具或者当获得工具结果以后返回 { action:final_answer, answer:... } ) #自定义一个工具 def calculator(expression): return eval(expression) user_input 帮我算(153)*8 messages[ {role: system, content: SYSTEM_PROMPT}, {role:user,content: user_input} ] response client.chat.completions.create( modeldeepseek-chat, messagesmessages, ) #print(response) reply_str response.choices[0].message.content print(f原始输出{reply_str}) print(fmessgaes{messages}) try: reply_dict json.loads(reply_str) except json.JSONDecodeError as e: print(fJSON 解析失败: {e}) reply_dict {action: final_answer, answer: 模型返回格式错误} if reply_dict[action] calculator: answer calculator(reply_dict[input]) answer json.dumps(answer) #Agent 要记住自己刚刚做过什么决策 messages.append({role: assistant, content: reply_str}) messages.append({role: user, content: fcalculator结果: {answer}}) response client.chat.completions.create( modeldeepseek-chat, messages messages, ) print(response) second_reply response.choices[0].message.content print(second_reply) else: print(over)进一步升级使用TOOLS以及加入while**字典就是把{key: value}变成keyvalue再传给函数def greet(name, age): return f{name} 今年 {age} 岁 data {name: 小王, age: 25} # 错误写法 greet(data) # ❌ 报错只传了一个参数但函数需要两个 # 正确写法 greet(**data) # ✅ 等价于 greet(name小王, age25)from http.client import responses import json import os from dotenv import load_dotenv from openai import OpenAI, api_key load_dotenv() #使用openai下的deepseek api client OpenAI( api_key os.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com/v1,#DeepSeek 在互联网上的一个 门牌号 。当你在代码里调用 client.chat.completions.create(...) 时openai 库就知道要带着你的问题和 API Key到这个地址去找 DeepSeek 的服务器 ) MODEL_NAME deepseek-chat SYSTEM_PROMPT (你是一个AI Agent 你是一个 AI Agent。 你只能输出合法 JSON。 禁止输出任何解释文字。 禁止输出 markdown。 你的回答必须能被 Python json.loads() 直接解析。你可以使用以下的工具: calculator:数学计算 当需要调用工具时。你必须返回 { action:工具名, input:输入内容 } 如果不需要工具或者当获得工具结果以后返回 { action:final_answer, answer:... } ) MAX_STEPS 10 step 0 #自定义一个工具 def calculator(expression): return eval(expression) TOOLS { calculator: calculator } user_input 帮我算(153)*8 20 messages[ {role: system, content: SYSTEM_PROMPT}, {role:user,content: user_input} ] while step MAX_STEPS: step 1 response client.chat.completions.create( modeldeepseek-chat, messagesmessages, ) #print(response) reply_str response.choices[0].message.content print(f原始输出{reply_str}) print(f{step}) try: reply_dict json.loads(reply_str) except json.JSONDecodeError as e: print(fJSON 解析失败: {e}) reply_dict {action: final_answer, answer: 模型返回格式错误} action reply_dict.get(action) if action final_answer: print(reply_dict[answer]) break else: tool_name reply_dict[action] if tool_name in TOOLS: tool_func TOOLS[tool_name] try: answer tool_func(reply_dict[input]) answer json.dumps(answer) messages.append({role: assistant, content: reply_str}) messages.append({role: user, content: f{tool_name}结果: {answer}}) except Exception as e: messages.append({role: assistant, content: reply_str}) messages.append({role: user, content:str(e)}) else: messages.append({role: user, content: fTool Error: {tool_name} not found}) if step MAX_STEPS: print(Agent reached maximum number of steps)增加get_time、get_weather工具:from http.client import responses import json import os from dotenv import load_dotenv from openai import OpenAI, api_key from datetime import datetime load_dotenv() #使用openai下的deepseek api client OpenAI( api_key os.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com/v1,#DeepSeek 在互联网上的一个 门牌号 。当你在代码里调用 client.chat.completions.create(...) 时openai 库就知道要带着你的问题和 API Key到这个地址去找 DeepSeek 的服务器 ) MODEL_NAME deepseek-chat SYSTEM_PROMPT SYSTEM_PROMPT (你是一个AI Agent 你只能输出合法 JSON。 禁止输出任何解释文字或 markdown。 你的回答必须能被 Python json.loads() 直接解析。 你可以使用以下工具: - calculator: 数学计算 - get_time: 获取当前时间 - get_weather:获取指定地点天气 例如 用户北京天气怎么样 返回 { action:get_weather, input:北京 } 当需要调用工具时你必须返回 { action: 工具名, input: 输入内容 } 如果不需要调用工具或者当获得工具结果以后返回 { action: final_answer, answer: ... } ) MAX_STEPS 10 step 0 #自定义工具 def calculator(expression): return eval(expression) def get_time(_): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(location): return f{location}今天晴35度 TOOLS { calculator: calculator, get_time: get_time, get_weather: get_weather } user_input 天津天气怎么样 messages[ {role: system, content: SYSTEM_PROMPT}, {role:user,content: user_input} ] while step MAX_STEPS: step 1 response client.chat.completions.create( modeldeepseek-chat, messagesmessages, ) #print(response) reply_str response.choices[0].message.content print(f原始输出{reply_str}) print(f{step}) try: reply_dict json.loads(reply_str) except json.JSONDecodeError as e: print(fJSON 解析失败: {e}) reply_dict {action: final_answer, answer: 模型返回格式错误} action reply_dict.get(action) if action final_answer: print(reply_dict[answer]) break else: tool_name reply_dict[action] if tool_name in TOOLS: tool_func TOOLS[tool_name] try: answer tool_func(reply_dict[input]) answer json.dumps(answer) messages.append({role: assistant, content: reply_str}) messages.append({role: user, content: f{tool_name}结果: {answer}}) except Exception as e: messages.append({role: assistant, content: reply_str}) messages.append({role: user, content:str(e)}) else: messages.append({role: user, content: fTool Error: {tool_name} not found}) if step MAX_STEPS: print(Agent reached maximum number of steps)注意一个问题模型是看不到TOOLS的内容的所以更新TOOLS时需要同步Prompt为了解决这个问题进一步增加Tool Registryprompt会根据变动的tools内容动态生成以后新增工具 TOOLS.append({name:search,description:搜索互联网信息,function:search})Prompt自动变成calculator:用于数学计算get_time:获取当前时间search:搜索互联网信息from http.client import responses import json import os from dotenv import load_dotenv from openai import OpenAI, api_key from datetime import datetime load_dotenv() #使用openai下的deepseek api client OpenAI( api_key os.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com/v1,#DeepSeek 在互联网上的一个 门牌号 。当你在代码里调用 client.chat.completions.create(...) 时openai 库就知道要带着你的问题和 API Key到这个地址去找 DeepSeek 的服务器 ) MODEL_NAME deepseek-chat MAX_STEPS 10 step 0 #自定义工具 def calculator(expression: str) - str: return eval(expression) def get_time(_)-str: return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(location:str)-str: return f{location}今天晴35度 #工具注册表 TOOLS [ { name: calculator, description: 用于数学计算, function: calculator }, { name: get_time, description: 获取当前时间, function: get_time }, { name: get_weather, description: 获取location的天气, function: get_weather } ] #现在TOOLS列表中每个元素是一个字典 tool_map {} for tool in TOOLS: tool_map[tool[name]] tool[function] tool_prompt for tool in TOOLS: tool_prompt ( f-{tool[name]}: {tool[description]}\n ) SYSTEM_PROMPT f 你是一个 AI Agent。 你只能输出合法 JSON。 你可以使用以下工具{tool_prompt} 当需要调用工具时返回 {{ action:工具名, input:输入 }} 当任务完成时返回 {{ action:final_answer, answer:... }} user_input 天津天气怎么样 messages[ {role: system, content: SYSTEM_PROMPT}, {role:user,content: user_input} ] while step MAX_STEPS: step 1 response client.chat.completions.create( modeldeepseek-chat, messagesmessages, ) #print(response) reply_str response.choices[0].message.content print(f原始输出{reply_str}) print(f{step}) try: reply_dict json.loads(reply_str) except json.JSONDecodeError as e: print(fJSON 解析失败: {e}) reply_dict {action: final_answer, answer: 模型返回格式错误} action reply_dict.get(action) if action final_answer: print(reply_dict[answer]) break else: tool_name reply_dict[action] #现在TOOLS里面每一个元素是字典 if tool_name in tool_map: tool_func tool_map[tool_name] try: answer tool_func(reply_dict[input]) answer json.dumps(answer) messages.append({role: assistant, content: reply_str}) messages.append({role: user, content: f{tool_name}结果: {answer}}) except Exception as e: messages.append({role: assistant, content: reply_str}) messages.append({role: user, content:str(e)}) else: messages.append({role: user, content: fTool Error: {tool_name} not found}) if step MAX_STEPS: print(Agent reached maximum number of steps)现在检验一下我们新加的Tool Registry同时也是新建两个复杂的工具为后续做铺垫在TOOLS中新加search工具和read_file工具测试输出一下prompt发现已经可以动态生成在项目目录中新建test.txt,内容ATL is the busiest airport in the US.将用户提问分别设置为中国首都是哪里 读取test.txt的内容可以看到输出如下