1. 从“黑盒”到“白盒”为什么我们要绕开SDK如果你最近在折腾AI应用开发尤其是想自己动手搭建一个能调用工具的智能体Agent那么DeepSeek的API和它的tool calling功能大概率已经进入了你的视野。市面上几乎所有的教程和快速上手指南都会告诉你去用官方SDK几行代码就能搞定。这没错用SDK是最高效的方式它帮你封装了HTTP请求、处理了认证、解析了响应甚至提供了优雅的异步接口。但不知道你有没有过这样的感觉——当SDK的某个方法报出一个你看不懂的错误或者你想实现一个SDK没有直接暴露的底层功能时那种深深的无力感。你面对的是一个“黑盒”你只知道输入和输出中间的过程对你而言是一片迷雾。这就是我决定从最原始的curl命令开始一步步手写一个Agent Runtime的原因。这不是为了标新立异也不是故意把简单问题复杂化。恰恰相反这是一种“降维学习”和“深度掌控”。通过亲手组装每一个HTTP请求头、解析每一段JSON响应、设计每一次工具调用的循环逻辑你会对以下几个核心问题有刻骨铭心的理解API通信的本质所谓的“调用API”底层到底交换了哪些数据Authorization头是怎么工作的stream模式下的数据流是如何分块返回的Tool Calling的协议大模型返回的所谓“工具调用”其数据结构究竟长什么样我们如何从一段对话历史中准确地提取出模型“想要调用工具”的意图和参数Runtime的状态管理一个Agent的运行周期Runtime包含哪些状态如何管理多轮对话的历史上下文如何处理工具执行的结果并反馈给模型错误处理的边界网络超时、API限流、模型返回格式异常、工具执行失败……这些情况发生时你的程序应该在哪个层面、以什么方式处理当你用curl手动成功调通一次API再去看SDK的源码你会恍然大悟“哦原来它在这里帮我做了这个判断。” 这种从底层构建的认知是单纯调用高级API无法给予的。它让你在遇到任何诡异问题时都有一条清晰的排查路径从最底层的HTTP报文开始查起。所以这篇文章不是另一个“五分钟快速集成DeepSeek”的教程。它是一个“解剖课”我们将用最基础的工具curl、任意编程语言的HTTP库从零开始构建一个虽然简陋但五脏俱全、完全受你控制的Agent Runtime核心引擎。当你完成它你获得的不仅仅是一个能跑的程序而是一张清晰的“地图”让你能在AI应用开发这片新大陆上自由探索而不是只能沿着SDK铺好的固定道路行走。2. 基石彻底弄懂DeepSeek Chat API的请求与响应在开始造轮子之前我们必须先彻底理解我们要用的“原料”——DeepSeek Chat Completions API。跳过SDK意味着我们需要自己构造符合其规范的HTTP请求。2.1 一次最简单的对话用curl发起请求我们从一个没有任何工具调用的纯对话开始。打开你的终端准备好你的API密钥假设为sk-your-deepseek-api-key-here。curl https://api.deepseek.com/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer sk-your-deepseek-api-key-here \ -d { model: deepseek-chat, messages: [ {role: user, content: 你好请介绍一下你自己。} ], stream: false }让我们拆解这个命令的每一个部分这很重要-H “Content-Type: application/json”: 告诉服务器我们发送的数据体是JSON格式。这是现代RESTful API的通用约定。-H “Authorization: Bearer your-api-key: 身份验证的核心。Bearer是一种令牌认证方式后面紧跟你的API密钥。没有这个头或密钥错误你会立刻收到401 Unauthorized错误。-d ‘{…}’:-d参数表示发送POST数据。后面跟的是一个JSON对象它定义了本次请求的“意图”。”model”: “deepseek-chat”: 指定使用的模型。根据网络信息也可能需要deepseek-v4-pro等请以官方文档为准。这是最容易出错的地方之一模型名不对会直接导致400 Bad Request。”messages”: 这是一个数组包含了对话的历史记录。即使这是第一句话我们也需要把它包装成一个role为user的消息对象。role还可以是system系统指令和assistant助理之前的回复。”stream”: false: 我们关闭了流式输出。这意味着服务器会处理完整个生成过程后一次性返回完整的响应。如果设为true你会收到一个持续的、分块的数据流这对于实现打字机效果很有用但解析起来稍复杂。执行这个命令你会得到一个完整的JSON响应。它可能长这样已简化{ “id”: “chatcmpl-xxx”, “object”: “chat.completion”, “created”: 1234567890, “model”: “deepseek-chat”, “choices”: [ { “index”: 0, “message”: { “role”: “assistant”, “content”: “你好我是DeepSeek一个由深度求索公司创造的AI助手...” }, “finish_reason”: “stop” } ], “usage”: { “prompt_tokens”: 20, “completion_tokens”: 150, “total_tokens”: 170 } }关键响应字段解读choices[0].message.content: 这是我们需要的AI回复文本。finish_reason: 结束原因。”stop”表示模型正常结束如果是”length”则表示因达到max_tokens限制而截断这个字段在后续的Tool Calling中会变得至关重要。usage: 本次请求消耗的令牌数用于计费和监控。注意将stream设为false进行调试是最佳实践。流式响应stream: true返回的是多个data: {...}格式的数据块需要特殊的解析逻辑。在构建核心逻辑时我们先避开这个复杂性。2.2 引入工具定义让AI知道它能“做什么”单纯的对话AI只是一个知识渊博的聊天对象。Agent的核心在于“行动”而行动的前提是AI需要知道有哪些“工具”可用。在DeepSeek API中我们通过tools参数来定义工具列表。假设我们想给AI两个工具一个查询天气一个计算器。我们修改之前的请求{ “model”: “deepseek-chat”, “messages”: [ {role: “user”, “content”: “今天北京天气怎么样”} ], “tools”: [ { “type”: “function”, “function”: { “name”: “get_current_weather”, “description”: “获取指定城市的当前天气情况”, “parameters”: { “type”: “object”, “properties”: { “location”: { “type”: “string”, “description”: “城市名称例如北京上海” }, “unit”: { “type”: “string”, “enum”: [“celsius”, “fahrenheit”], “description”: “温度单位”, “default”: “celsius” } }, “required”: [“location”] } } }, { “type”: “function”, “function”: { “name”: “calculator”, “description”: “执行数学计算”, “parameters”: { “type”: “object”, “properties”: { “expression”: { “type”: “string”, “description”: “数学表达式例如3 5 * (2 - 1)” } }, “required”: [“expression”] } } } ], “stream”: false }工具定义的精髓description至关重要这是AI理解工具用途的唯一文字依据。描述必须清晰、准确。例如“获取天气”就比“查询天气”更指向“当前”状态。parameters是契约它严格定义了AI必须提供哪些参数、参数的类型、格式以及可选值。这就像你和AI签订的一份调用合同。定义得越细致AI调用时出错的可能性就越低。required字段明确哪些参数是调用时必须的。如果AI在需要调用工具时没有提供必填参数那说明我们的提示工程或工具定义可能有问题。发送这个请求响应会有一个关键变化。AI不会直接回答“北京天气如何”因为它发现自己没有实时天气数据。相反它的finish_reason会变成”tool_calls”并且在message中会包含一个tool_calls数组。{ …, “choices”: [ { “index”: 0, “message”: { “role”: “assistant”, “content”: null, “tool_calls”: [ { “id”: “call_abc123”, “type”: “function”, “function”: { “name”: “get_current_weather”, “arguments”: “{\”location\“: \”北京\“, \”unit\“: \”celsius\“}” } } ] }, “finish_reason”: “tool_calls” } ], … }这是整个Tool Calling流程的起点AI的content为null因为它决定不直接生成文本而是发起一个工具调用。tool_calls数组里包含了调用详情id是本次调用的唯一标识后续关联结果用name对应我们定义的工具名arguments是一个JSON字符串包含了模型根据我们定义的parameters规范生成的参数。至此我们完成了单向通信我们告诉AI有什么工具AI告诉我们它想调用哪个工具以及参数是什么。接下来我们需要完成这个循环执行工具并把结果返回给AI。3. 构建循环手写Agent Runtime的核心逻辑一个Agent Runtime本质上是一个状态机它在“等待用户输入”、“调用模型”、“执行工具”、“处理工具结果”这几个状态间循环直到模型决定输出最终答案给用户。我们现在就用代码来模拟这个状态机。3.1 设计Runtime的数据结构首先我们需要定义几个核心数据结构来管理状态。这里以Python为例因其表达清晰但逻辑通用。from typing import Dict, Any, List, Optional, Callable import json # 定义工具一个可调用函数 其JSON Schema描述 class Tool: def __init__(self, name: str, description: str, parameters_schema: Dict, func: Callable): self.name name self.description description self.parameters_schema parameters_schema self.func func # 实际执行工具的函数 def to_api_schema(self) - Dict: 将工具转换为API需要的tools字段格式 return { “type”: “function”, “function”: { “name”: self.name, “description”: self.description, “parameters”: self.parameters_schema } } def execute(self, arguments: str) - Any: 执行工具。arguments是JSON字符串。 try: args_dict json.loads(arguments) return self.func(**args_dict) except json.JSONDecodeError as e: return f“参数解析失败: {e}” except Exception as e: return f“工具执行错误: {e}” # 对话消息 class Message: def __init__(self, role: str, content: Optional[str], tool_calls: Optional[List] None, tool_call_id: Optional[str] None): self.role role # ‘user‘ ’assistant‘ ’tool‘ self.content content self.tool_calls tool_calls # 仅当role‘assistant‘且发起工具调用时有值 self.tool_call_id tool_call_id # 仅当role‘tool‘时有值对应之前的tool_calls[i].id def to_api_dict(self) - Dict: 转换为API请求/历史中的消息格式 msg {“role”: self.role, “content”: self.content} if self.role ‘assistant‘ and self.tool_calls: msg[“tool_calls”] self.tool_calls if self.role ‘tool‘: msg[“tool_call_id”] self.tool_call_id return msg # Agent Runtime 核心类 class AgentRuntime: def __init__(self, api_key: str, model: str “deepseek-chat”): self.api_key api_key self.model model self.tools: Dict[str, Tool] {} # 工具名 - Tool对象 self.message_history: List[Message] [] # 完整的对话历史 self.api_base “https://api.deepseek.com/chat/completions”这个设计的关键点在于Message类。它需要能表示三种角色user: 用户输入。assistant: AI的回复。它可能包含纯文本(content)也可能包含工具调用意图(tool_calls)。tool: 工具执行的结果。它必须通过tool_call_id与之前AI发起的某个工具调用关联起来。AgentRuntime类维护着工具集和历史记录这是它进行多轮推理的“记忆”。3.2 实现单轮对话与工具调用循环现在我们实现Runtime最核心的方法run_cycle。它处理一轮交互发送历史给模型解析响应执行工具并将结果追加回历史。import requests class AgentRuntime: # … __init__ 等 … def add_tool(self, tool: Tool): self.tools[tool.name] tool def _call_api(self, messages: List[Dict]) - Dict: 底层HTTP调用。返回原始的API响应JSON。 headers { “Authorization”: f“Bearer {self.api_key}”, “Content-Type”: “application/json” } payload { “model”: self.model, “messages”: messages, “stream”: False } # 只有在历史中存在工具定义或本轮需要工具时才发送tools参数 # 一个简单的策略如果注册了工具就每次都发送。更复杂的策略可以动态管理。 if self.tools: payload[“tools”] [tool.to_api_schema() for tool in self.tools.values()] try: response requests.post(self.api_base, headersheaders, jsonpayload, timeout30) response.raise_for_status() # 如果状态码不是200抛出HTTPError return response.json() except requests.exceptions.RequestException as e: # 这里应该实现更健壮的错误处理如重试、降级等 raise Exception(f“API调用失败: {e}”) def run_cycle(self, user_input: str) - str: 运行一个完整的Agent循环。 1. 将用户输入加入历史。 2. 调用API。 3. 如果API返回工具调用则执行工具并将结果作为新的‘tool‘消息加入历史然后回到第2步。 4. 如果API返回最终文本则将其加入历史并返回该文本给用户。 # 1. 添加用户消息 self.message_history.append(Message(role“user”, contentuser_input)) # 进入循环因为一次用户输入可能引发多轮“模型-工具-模型”的调用 max_iterations 10 # 防止无限循环 for _ in range(max_iterations): # 2. 准备并发送API请求 api_messages [msg.to_api_dict() for msg in self.message_history] api_response self._call_api(api_messages) # 解析API响应 choice api_response[“choices”][0] message_data choice[“message”] finish_reason choice[“finish_reason”] assistant_msg Message( role“assistant”, contentmessage_data.get(“content”), # 可能是None tool_callsmessage_data.get(“tool_calls”) # 可能是None ) self.message_history.append(assistant_msg) # 3. 判断结束原因 if finish_reason “stop”: # 模型给出了最终答案 final_content assistant_msg.content return final_content if final_content else “(模型返回了空内容)” elif finish_reason “tool_calls”: # 模型要求调用工具 tool_calls assistant_msg.tool_calls if not tool_calls: raise Exception(“API返回finish_reason为tool_calls但未包含tool_calls字段”) # 执行每一个工具调用 for tc in tool_calls: tool_name tc[“function”][“name”] tool_args tc[“function”][“arguments”] call_id tc[“id”] if tool_name not in self.tools: # 工具不存在将错误信息作为工具结果返回 tool_result f“错误未知的工具 ‘{tool_name}‘” else: # 执行工具 tool self.tools[tool_name] tool_result tool.execute(tool_args) # 4. 将工具执行结果作为新的‘tool‘角色消息加入历史 # 注意结果需要是字符串。如果是复杂对象先序列化为JSON字符串。 if not isinstance(tool_result, str): try: tool_result json.dumps(tool_result, ensure_asciiFalse) except: tool_result str(tool_result) tool_msg Message( role“tool”, contenttool_result, tool_call_idcall_id ) self.message_history.append(tool_msg) # 工具执行结果已加入历史循环继续下一次迭代会将包含工具结果的历史再次发给模型 continue else: # 其他情况如 length达到token限制 raise Exception(f“模型生成因‘{finish_reason}‘而结束无法继续处理。”) raise Exception(f“在{max_iterations}轮内未得到最终回复可能陷入循环。”)这个循环是Agent Runtime的灵魂。它清晰地展示了状态流转用户输入 - 加入历史。将整个历史包括之前的用户消息、AI回复、工具结果发送给API。检查finish_reasonstop: 结束返回最终内容。tool_calls: 提取调用信息在本地区域执行对应的工具函数将执行结果以role: tool的消息格式追加到历史中。然后跳回第2步将包含了工具结果的新历史再次发送给模型让模型基于这个结果进行下一步的思考或回答。length: 处理token耗尽的情况实际项目中需要处理比如截断历史或提示用户。3.3 组装并测试一个完整的可运行示例让我们把上面的代码片段组装起来并定义两个简单的工具来测试。# 定义工具函数 def get_weather(location: str, unit: str “celsius”) - str: # 这里应该是调用真实天气API我们模拟一下 weather_data { “北京”: {“celsius”: “22度晴”, “fahrenheit”: “72度晴”}, “上海”: {“celsius”: “25度多云”, “fahrenheit”: “77度多云”}, } city_weather weather_data.get(location, {}) return city_weather.get(unit, “抱歉未找到该城市天气信息。”) def calculate(expression: str) - str: try: # 警告实际生产中直接eval用户/模型提供的表达式极其危险这里仅作演示。 # 应使用安全的数学表达式解析库如 ast.literal_eval 限制更严或专门库。 result eval(expression) return str(result) except Exception as e: return f“计算错误: {e}” # 创建Runtime实例 agent AgentRuntime(api_key“你的真实API密钥”, model“deepseek-chat”) # 创建并注册工具 weather_tool Tool( name“get_current_weather”, description“获取指定城市的当前天气情况”, parameters_schema{ “type”: “object”, “properties”: { “location”: {“type”: “string”, “description”: “城市名称”}, “unit”: {“type”: “string”, “enum”: [“celsius”, “fahrenheit”], “default”: “celsius”} }, “required”: [“location”] }, funcget_weather ) calc_tool Tool( name“calculator”, description“执行数学计算”, parameters_schema{ “type”: “object”, “properties”: { “expression”: {“type”: “string”, “description”: “数学表达式如3 5 * 2”} }, “required”: [“expression”] }, funccalculate ) agent.add_tool(weather_tool) agent.add_tool(calc_tool) # 开始对话 print(“Agent启动。输入‘退出’结束。”) while True: user_input input(“\n你: “) if user_input.lower() in [“退出”, “exit”, “quit”]: break try: response agent.run_cycle(user_input) print(f“Agent: {response}”) except Exception as e: print(f“出错: {e}”) # 可以选择清空历史或保留错误上下文取决于你的策略 # agent.message_history []运行这个程序你可以尝试以下输入“今天北京天气怎么样” - Agent会调用天气工具并返回结果。“那上海呢” - 注意历史中已经有了之前的对话和工具结果AI能理解“上海”指的是天气查询。“计算一下北京气温22摄氏度相当于多少华氏度” - 这是一个组合请求。AI可能会先调用计算器工具进行单位换算也可能直接调用天气工具如果它知道换算公式。这取决于模型的理解和你的工具描述。通过这个简单的循环你已经亲手实现了一个具备基础工具调用、状态管理和多轮对话能力的Agent Runtime核心。它没有使用任何DeepSeek SDK但完整地复现了其最核心的交互协议。4. 从“能跑”到“好用”关键细节与生产环境考量我们有了一个可以工作的原型但把它用于实际项目还需要解决一系列工程化问题。这些才是区分“玩具”和“工具”的关键。4.1 历史上下文的长度管理与Token消耗大模型API按Token收费并且有上下文长度限制如DeepSeek Chat的32K。我们的message_history会随着对话进行无限增长必须进行管理。策略1固定窗口长度只保留最近N条消息。简单粗暴但可能丢失重要的早期指令比如system消息。def trim_history(self, max_messages: int 20): 保留最近max_messages条消息但尽量保留第一条系统消息如果有 if len(self.message_history) max_messages: return # 假设第一条可能是系统消息 system_message None if self.message_history and self.message_history[0].role “system”: system_message self.message_history[0] # 保留系统消息和最新的N-1条消息 messages_to_keep [system_message] if system_message else [] messages_to_keep.extend(self.message_history[-(max_messages - len(messages_to_keep)):]) self.message_history [msg for msg in messages_to_keep if msg is not None]策略2基于Token数的智能截断更精细但需要计算Token数。你可以使用tiktoken库OpenAI开源或模型的Tokenizer来估算。基本思路是从历史尾部开始删除最旧的消息直到总Token数低于阈值。import tiktoken # 需要安装pip install tiktoken class AgentRuntime: def __init__(self, …): # … self.encoding tiktoken.encoding_for_model(“gpt-4”) # 使用一个近似编码器或DeepSeek官方若提供则使用 self.max_context_tokens 8000 # 设定一个安全阈值 def _count_tokens(self, messages: List[Dict]) - int: 粗略估算消息列表的token数。实际生产需更精确。 text “” for msg in messages: text msg.get(“content”, “”) “ ” if msg.get(“tool_calls”): for tc in msg[“tool_calls”]: text tc[“function”][“name”] tc[“function”][“arguments”] return len(self.encoding.encode(text)) def _trim_history_by_tokens(self): 修剪历史确保其token数不超过上限但保留系统消息和最近的关键交互。 while len(self.message_history) 1 and self._count_tokens([m.to_api_dict() for m in self.message_history]) self.max_context_tokens: # 从索引1开始删保留索引0的系统消息或者从最旧的非系统消息开始删 removed False for i in range(1, len(self.message_history)): # 跳过第一条假设是系统消息 if self.message_history[i].role ! “system”: self.message_history.pop(i) removed True break if not removed: # 如果没有非系统消息可删只能删系统消息之后最旧的一条可能是用户消息 if len(self.message_history) 1: self.message_history.pop(1) else: break # 只剩一条无法再删策略3总结压缩这是高级策略。当历史过长时调用模型本身对之前的对话内容进行总结然后用一条“总结消息”替换掉一大段旧历史。这需要额外的API调用和提示词设计成本较高但效果最好。4.2 错误处理与鲁棒性增强我们的原型代码错误处理很基础。在生产环境中必须考虑API调用失败网络超时、5xx服务器错误、速率限制429错误。需要实现重试机制带退避策略和优雅降级。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def _call_api_with_retry(self, messages): # … 调用逻辑 … if response.status_code 429: # 可以从响应头中读取Retry-After raise Exception(“速率限制需要重试”) response.raise_for_status() return response.json()工具执行失败工具函数可能抛出异常、返回错误格式、或长时间不返回。需要超时控制和结果标准化。import concurrent.futures def execute_tool_with_timeout(self, tool: Tool, arguments: str, timeout: int 5): with concurrent.futures.ThreadPoolExecutor() as executor: future executor.submit(tool.execute, arguments) try: return future.result(timeouttimeout) except concurrent.futures.TimeoutError: return “工具执行超时” except Exception as e: return f“工具执行异常: {e}”模型返回格式异常模型可能返回不符合tool_calls格式的arguments不是合法JSON或者调用了未定义的工具。需要在解析时进行验证并给模型反馈友好的错误信息。无限循环防护我们代码中用了max_iterations这是必要的。还可以设置总Token消耗上限或总时间上限。4.3 流式输出Streaming的支持为了更好的用户体验特别是生成长文本时支持流式输出是必要的。这需要改变我们的_call_api方法和run_cycle逻辑。def run_cycle_streaming(self, user_input: str): 流式版本的run_cycle生成器yield每个数据块。 self.message_history.append(Message(role“user”, contentuser_input)) max_iterations 10 for _ in range(max_iterations): api_messages [msg.to_api_dict() for msg in self.message_history] # 这次使用streamTrue headers {…} payload {…, “stream”: True} if self.tools: payload[“tools”] … response requests.post(self.api_base, headersheaders, jsonpayload, streamTrue, timeout30) response.raise_for_status() accumulated_content “” tool_calls_accumulator {} # 可能需要累积多个tool_calls数据块 for line in response.iter_lines(): if line: line line.decode(‘utf-8’) if line.startswith(‘data: ‘): data line[6:] # 去掉‘data: ‘前缀 if data ‘[DONE]‘: break try: chunk json.loads(data) choice chunk.get(“choices”, [{}])[0] delta choice.get(“delta”, {}) # 累积content if “content” in delta and delta[“content”]: accumulated_content delta[“content”] yield {“type”: “content”, “data”: delta[“content”]} # 处理tool_calls (流式下tool_calls也是分块传的) if “tool_calls” in delta: # 这里需要更复杂的逻辑来组装完整的tool_calls对象 # 通常delta.tool_calls是一个数组每个元素有index, id, function等部分信息 # 需要根据index和id进行累积组装 pass except json.JSONDecodeError: continue # 流结束组装完整的assistant消息这里简化实际需要从累积数据组装 # 判断finish_reason等逻辑也需要调整因为流式响应最后会有一个单独的块包含finish_reason # … 后续工具调用、结果返回的逻辑与非流式类似但需要小心处理 …流式处理复杂得多因为模型返回的tool_calls也是分块的。你需要根据index、id等字段将零散的信息片段重新组装成完整的工具调用对象。许多SDK的价值就在于优雅地封装了这些繁琐的细节。4.4 系统提示词System Prompt的集成系统提示词用于设定AI的角色和行为准则。在我们的框架中只需在初始化message_history时在最前面插入一条role为system的消息即可。class AgentRuntime: def __init__(self, api_key: str, model: str “deepseek-chat”, system_prompt: str “”): # … self.message_history [] if system_prompt: self.message_history.append(Message(role“system”, contentsystem_prompt))一个强大的系统提示词能极大提升Agent的可靠性和专业性例如“你是一个严谨的数学助手。在回答数学问题前必须使用计算器工具进行验算。所有最终答案必须包含推导步骤。”5. 超越基础架构演进与高级模式当我们掌握了核心循环就可以在此基础上构建更复杂的Agent架构。5.1 多工具并行执行与结果合并在我们的循环中如果模型返回多个tool_calls我们是顺序执行的。但有些工具之间没有依赖关系例如同时查询北京和上海的天气可以并行执行以降低延迟。import concurrent.futures def execute_tool_calls_parallel(self, tool_calls): 并行执行多个工具调用。 results {} with concurrent.futures.ThreadPoolExecutor() as executor: future_to_call_id {} for tc in tool_calls: tool_name tc[“function”][“name”] if tool_name in self.tools: future executor.submit(self.tools[tool_name].execute, tc[“function”][“arguments”]) future_to_call_id[future] tc[“id”] else: results[tc[“id”]] f“错误未知工具 ‘{tool_name}‘” for future in concurrent.futures.as_completed(future_to_call_id): call_id future_to_call_id[future] try: results[call_id] future.result() except Exception as e: results[call_id] f“工具执行失败: {e}” return results # 返回 call_id - result 的映射然后在run_cycle中根据call_id将并行执行的结果分别构造成tool消息并保持按call_id顺序添加到历史中虽然执行是并行的但添加顺序可能影响模型理解通常按原tool_calls列表顺序添加更安全。5.2 支持“长思考”Chain of Thought与自我反思有时模型在调用工具前需要“思考”一下。我们可以通过提示词鼓励模型在content中输出它的推理过程即使它决定调用工具。我们只需正常地将这些content也存入历史。更高级的模式是让模型在工具执行后不仅基于结果回答还能评估结果的质量甚至决定是否需要重试或调用其他工具。这需要更复杂的循环状态设计。5.3 外部记忆与知识库集成我们的message_history是对话的短期记忆。对于需要长期记忆或访问大量外部知识的场景可以在run_cycle的第一步处理用户输入后加入一个“检索”步骤将用户查询向量化从向量数据库检索相关文档片段并将这些片段作为上下文插入到发送给模型的提示中。这其实就是RAG检索增强生成与Tool Calling的结合。5.4 从脚本到服务构建API接口最终你可能希望将你的Agent Runtime封装成一个Web服务如使用FastAPI对外提供统一的API。这时你需要考虑会话管理为每个对话会话session创建独立的AgentRuntime实例或上下文。异步处理使用async/await和非阻塞HTTP客户端如aiohttp来提高并发性能。可观测性记录每次API调用、工具执行的日志监控Token消耗和延迟。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uuid app FastAPI() sessions {} # session_id - AgentRuntime class ChatRequest(BaseModel): message: str session_id: str None app.post(“/chat”) async def chat_endpoint(request: ChatRequest): session_id request.session_id or str(uuid.uuid4()) if session_id not in sessions: sessions[session_id] AgentRuntime(api_key“你的密钥”) # 可以初始化系统提示词等 sessions[session_id].message_history.append(Message(role“system”, content“你是一个有帮助的助手。”)) agent sessions[session_id] try: response agent.run_cycle(request.message) return {“session_id”: session_id, “response”: response} except Exception as e: raise HTTPException(status_code500, detailstr(e))通过这一系列从底层到上层、从原理到实践的拆解我们完成了一次不依赖任何SDK的Agent Runtime构建之旅。你现在拥有的不仅仅是一个能运行的代码片段而是一个可以随意拆解、改装、扩展的理解。当SDK更新、出现新功能或者遇到难以调试的边界情况时这份从curl开始建立起的认知将成为你最可靠的导航仪。