1. 项目概述为什么我们需要一个“最简Harness”最近在AI应用开发圈里Harness这个词的热度越来越高经常和Agent、LLM、Tool这些概念一起出现。很多刚接触的朋友可能会有点懵Harness到底是什么它和Agent有什么区别为什么我需要从零开始实现一个自己的Harness项目简单来说你可以把Harness理解为一个“AI应用的脚手架”或“执行框架”。如果说LLM大语言模型是大脑Tool工具是手和脚那么Harness就是连接大脑和手脚的神经系统和骨骼。它负责调度、编排、管理整个AI应用的执行流程。而Agent智能体则更像是一个完整的“角色”它拥有目标、记忆、决策能力通常会构建在Harness之上。一个Agent内部可能包含一个或多个Harness来执行具体的任务链。那么为什么要自己动手实现一个最简版本呢市面不是有LangChain、LangGraph这些成熟的框架吗原因有三点第一理解本质。自己动手搭一遍你才能真正理解LLM应用的核心编排逻辑比如工具调用、上下文管理、错误处理而不是停留在API调用的层面。第二轻量与定制。大型框架功能全面但也复杂对于特定场景可能过于臃肿。一个最简Harness只包含你最需要的核心功能没有冗余依赖部署和调试都更简单。第三学习与掌控。这是提升工程能力的最佳路径。当你理解了从提示词构造、函数调用解析到结果归并的完整闭环你就能更自信地应对各种复杂需求。这个项目就是带你从零开始用几百行代码构建一个能跑起来的、功能完整的个人最简Harness。它不追求大而全而是聚焦于最核心的“LLM 工具调用 上下文管理”流程让你彻底搞懂背后的原理。2. 核心设计拆解一个Harness的骨架在动手写代码之前我们必须先想清楚一个最简Harness应该包含哪些核心模块。一个好的设计是成功的一半。2.1 模块化设计思路一个可用的Harness至少需要五大核心组件LLM客户端LLM Client负责与大模型API如OpenAI、DeepSeek、本地模型通信。这是Harness的“思考引擎”。工具管理器Tool Manager负责注册、管理和调用各种外部工具Tool。工具可以是搜索、计算、文件操作等任何函数。上下文管理器Context Manager这是Harness的“记忆体”。它需要管理对话历史、工具调用结果并处理最让人头疼的上下文窗口Context Window限制。网络热词中频繁出现的“maximum context length”错误就是我们要在这里重点解决的问题。编排引擎Orchestration Engine这是Harness的“大脑皮层”。它根据LLM的回复决定下一步是直接回答用户还是调用某个工具或者是进行多轮对话。它实现了最基础的Agent推理循环。主循环Main Loop将以上所有组件串联起来处理用户输入驱动整个交互流程。它们之间的关系可以用一个简单的数据流来描述用户输入 - 主循环 - 上下文管理器组装当前上下文- 编排引擎 - LLM客户端获取决策- 工具管理器如需调用- 上下文管理器更新历史- 输出给用户。2.2 技术选型与考量为什么用Python因为它在AI生态中拥有最丰富的库支持从HTTP请求到JSON解析都极其方便。我们几乎不需要任何外部框架依赖只用标准库和requests足矣。关于LLM API的选择为了通用性我们将设计一个适配器模式。你可以轻松接入OpenAI格式的API包括Azure OpenAI、DeepSeek、Ollama等。核心是定义一个统一的generate方法。关于上下文长度限制这是实战中的高频痛点。模型如GPT-4可能有128K上下文但更经济的模型可能只有4K或8K。我们的Context Manager必须具备“摘要”或“滑动窗口”能力当历史对话超过阈值时能自动压缩旧消息保留最关键信息而不是直接报错“maximum context length is 1048576 tokens”。关于工具调用我们将采用OpenAI的Function Calling格式作为标准。这是一种被广泛支持的、结构化的方式LLM会返回一个包含工具名和参数的JSON方便我们解析和执行。注意在工具设计上务必遵循“单一职责”和“无状态”原则。每个工具函数只做一件事并且尽量不依赖外部可变状态。这能极大降低调试和测试的复杂度。3. 逐步实现从零搭建代码理论说够了我们开始动手。我会分步骤给出核心代码并解释每一行的意图。3.1 第一步构建LLM客户端适配器我们首先定义一个LLM客户端的抽象基类然后实现一个OpenAI兼容的版本。# llm_client.py import json from abc import ABC, abstractmethod from typing import Dict, Any, List, Optional class BaseLLMClient(ABC): LLM客户端的抽象基类 abstractmethod def generate(self, messages: List[Dict[str, str]], tools: Optional[List[Dict]] None) - Dict[str, Any]: 发送消息列表给LLM并可选提供工具列表。 返回LLM的原始响应字典。 pass class OpenAIClient(BaseLLMClient): 兼容OpenAI API格式的客户端 def __init__(self, api_key: str, base_url: str https://api.openai.com/v1, model: str gpt-3.5-turbo): self.api_key api_key self.base_url base_url.rstrip(/) self.model model self.headers { Authorization: fBearer {api_key}, Content-Type: application/json } def generate(self, messages: List[Dict[str, str]], tools: Optional[List[Dict]] None) - Dict[str, Any]: import requests payload { model: self.model, messages: messages, temperature: 0.1 # 降低随机性使工具调用更稳定 } if tools: payload[tools] tools # 强制模型进行工具调用思考 payload[tool_choice] auto try: response requests.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, timeout30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: # 实战心得这里一定要把错误信息详细记录LLM API调用是主要故障点 raise Exception(fLLM API调用失败: {e})关键点解析tool_choice: 设置为”auto”让模型自己决定是否调用工具。你也可以设置为{“type”: “function”, “function”: {“name”: “特定工具名”}}来强制调用。temperature: 设置为较低的0.1因为工具调用需要精确的结构化输出高随机性会导致JSON解析失败。错误处理网络请求必须包含超时和异常捕获在生产环境中还需要加入重试机制和熔断器。3.2 第二步实现工具管理器工具管理器负责维护一个工具注册表并能根据名称安全地调用工具。# tool_manager.py from typing import Dict, Any, Callable, List import inspect class ToolManager: def __init__(self): self._tools: Dict[str, Dict] {} # 存储工具定义 self._functions: Dict[str, Callable] {} # 存储工具函数 def register_tool(self, func: Callable): 通过装饰器或直接调用注册一个工具。 自动从函数签名和docstring生成OpenAI格式的工具定义。 func_name func.__name__ func_doc inspect.getdoc(func) or # 解析函数参数 sig inspect.signature(func) parameters {} required [] for name, param in sig.parameters.items(): if name self: continue param_type string # 简化处理默认string if param.annotation ! inspect.Parameter.empty: # 这里可以做得更精细将Python类型映射为JSON Schema类型 if param.annotation int: param_type integer elif param.annotation float: param_type number elif param.annotation bool: param_type boolean param_def {type: param_type, description: } if param.default inspect.Parameter.empty: required.append(name) parameters[name] param_def tool_def { type: function, function: { name: func_name, description: func_doc.split(\n)[0] if func_doc else , parameters: { type: object, properties: parameters, required: required, } } } self._tools[func_name] tool_def self._functions[func_name] func return func # 支持装饰器语法 def get_tools_definitions(self) - List[Dict]: 获取所有工具的OpenAI格式定义用于发送给LLM。 return list(self._tools.values()) def execute_tool(self, tool_name: str, tool_args: Dict[str, Any]) - Any: 根据工具名和参数执行工具。 if tool_name not in self._functions: raise ValueError(f工具 {tool_name} 未注册。) func self._functions[tool_name] try: # 将参数字典解包传递给函数 return func(**tool_args) except Exception as e: # 实操心得工具执行错误必须被捕获并返回清晰信息供LLM或用户知晓。 return f工具执行错误: {e} # 示例工具定义 ToolManager.register_tool def get_weather(city: str) - str: 获取指定城市的天气信息。 # 这里应该是调用真实天气API我们返回模拟数据 return f{city}的天气是晴朗25摄氏度。 ToolManager.register_tool def calculator(expression: str) - str: 计算一个数学表达式的结果。 try: # 警告使用eval有安全风险仅作演示。生产环境应用安全计算库如ast.literal_eval。 result eval(expression) return f{expression} {result} except Exception as e: return f计算错误: {e}避坑指南工具描述的准确性LLM完全依赖你提供的工具描述description和参数定义来决定是否及如何调用。描述必须清晰、准确。例如“获取天气”比“查询天气信息”更好。错误处理execute_tool方法中的try-except至关重要。工具执行可能因为网络、权限、输入错误而失败必须优雅处理将错误信息返回给上下文让LLM知道发生了什么。安全警告示例中的calculator工具使用了eval这在生产环境中是极度危险的因为它允许执行任意代码。此处仅用于演示原理真实场景必须使用沙箱或安全的表达式求值库。3.3 第三步打造智能上下文管理器上下文管理器是Harness的“记忆中枢”也是最容易出性能问题和上下文溢出错误的地方。# context_manager.py from typing import List, Dict, Any import tiktoken # OpenAI开源的Tokenizer用于估算Token数 class ContextManager: def __init__(self, max_tokens: int 4000, model: str gpt-3.5-turbo): 初始化上下文管理器。 :param max_tokens: 允许的最大上下文Token数需预留一部分给LLM的回复。 :param model: 模型名称用于选择合适的tokenizer。 self.max_tokens max_tokens self.model model self._message_history: List[Dict[str, str]] [] try: self._encoding tiktoken.encoding_for_model(model) except KeyError: # 如果模型未识别使用cl100k_base作为后备GPT-3.5/4通用 self._encoding tiktoken.get_encoding(cl100k_base) def add_message(self, role: str, content: str): 添加一条消息到历史记录。 self._message_history.append({role: role, content: content}) def add_tool_result(self, tool_name: str, result: str): 添加工具调用结果到历史记录格式遵循OpenAI的tool角色。 self._message_history.append({ role: tool, content: result, tool_call_id: call_placeholder # 简化处理实际需与LLM返回的call_id对应 }) def _count_tokens(self, messages: List[Dict]) - int: 估算一组消息消耗的Token数。 # 简化估算将消息字典转换为文本字符串进行计数 total_tokens 0 for message in messages: # 将消息内容序列化为字符串 text f{message[role]}: {message[content]} total_tokens len(self._encoding.encode(text)) return total_tokens def get_current_context(self, system_prompt: str 你是一个有帮助的AI助手。) - List[Dict[str, str]]: 获取当前的对话上下文。 如果历史消息过长会采用滑动窗口策略丢弃最早的对话但保留系统提示和最近的消息。 # 始终以系统提示开始 context_messages [{role: system, content: system_prompt}] # 加入历史消息 context_messages.extend(self._message_history) # 检查并处理上下文超限 while self._count_tokens(context_messages) self.max_tokens: if len(self._message_history) 1: # 如果只有一条用户消息都超了那只能截断消息内容本身极端情况 # 这里可以设计更复杂的策略比如总结超长的单条消息 break # 滑动窗口丢弃历史记录中最早的一条非系统消息 # 我们确保至少保留最近的一条用户消息和其后的AI/工具消息 # 这里实现一个简单的策略从历史记录头部开始丢弃 if len(self._message_history) 0: self._message_history.pop(0) # 移除最早的历史消息 # 重建上下文 context_messages [{role: system, content: system_prompt}] context_messages.extend(self._message_history) return context_messages def clear_history(self): 清空对话历史。 self._message_history.clear()核心策略与实战心得Token估算使用tiktoken是行业标准做法比用简单字数除以某个系数准确得多。务必在初始化时处理好模型兼容性问题。滑动窗口Sliding Window这是处理长上下文最基本、最有效的策略。当Token数超过限制时丢弃最早的历史对话。我们的实现保证了系统提示始终保留并且至少尝试保留最近的交互。更高级的策略对于需要长期记忆的复杂Agent滑动窗口不够。此时可以引入“摘要”功能当历史过长时调用LLM自己将之前的对话总结成一段简短的摘要然后用摘要替换掉旧的历史。这能保留核心信息但会增加成本和延迟。tool角色消息在OpenAI的消息格式中工具执行结果需要以role: “tool”的消息返回并且包含一个tool_call_id来匹配之前的调用请求。我们这里做了简化实际完整实现需要维护这个ID的映射关系。3.4 第四步编写编排引擎与主循环编排引擎是Harness的“决策者”它解析LLM的响应决定下一步动作。# orchestration_engine.py from typing import Dict, Any, Optional import json class OrchestrationEngine: def __init__(self, llm_client, tool_manager, context_manager): self.llm llm_client self.tools tool_manager self.context context_manager def process_llm_response(self, llm_response: Dict[str, Any]) - Dict[str, Any]: 处理LLM的响应。 返回一个字典包含action(‘reply‘, ‘tool_call‘, ‘error‘), content。 choice llm_response.get(choices, [{}])[0] message choice.get(message, {}) finish_reason choice.get(finish_reason) # 1. 检查是否要求调用工具 tool_calls message.get(tool_calls) if tool_calls: # 简化处理只取第一个工具调用实际可支持多个 tool_call tool_calls[0] tool_name tool_call[function][name] try: tool_args json.loads(tool_call[function][arguments]) except json.JSONDecodeError: return {action: error, content: LLM返回的工具参数不是有效的JSON。} return { action: tool_call, tool_name: tool_name, tool_args: tool_args, tool_call_id: tool_call.get(id) # 保存ID以备后用 } # 2. 检查是否为正常回复 content message.get(content) if content and finish_reason stop: return {action: reply, content: content} # 3. 其他情况视为错误或需要特殊处理 return {action: error, content: f无法解析的LLM响应: {finish_reason}} def run_one_cycle(self, user_input: str) - str: 运行一个完整的“用户输入-LLM思考-执行/回复”循环。 返回最终给用户的文本回复。 # 1. 将用户输入加入上下文 self.context.add_message(user, user_input) # 2. 获取当前上下文和工具定义 current_context self.context.get_current_context() available_tools self.tools.get_tools_definitions() # 3. 调用LLM llm_response self.llm.generate(current_context, available_tools) # 4. 处理LLM响应 decision self.process_llm_response(llm_response) final_reply if decision[action] reply: final_reply decision[content] # 将AI的回复加入上下文 self.context.add_message(assistant, final_reply) elif decision[action] tool_call: tool_name decision[tool_name] tool_args decision[tool_args] # 5. 执行工具 tool_result self.tools.execute_tool(tool_name, tool_args) # 6. 将工具执行结果加入上下文 self.context.add_tool_result(tool_name, tool_result) # 7. **关键步骤**重新调用LLM让它结合工具结果进行回答 # 我们需要构建一个新的上下文包含工具结果然后再次调用LLM # 注意这里简化了实际应该用一个新的循环或递归来处理多轮工具调用 new_context self.context.get_current_context() second_llm_response self.llm.generate(new_context, available_tools) second_decision self.process_llm_response(second_llm_response) if second_decision[action] reply: final_reply second_decision[content] self.context.add_message(assistant, final_reply) else: final_reply f工具调用成功但LLM未生成有效回复。工具结果{tool_result} elif decision[action] error: final_reply f系统处理出现错误{decision[content]} return final_reply主循环与启动脚本# main.py from llm_client import OpenAIClient from tool_manager import ToolManager from context_manager import ContextManager from orchestration_engine import OrchestrationEngine def main(): # 1. 初始化各个组件 # 注意请替换为你的真实API密钥和Base URL如果是第三方兼容API llm_client OpenAIClient(api_keyyour-api-key-here, base_urlhttps://api.openai.com/v1, modelgpt-3.5-turbo) tool_manager ToolManager() # 注册工具装饰器已在定义时注册 context_manager ContextManager(max_tokens3000) # 预留约1K Token给LLM生成 # 2. 组装Harness引擎 engine OrchestrationEngine(llm_client, tool_manager, context_manager) print(个人最简Harness已启动输入‘退出’或‘quit’结束对话。) print(- * 40) while True: try: user_input input(\n你: ).strip() if user_input.lower() in [退出, quit, exit]: print(再见) break if not user_input: continue # 3. 运行核心循环 reply engine.run_one_cycle(user_input) print(fAI: {reply}) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f\n系统发生未预期错误: {e}) # 生产环境应记录日志而非直接打印 if __name__ __main__: main()4. 运行、调试与问题排查代码写完了但让它跑起来并稳定工作才是真正的开始。这部分是教科书里不会写的“战场经验”。4.1 首次运行与配置安装依赖创建requirements.txt文件内容为requests2.28.0 tiktoken0.5.0执行pip install -r requirements.txt。配置API在main.py中替换your-api-key-here。如果你使用DeepSeek、Ollama等还需要修改base_url为对应的端点如https://api.deepseek.com/v1。运行测试在命令行执行python main.py。尝试输入“北京天气怎么样” 理论上Harness会调用get_weather工具然后给出回答。4.2 常见问题与解决方案实录以下是我在实现和测试过程中踩过的坑以及解决方法问题1LLM不调用工具总是直接回答。现象输入“计算11”AI直接说出答案而不是调用calculator工具。排查检查工具描述LLM是否理解你的工具确保description清晰明确如“计算一个数学表达式的结果”并包含“计算”、“表达式”等关键词。检查系统提示系统提示system_prompt可以引导AI使用工具。尝试改为“你是一个必须使用工具来回答问题的助手。当用户需要计算或查询信息时请务必调用相应的工具。”检查API参数确保调用llm.generate时tools参数正确传入了工具定义列表。提升温度虽然之前说降低温度稳定但有时模型过于保守。可以尝试将temperature稍微提高到0.3-0.5鼓励其探索工具调用。根本原因工具调用本质上是LLM的一种“推理决策”受提示词、工具描述、模型本身倾向性影响很大。需要耐心调试提示工程。问题2上下文长度超限报错“maximum context length is X tokens”。现象对话轮次多了之后程序报错或回复质量下降。排查确认max_tokens设置我们的ContextManager初始化时设置的max_tokens必须小于模型的实际上下文长度并预留生成空间。例如GPT-3.5-Turbo是16K你设置max_tokens12000比较安全。检查滑动窗口逻辑在get_current_context方法中加入调试打印输出每次估算的Token数和历史消息条数确认滑动窗口是否正常触发。单条消息过长如果用户粘贴了大段文本可能单条消息就超限。需要在add_message时做检查或者实现消息分割/总结功能。解决方案除了滑动窗口对于长文档处理可以引入“向量数据库检索”模式。将长文本切片存储根据用户问题检索相关片段注入上下文而不是塞入全部历史。问题3工具调用参数解析失败。现象LLM决定调用工具了但json.loads解析参数时出错。排查打印原始参数在process_llm_response中打印出tool_call[‘function’][‘arguments’]看是否是合法JSON。常见问题是LLM返回了包含换行或额外注释的JSON。使用更鲁棒的解析可以用json.loads(arguments.strip())或者使用ast.literal_eval作为后备但需注意安全。结构化提示在工具定义的description中明确要求“参数必须是一个严格的JSON对象键名与定义一致。”实战技巧对于复杂参数可以在工具函数内部做类型转换和验证而不是完全依赖LLM。例如即使参数定义是integerLLM也可能传字符串”10″你的工具函数应该能处理int(“10”)。问题4多轮工具调用逻辑混乱。现象我们的run_one_cycle只处理了一轮工具调用。如果LLM在一次回复中要求连续调用多个工具或者根据第一个工具的结果决定调用第二个工具当前逻辑会断裂。解决方案这是区分“玩具”和“可用”Harness的关键。需要将run_one_cycle改造成一个循环或递归函数。在process_llm_response中支持处理tool_calls数组。在run_one_cycle中当action为tool_call时不立即用结果去问LLM而是将每个工具结果按顺序加入上下文。然后用同一个上下文已包含所有工具结果重新调用LLM。如果LLM的新响应还是tool_call则继续循环直到它返回reply或达到最大循环次数防止死循环。4.3 性能优化与扩展思路当你的最简Harness稳定运行后可以考虑以下增强异步化requests是同步的会阻塞。使用aiohttp重写LLM客户端和工具调用可以同时处理多个请求或并行调用多个工具大幅提升响应速度。流式输出修改LLM客户端以支持SSEServer-Sent Events流式响应让AI的回复可以一个字一个字显示体验更佳。持久化记忆将ContextManager中的_message_history保存到数据库或文件实现会话的持久化。下次启动时可以恢复对话。工具动态注册实现一个/tool/register接口允许在运行时动态添加新的工具函数使Harness更具扩展性。验证与监控为工具调用增加权限验证、输入验证。为整个Harness添加日志记录和性能指标监控如每次LLM调用的耗时、Token消耗。从零实现这个最简Harness的过程就像亲手搭建了一个乐高发动机。你看到了每个零件模块如何制造又如何咬合在一起运转。这远比直接使用一个封装好的框架更能带来深刻的理解和掌控感。当你再遇到LangChain中某个抽象概念时你会立刻意识到“哦这背后大概就是在处理我Harness里的那个问题。” 这种从底层构建的认知是应对未来更复杂的AI工程挑战时最宝贵的底气。 从零构建最简AI应用框架：深入理解Harness核心原理与实现
1. 项目概述为什么我们需要一个“最简Harness”最近在AI应用开发圈里Harness这个词的热度越来越高经常和Agent、LLM、Tool这些概念一起出现。很多刚接触的朋友可能会有点懵Harness到底是什么它和Agent有什么区别为什么我需要从零开始实现一个自己的Harness项目简单来说你可以把Harness理解为一个“AI应用的脚手架”或“执行框架”。如果说LLM大语言模型是大脑Tool工具是手和脚那么Harness就是连接大脑和手脚的神经系统和骨骼。它负责调度、编排、管理整个AI应用的执行流程。而Agent智能体则更像是一个完整的“角色”它拥有目标、记忆、决策能力通常会构建在Harness之上。一个Agent内部可能包含一个或多个Harness来执行具体的任务链。那么为什么要自己动手实现一个最简版本呢市面不是有LangChain、LangGraph这些成熟的框架吗原因有三点第一理解本质。自己动手搭一遍你才能真正理解LLM应用的核心编排逻辑比如工具调用、上下文管理、错误处理而不是停留在API调用的层面。第二轻量与定制。大型框架功能全面但也复杂对于特定场景可能过于臃肿。一个最简Harness只包含你最需要的核心功能没有冗余依赖部署和调试都更简单。第三学习与掌控。这是提升工程能力的最佳路径。当你理解了从提示词构造、函数调用解析到结果归并的完整闭环你就能更自信地应对各种复杂需求。这个项目就是带你从零开始用几百行代码构建一个能跑起来的、功能完整的个人最简Harness。它不追求大而全而是聚焦于最核心的“LLM 工具调用 上下文管理”流程让你彻底搞懂背后的原理。2. 核心设计拆解一个Harness的骨架在动手写代码之前我们必须先想清楚一个最简Harness应该包含哪些核心模块。一个好的设计是成功的一半。2.1 模块化设计思路一个可用的Harness至少需要五大核心组件LLM客户端LLM Client负责与大模型API如OpenAI、DeepSeek、本地模型通信。这是Harness的“思考引擎”。工具管理器Tool Manager负责注册、管理和调用各种外部工具Tool。工具可以是搜索、计算、文件操作等任何函数。上下文管理器Context Manager这是Harness的“记忆体”。它需要管理对话历史、工具调用结果并处理最让人头疼的上下文窗口Context Window限制。网络热词中频繁出现的“maximum context length”错误就是我们要在这里重点解决的问题。编排引擎Orchestration Engine这是Harness的“大脑皮层”。它根据LLM的回复决定下一步是直接回答用户还是调用某个工具或者是进行多轮对话。它实现了最基础的Agent推理循环。主循环Main Loop将以上所有组件串联起来处理用户输入驱动整个交互流程。它们之间的关系可以用一个简单的数据流来描述用户输入 - 主循环 - 上下文管理器组装当前上下文- 编排引擎 - LLM客户端获取决策- 工具管理器如需调用- 上下文管理器更新历史- 输出给用户。2.2 技术选型与考量为什么用Python因为它在AI生态中拥有最丰富的库支持从HTTP请求到JSON解析都极其方便。我们几乎不需要任何外部框架依赖只用标准库和requests足矣。关于LLM API的选择为了通用性我们将设计一个适配器模式。你可以轻松接入OpenAI格式的API包括Azure OpenAI、DeepSeek、Ollama等。核心是定义一个统一的generate方法。关于上下文长度限制这是实战中的高频痛点。模型如GPT-4可能有128K上下文但更经济的模型可能只有4K或8K。我们的Context Manager必须具备“摘要”或“滑动窗口”能力当历史对话超过阈值时能自动压缩旧消息保留最关键信息而不是直接报错“maximum context length is 1048576 tokens”。关于工具调用我们将采用OpenAI的Function Calling格式作为标准。这是一种被广泛支持的、结构化的方式LLM会返回一个包含工具名和参数的JSON方便我们解析和执行。注意在工具设计上务必遵循“单一职责”和“无状态”原则。每个工具函数只做一件事并且尽量不依赖外部可变状态。这能极大降低调试和测试的复杂度。3. 逐步实现从零搭建代码理论说够了我们开始动手。我会分步骤给出核心代码并解释每一行的意图。3.1 第一步构建LLM客户端适配器我们首先定义一个LLM客户端的抽象基类然后实现一个OpenAI兼容的版本。# llm_client.py import json from abc import ABC, abstractmethod from typing import Dict, Any, List, Optional class BaseLLMClient(ABC): LLM客户端的抽象基类 abstractmethod def generate(self, messages: List[Dict[str, str]], tools: Optional[List[Dict]] None) - Dict[str, Any]: 发送消息列表给LLM并可选提供工具列表。 返回LLM的原始响应字典。 pass class OpenAIClient(BaseLLMClient): 兼容OpenAI API格式的客户端 def __init__(self, api_key: str, base_url: str https://api.openai.com/v1, model: str gpt-3.5-turbo): self.api_key api_key self.base_url base_url.rstrip(/) self.model model self.headers { Authorization: fBearer {api_key}, Content-Type: application/json } def generate(self, messages: List[Dict[str, str]], tools: Optional[List[Dict]] None) - Dict[str, Any]: import requests payload { model: self.model, messages: messages, temperature: 0.1 # 降低随机性使工具调用更稳定 } if tools: payload[tools] tools # 强制模型进行工具调用思考 payload[tool_choice] auto try: response requests.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, timeout30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: # 实战心得这里一定要把错误信息详细记录LLM API调用是主要故障点 raise Exception(fLLM API调用失败: {e})关键点解析tool_choice: 设置为”auto”让模型自己决定是否调用工具。你也可以设置为{“type”: “function”, “function”: {“name”: “特定工具名”}}来强制调用。temperature: 设置为较低的0.1因为工具调用需要精确的结构化输出高随机性会导致JSON解析失败。错误处理网络请求必须包含超时和异常捕获在生产环境中还需要加入重试机制和熔断器。3.2 第二步实现工具管理器工具管理器负责维护一个工具注册表并能根据名称安全地调用工具。# tool_manager.py from typing import Dict, Any, Callable, List import inspect class ToolManager: def __init__(self): self._tools: Dict[str, Dict] {} # 存储工具定义 self._functions: Dict[str, Callable] {} # 存储工具函数 def register_tool(self, func: Callable): 通过装饰器或直接调用注册一个工具。 自动从函数签名和docstring生成OpenAI格式的工具定义。 func_name func.__name__ func_doc inspect.getdoc(func) or # 解析函数参数 sig inspect.signature(func) parameters {} required [] for name, param in sig.parameters.items(): if name self: continue param_type string # 简化处理默认string if param.annotation ! inspect.Parameter.empty: # 这里可以做得更精细将Python类型映射为JSON Schema类型 if param.annotation int: param_type integer elif param.annotation float: param_type number elif param.annotation bool: param_type boolean param_def {type: param_type, description: } if param.default inspect.Parameter.empty: required.append(name) parameters[name] param_def tool_def { type: function, function: { name: func_name, description: func_doc.split(\n)[0] if func_doc else , parameters: { type: object, properties: parameters, required: required, } } } self._tools[func_name] tool_def self._functions[func_name] func return func # 支持装饰器语法 def get_tools_definitions(self) - List[Dict]: 获取所有工具的OpenAI格式定义用于发送给LLM。 return list(self._tools.values()) def execute_tool(self, tool_name: str, tool_args: Dict[str, Any]) - Any: 根据工具名和参数执行工具。 if tool_name not in self._functions: raise ValueError(f工具 {tool_name} 未注册。) func self._functions[tool_name] try: # 将参数字典解包传递给函数 return func(**tool_args) except Exception as e: # 实操心得工具执行错误必须被捕获并返回清晰信息供LLM或用户知晓。 return f工具执行错误: {e} # 示例工具定义 ToolManager.register_tool def get_weather(city: str) - str: 获取指定城市的天气信息。 # 这里应该是调用真实天气API我们返回模拟数据 return f{city}的天气是晴朗25摄氏度。 ToolManager.register_tool def calculator(expression: str) - str: 计算一个数学表达式的结果。 try: # 警告使用eval有安全风险仅作演示。生产环境应用安全计算库如ast.literal_eval。 result eval(expression) return f{expression} {result} except Exception as e: return f计算错误: {e}避坑指南工具描述的准确性LLM完全依赖你提供的工具描述description和参数定义来决定是否及如何调用。描述必须清晰、准确。例如“获取天气”比“查询天气信息”更好。错误处理execute_tool方法中的try-except至关重要。工具执行可能因为网络、权限、输入错误而失败必须优雅处理将错误信息返回给上下文让LLM知道发生了什么。安全警告示例中的calculator工具使用了eval这在生产环境中是极度危险的因为它允许执行任意代码。此处仅用于演示原理真实场景必须使用沙箱或安全的表达式求值库。3.3 第三步打造智能上下文管理器上下文管理器是Harness的“记忆中枢”也是最容易出性能问题和上下文溢出错误的地方。# context_manager.py from typing import List, Dict, Any import tiktoken # OpenAI开源的Tokenizer用于估算Token数 class ContextManager: def __init__(self, max_tokens: int 4000, model: str gpt-3.5-turbo): 初始化上下文管理器。 :param max_tokens: 允许的最大上下文Token数需预留一部分给LLM的回复。 :param model: 模型名称用于选择合适的tokenizer。 self.max_tokens max_tokens self.model model self._message_history: List[Dict[str, str]] [] try: self._encoding tiktoken.encoding_for_model(model) except KeyError: # 如果模型未识别使用cl100k_base作为后备GPT-3.5/4通用 self._encoding tiktoken.get_encoding(cl100k_base) def add_message(self, role: str, content: str): 添加一条消息到历史记录。 self._message_history.append({role: role, content: content}) def add_tool_result(self, tool_name: str, result: str): 添加工具调用结果到历史记录格式遵循OpenAI的tool角色。 self._message_history.append({ role: tool, content: result, tool_call_id: call_placeholder # 简化处理实际需与LLM返回的call_id对应 }) def _count_tokens(self, messages: List[Dict]) - int: 估算一组消息消耗的Token数。 # 简化估算将消息字典转换为文本字符串进行计数 total_tokens 0 for message in messages: # 将消息内容序列化为字符串 text f{message[role]}: {message[content]} total_tokens len(self._encoding.encode(text)) return total_tokens def get_current_context(self, system_prompt: str 你是一个有帮助的AI助手。) - List[Dict[str, str]]: 获取当前的对话上下文。 如果历史消息过长会采用滑动窗口策略丢弃最早的对话但保留系统提示和最近的消息。 # 始终以系统提示开始 context_messages [{role: system, content: system_prompt}] # 加入历史消息 context_messages.extend(self._message_history) # 检查并处理上下文超限 while self._count_tokens(context_messages) self.max_tokens: if len(self._message_history) 1: # 如果只有一条用户消息都超了那只能截断消息内容本身极端情况 # 这里可以设计更复杂的策略比如总结超长的单条消息 break # 滑动窗口丢弃历史记录中最早的一条非系统消息 # 我们确保至少保留最近的一条用户消息和其后的AI/工具消息 # 这里实现一个简单的策略从历史记录头部开始丢弃 if len(self._message_history) 0: self._message_history.pop(0) # 移除最早的历史消息 # 重建上下文 context_messages [{role: system, content: system_prompt}] context_messages.extend(self._message_history) return context_messages def clear_history(self): 清空对话历史。 self._message_history.clear()核心策略与实战心得Token估算使用tiktoken是行业标准做法比用简单字数除以某个系数准确得多。务必在初始化时处理好模型兼容性问题。滑动窗口Sliding Window这是处理长上下文最基本、最有效的策略。当Token数超过限制时丢弃最早的历史对话。我们的实现保证了系统提示始终保留并且至少尝试保留最近的交互。更高级的策略对于需要长期记忆的复杂Agent滑动窗口不够。此时可以引入“摘要”功能当历史过长时调用LLM自己将之前的对话总结成一段简短的摘要然后用摘要替换掉旧的历史。这能保留核心信息但会增加成本和延迟。tool角色消息在OpenAI的消息格式中工具执行结果需要以role: “tool”的消息返回并且包含一个tool_call_id来匹配之前的调用请求。我们这里做了简化实际完整实现需要维护这个ID的映射关系。3.4 第四步编写编排引擎与主循环编排引擎是Harness的“决策者”它解析LLM的响应决定下一步动作。# orchestration_engine.py from typing import Dict, Any, Optional import json class OrchestrationEngine: def __init__(self, llm_client, tool_manager, context_manager): self.llm llm_client self.tools tool_manager self.context context_manager def process_llm_response(self, llm_response: Dict[str, Any]) - Dict[str, Any]: 处理LLM的响应。 返回一个字典包含action(‘reply‘, ‘tool_call‘, ‘error‘), content。 choice llm_response.get(choices, [{}])[0] message choice.get(message, {}) finish_reason choice.get(finish_reason) # 1. 检查是否要求调用工具 tool_calls message.get(tool_calls) if tool_calls: # 简化处理只取第一个工具调用实际可支持多个 tool_call tool_calls[0] tool_name tool_call[function][name] try: tool_args json.loads(tool_call[function][arguments]) except json.JSONDecodeError: return {action: error, content: LLM返回的工具参数不是有效的JSON。} return { action: tool_call, tool_name: tool_name, tool_args: tool_args, tool_call_id: tool_call.get(id) # 保存ID以备后用 } # 2. 检查是否为正常回复 content message.get(content) if content and finish_reason stop: return {action: reply, content: content} # 3. 其他情况视为错误或需要特殊处理 return {action: error, content: f无法解析的LLM响应: {finish_reason}} def run_one_cycle(self, user_input: str) - str: 运行一个完整的“用户输入-LLM思考-执行/回复”循环。 返回最终给用户的文本回复。 # 1. 将用户输入加入上下文 self.context.add_message(user, user_input) # 2. 获取当前上下文和工具定义 current_context self.context.get_current_context() available_tools self.tools.get_tools_definitions() # 3. 调用LLM llm_response self.llm.generate(current_context, available_tools) # 4. 处理LLM响应 decision self.process_llm_response(llm_response) final_reply if decision[action] reply: final_reply decision[content] # 将AI的回复加入上下文 self.context.add_message(assistant, final_reply) elif decision[action] tool_call: tool_name decision[tool_name] tool_args decision[tool_args] # 5. 执行工具 tool_result self.tools.execute_tool(tool_name, tool_args) # 6. 将工具执行结果加入上下文 self.context.add_tool_result(tool_name, tool_result) # 7. **关键步骤**重新调用LLM让它结合工具结果进行回答 # 我们需要构建一个新的上下文包含工具结果然后再次调用LLM # 注意这里简化了实际应该用一个新的循环或递归来处理多轮工具调用 new_context self.context.get_current_context() second_llm_response self.llm.generate(new_context, available_tools) second_decision self.process_llm_response(second_llm_response) if second_decision[action] reply: final_reply second_decision[content] self.context.add_message(assistant, final_reply) else: final_reply f工具调用成功但LLM未生成有效回复。工具结果{tool_result} elif decision[action] error: final_reply f系统处理出现错误{decision[content]} return final_reply主循环与启动脚本# main.py from llm_client import OpenAIClient from tool_manager import ToolManager from context_manager import ContextManager from orchestration_engine import OrchestrationEngine def main(): # 1. 初始化各个组件 # 注意请替换为你的真实API密钥和Base URL如果是第三方兼容API llm_client OpenAIClient(api_keyyour-api-key-here, base_urlhttps://api.openai.com/v1, modelgpt-3.5-turbo) tool_manager ToolManager() # 注册工具装饰器已在定义时注册 context_manager ContextManager(max_tokens3000) # 预留约1K Token给LLM生成 # 2. 组装Harness引擎 engine OrchestrationEngine(llm_client, tool_manager, context_manager) print(个人最简Harness已启动输入‘退出’或‘quit’结束对话。) print(- * 40) while True: try: user_input input(\n你: ).strip() if user_input.lower() in [退出, quit, exit]: print(再见) break if not user_input: continue # 3. 运行核心循环 reply engine.run_one_cycle(user_input) print(fAI: {reply}) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f\n系统发生未预期错误: {e}) # 生产环境应记录日志而非直接打印 if __name__ __main__: main()4. 运行、调试与问题排查代码写完了但让它跑起来并稳定工作才是真正的开始。这部分是教科书里不会写的“战场经验”。4.1 首次运行与配置安装依赖创建requirements.txt文件内容为requests2.28.0 tiktoken0.5.0执行pip install -r requirements.txt。配置API在main.py中替换your-api-key-here。如果你使用DeepSeek、Ollama等还需要修改base_url为对应的端点如https://api.deepseek.com/v1。运行测试在命令行执行python main.py。尝试输入“北京天气怎么样” 理论上Harness会调用get_weather工具然后给出回答。4.2 常见问题与解决方案实录以下是我在实现和测试过程中踩过的坑以及解决方法问题1LLM不调用工具总是直接回答。现象输入“计算11”AI直接说出答案而不是调用calculator工具。排查检查工具描述LLM是否理解你的工具确保description清晰明确如“计算一个数学表达式的结果”并包含“计算”、“表达式”等关键词。检查系统提示系统提示system_prompt可以引导AI使用工具。尝试改为“你是一个必须使用工具来回答问题的助手。当用户需要计算或查询信息时请务必调用相应的工具。”检查API参数确保调用llm.generate时tools参数正确传入了工具定义列表。提升温度虽然之前说降低温度稳定但有时模型过于保守。可以尝试将temperature稍微提高到0.3-0.5鼓励其探索工具调用。根本原因工具调用本质上是LLM的一种“推理决策”受提示词、工具描述、模型本身倾向性影响很大。需要耐心调试提示工程。问题2上下文长度超限报错“maximum context length is X tokens”。现象对话轮次多了之后程序报错或回复质量下降。排查确认max_tokens设置我们的ContextManager初始化时设置的max_tokens必须小于模型的实际上下文长度并预留生成空间。例如GPT-3.5-Turbo是16K你设置max_tokens12000比较安全。检查滑动窗口逻辑在get_current_context方法中加入调试打印输出每次估算的Token数和历史消息条数确认滑动窗口是否正常触发。单条消息过长如果用户粘贴了大段文本可能单条消息就超限。需要在add_message时做检查或者实现消息分割/总结功能。解决方案除了滑动窗口对于长文档处理可以引入“向量数据库检索”模式。将长文本切片存储根据用户问题检索相关片段注入上下文而不是塞入全部历史。问题3工具调用参数解析失败。现象LLM决定调用工具了但json.loads解析参数时出错。排查打印原始参数在process_llm_response中打印出tool_call[‘function’][‘arguments’]看是否是合法JSON。常见问题是LLM返回了包含换行或额外注释的JSON。使用更鲁棒的解析可以用json.loads(arguments.strip())或者使用ast.literal_eval作为后备但需注意安全。结构化提示在工具定义的description中明确要求“参数必须是一个严格的JSON对象键名与定义一致。”实战技巧对于复杂参数可以在工具函数内部做类型转换和验证而不是完全依赖LLM。例如即使参数定义是integerLLM也可能传字符串”10″你的工具函数应该能处理int(“10”)。问题4多轮工具调用逻辑混乱。现象我们的run_one_cycle只处理了一轮工具调用。如果LLM在一次回复中要求连续调用多个工具或者根据第一个工具的结果决定调用第二个工具当前逻辑会断裂。解决方案这是区分“玩具”和“可用”Harness的关键。需要将run_one_cycle改造成一个循环或递归函数。在process_llm_response中支持处理tool_calls数组。在run_one_cycle中当action为tool_call时不立即用结果去问LLM而是将每个工具结果按顺序加入上下文。然后用同一个上下文已包含所有工具结果重新调用LLM。如果LLM的新响应还是tool_call则继续循环直到它返回reply或达到最大循环次数防止死循环。4.3 性能优化与扩展思路当你的最简Harness稳定运行后可以考虑以下增强异步化requests是同步的会阻塞。使用aiohttp重写LLM客户端和工具调用可以同时处理多个请求或并行调用多个工具大幅提升响应速度。流式输出修改LLM客户端以支持SSEServer-Sent Events流式响应让AI的回复可以一个字一个字显示体验更佳。持久化记忆将ContextManager中的_message_history保存到数据库或文件实现会话的持久化。下次启动时可以恢复对话。工具动态注册实现一个/tool/register接口允许在运行时动态添加新的工具函数使Harness更具扩展性。验证与监控为工具调用增加权限验证、输入验证。为整个Harness添加日志记录和性能指标监控如每次LLM调用的耗时、Token消耗。从零实现这个最简Harness的过程就像亲手搭建了一个乐高发动机。你看到了每个零件模块如何制造又如何咬合在一起运转。这远比直接使用一个封装好的框架更能带来深刻的理解和掌控感。当你再遇到LangChain中某个抽象概念时你会立刻意识到“哦这背后大概就是在处理我Harness里的那个问题。” 这种从底层构建的认知是应对未来更复杂的AI工程挑战时最宝贵的底气。