从LLM裸奔到智能体工程化：Thin Harness与Fat Skills架构实战

1. 项目概述从“裸奔”到“武装”的LLM应用范式转变最近在折腾各种AI应用和智能体Agent项目时我越来越深刻地感受到一个痛点很多团队和个人开发者包括早期的我自己都在让大语言模型LLM“裸奔”。什么叫“裸奔”就是直接把一个原始的、未经任何“工程化包装”的LLM API扔给业务指望它理解复杂的上下文、精准调用工具、并稳定可靠地完成多步骤任务。结果往往是模型要么“一本正经地胡说八道”幻觉要么在复杂的逻辑链条中迷失方向要么因为外部工具调用失败而整个流程崩掉。这种开发模式就像给士兵一把最先进的步枪却不给他配备瞄准镜、战术背心、通讯设备和后勤支援直接让他冲进复杂的战场结果可想而知。于是“Thin Harness, Fat Skills”瘦约束胖技能这个理念开始进入我的视野并彻底改变了我的LLM应用架构思路。这不仅仅是一个技术名词更是一种高效的工程哲学。简单来说“Harness”指的是约束或框架它应该保持“瘦”——即轻量、通用、专注于流程编排、状态管理和异常处理等基础保障。而“Skills”指的是技能或工具它应该追求“胖”——即丰富、强大、专精每一个Skill都像瑞士军刀上的一个工具能独立、可靠地完成一项特定任务比如查天气、写数据库、调API、分析文档。LLM在这个架构中的角色从一个“全知全能但不可靠的魔法黑盒”转变为一个在轻量级框架Harness引导下能够精准调用各种强大工具Skills的“智能调度员”或“决策大脑”。这种转变的核心价值在于它将LLM的“认知不确定性”与外部工具的“执行确定性”进行了完美解耦。LLM擅长理解、规划和决策但在精确计算、数据查询、执行原子操作方面容易出错。而Fat Skills提供了确定性的能力。一个设计良好的Harness其核心任务就是确保LLM在正确的时机以正确的参数调用正确的Skill并妥善处理调用前、调用中、调用后的各种状态和异常。这相当于为LLM这位“天才但粗心的指挥官”配备了一个高效、可靠的“参谋部”和“特种部队”。接下来我将结合我最近在一个内部知识库问答机器人和一个自动化流程审批Agent中的实践详细拆解如何从零开始构建一个“Thin Harness, Fat Skills”架构分享其中的设计思路、核心实现、踩坑经验以及性能调优技巧。无论你是刚开始接触Agent开发的初学者还是正在为LLM应用稳定性头疼的资深工程师相信这套方法论都能给你带来直接的启发和可落地的方案。2. 架构核心深入理解“瘦约束”与“胖技能”的设计哲学在动手写代码之前我们必须把“Thin Harness, Fat Skills”背后的设计哲学吃透。这决定了我们整个系统的高度和健壮性。2.1 为什么Harness要“瘦”Harness的“瘦”体现在其职责的纯粹性和核心性上。它不应该臃肿不应该去抢LLM或Skill的活儿。一个理想的Thin Harness主要承担以下几项核心职责对话与任务流程管理这是Harness最核心的职责。它需要维护与LLM交互的会话状态Session State管理多轮对话的上下文Context Window。更重要的是它需要实现并驱动一个任务执行循环例如经典的ReActReasoning and Acting模式解析用户输入 - LLM思考Reason下一步行动 - 根据思考结果决定是调用Skill还是直接回复 - 执行调用并获取结果 - 将结果反馈给LLM进行下一轮思考。这个循环的驱动引擎就是Harness。Skill的注册、发现与路由Harness需要提供一个清晰的机制让各种各样的Skill能够“插拔”进来。通常这会通过一个Skill注册表Registry来实现。每个Skill需要向Harness声明自己的能力描述比如“这是一个用于查询天气的Skill需要城市名作为参数”。当LLM在“思考”阶段决定要调用某个工具时Harness需要能根据LLM的输出例如一个标准化的函数调用JSON快速、准确地找到对应的Skill实例并将参数传递给它。统一的输入/输出I/O适配与标准化LLM的输入提示词和输出解析需要被标准化。Harness负责组装系统提示词System Prompt、用户历史对话、当前查询以及可用的Skill描述列表形成最终的LLM请求。同时它需要解析LLM的返回结果这个结果可能是一段自然语言回答也可能是一个结构化的函数调用请求。Harness需要能稳健地处理这两种情况并将结构化的调用请求转换为对Skill的实际调用。异常处理与回退机制这是保障系统鲁棒性的关键。Skill调用可能失败网络超时、API限流、参数错误LLM的输出可能无法解析甚至LLM本身可能返回错误。一个健壮的Harness必须内置一套异常处理流程。例如当Skill调用失败时Harness可以捕获异常将错误信息格式化后重新注入上下文让LLM“知道”刚才的动作失败了并尝试新的方案比如换一个Skill或者调整参数。这比整个对话直接崩溃要好得多。可观测性Observability埋点为了后续的调试和优化Harness需要在关键节点如接收请求、发送LLM请求、调用Skill、返回结果记录日志、度量指标Metrics和追踪Traces。这能帮助我们清晰地看到一次请求的完整生命周期每个环节的耗时以及问题出在哪里。注意Harness的“瘦”是相对的是指其功能聚焦。上述核心功能一个都不能少但除此之外的功能比如具体的业务逻辑、复杂的数据处理都应该下放到Skill中去实现。切忌把Harness做成一个“大泥球”Big Ball of Mud。2.2 为什么Skill要“胖”Skill的“胖”体现在其能力的专精性、可靠性和封装完整性上。一个好的Fat Skill应该像一个微服务对外提供清晰、稳定的接口对内隐藏所有实现复杂性。功能原子性与高内聚一个Skill只做好一件事并且把它做到极致。“查询未来三天某城市天气”是一个好的Skill“处理与天气相关的一切事务”就不是。原子性保证了Skill的可靠性和可测试性。高内聚意味着Skill内部的所有代码和逻辑都紧密围绕这一个核心功能。完备的自描述能力Skill必须能够向Harness清晰地描述自己。这通常通过一个标准化的“描述符”Descriptor来实现至少包含名称Name唯一标识符如get_weather。描述Description用自然语言清晰说明这个Skill是做什么的LLM会阅读这个描述来决定是否调用它。例如“根据提供的城市名称获取该城市当前及未来两天的天气情况。”参数模式Parameters Schema一个结构化的定义通常是JSON Schema详细说明调用这个Skill需要哪些参数每个参数的类型、是否必需、以及含义。例如{“city”: {“type”: “string”, “description”: “城市名称如‘北京’、‘Shanghai’”}}。返回模式Returns Schema描述Skill返回结果的结构。这有助于LLM理解返回数据的含义并进行后续推理。强大的错误处理与边界检查Skill不能假设输入一定是完美的。它必须在执行核心逻辑前对输入参数进行严格的验证和清洗。例如一个查询股票的Skill在调用金融API前需要检查股票代码格式是否正确、市场是否存在。如果参数无效或API调用失败Skill应该抛出结构清晰、信息明确的异常并由Harness统一捕获处理而不是让进程崩溃。内置缓存、重试与降级策略对于依赖外部API或耗时计算的Skill应考虑加入缓存层如对天气数据缓存10分钟避免重复调用和限流。对于可能临时失败的操作如网络请求应实现指数退避的重试机制。在极端情况下甚至可以有降级方案如从缓存返回旧数据或返回一个友好的提示信息。独立可测试性由于Skill是原子化的它应该能够被独立地单元测试和集成测试而不需要启动整个Harness和LLM。这极大地提高了开发效率和代码质量。一个生动的类比把构建LLM应用想象成组建一个特种作战小队。LLM是小队指挥官他博学、擅长分析局势、制定宏观计划Thin Harness是他遵循的作战条例和通讯协议轻便但关键。而各个Fat Skills就是小队里的特种兵狙击手精准查询Skill、爆破手数据写入/更新Skill、通信兵API调用Skill、医疗兵错误处理/回退Skill。指挥官LLM根据条例Harness通过电台标准化接口指挥最合适的特种兵Skill去执行具体的、高难度的确定性任务。这样组合起来的团队战斗力远胜于一个什么都会一点但什么都不精通的“全能超人”去单打独斗。3. 从零搭建一个Thin Harness的实现蓝图理论讲完了我们来看看如何用代码实现一个足够“瘦”但功能完备的Harness。这里我会用一个Python的简化示例来展示核心概念你可以根据这个蓝图用任何语言如JavaScript/TypeScript, Go等进行扩展。3.1 定义核心数据模型首先我们需要定义几个核心的数据类来规范整个系统的交互。这是保证类型安全和清晰度的第一步。from pydantic import BaseModel, Field from typing import Any, Dict, List, Optional, Callable from enum import Enum class SkillDescriptor(BaseModel): Skill的描述符用于向Harness和LLM声明自己 name: str Field(..., descriptionSkill的唯一标识符如 search_web) description: str Field(..., description用自然语言描述此Skill的功能LLM会阅读它) parameters_schema: Dict[str, Any] Field(..., description符合JSON Schema的参数定义) # 注意实际实现中returns_schema也很有用这里为简化暂不包含 class SkillInvocation(BaseModel): LLM决定调用某个Skill时产生的结构化请求 skill_name: str Field(..., description要调用的Skill名称) arguments: Dict[str, Any] Field(default_factorydict, description调用参数) class LLMResponse(BaseModel): 标准化LLM的响应可能是纯文本或技能调用 reasoning: Optional[str] Field(None, descriptionLLM的思考过程用于ReAct等模式) content: Optional[str] Field(None, description直接回复用户的纯文本内容) tool_calls: Optional[List[SkillInvocation]] Field(None, description需要调用的Skill列表) class Message(BaseModel): 对话中的一条消息 role: str Field(..., descriptionsystem, user, assistant, 或 tool) content: str Field(..., description消息内容。对于tool角色可以是Skill执行结果) class SessionState(BaseModel): 维护对话会话状态 session_id: str message_history: List[Message] Field(default_factorylist) max_history_turns: int 10 # 控制上下文长度防止无限增长3.2 实现Skill基类与注册中心接下来我们定义所有Skill都需要继承的基类并创建一个全局的注册中心来管理它们。class BaseSkill: 所有Skill的基类 def __init__(self): self.descriptor self._create_descriptor() def _create_descriptor(self) - SkillDescriptor: 子类必须重写此方法返回自己的描述符 raise NotImplementedError async def execute(self, arguments: Dict[str, Any]) - Any: 执行Skill的核心逻辑。arguments是经过Harness验证的参数。 raise NotImplementedError def __repr__(self): return fSkill: {self.descriptor.name} class SkillRegistry: Skill注册中心单例模式管理所有可用Skill _instance None _skills: Dict[str, BaseSkill] {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, skill: BaseSkill): 注册一个Skill实例 name skill.descriptor.name if name in self._skills: raise ValueError(fSkill with name {name} is already registered.) self._skills[name] skill print(fRegistered skill: {name}) def get(self, skill_name: str) - Optional[BaseSkill]: 根据名称获取Skill实例 return self._skills.get(skill_name) def get_all_descriptors(self) - List[SkillDescriptor]: 获取所有Skill的描述符用于组装给LLM的提示词 return [skill.descriptor for skill in self._skills.values()] # 全局注册中心实例 skill_registry SkillRegistry()3.3 构建核心Harness引擎现在我们来构建最关键的Harness引擎。它将串联起状态管理、LLM调用、Skill路由和循环执行。import json import asyncio from abc import ABC, abstractmethod class LLMClient(ABC): LLM客户端的抽象层便于切换不同模型提供商OpenAI, Anthropic, 本地模型等 abstractmethod async def generate(self, messages: List[Dict], tools: Optional[List[Dict]] None) - LLMResponse: pass class ThinHarness: 瘦约束引擎 def __init__(self, llm_client: LLMClient, system_prompt: str 你是一个有帮助的AI助手可以调用工具来解决问题。): self.llm llm_client self.system_prompt system_prompt self.sessions: Dict[str, SessionState] {} def _get_or_create_session(self, session_id: str) - SessionState: 获取或创建一个会话状态 if session_id not in self.sessions: self.sessions[session_id] SessionState(session_idsession_id) # 初始化系统消息 self.sessions[session_id].message_history.append( Message(rolesystem, contentself.system_prompt) ) return self.sessions[session_id] def _format_messages_for_llm(self, session: SessionState) - List[Dict]: 将内部Message格式转换为LLM客户端所需的格式 formatted [] for msg in session.message_history[-session.max_history_turns:]: # 截断历史 if msg.role tool: # 对于工具执行结果通常需要特殊格式这里简化处理 formatted.append({role: tool, content: msg.content, tool_call_id: call_1}) # 简化tool_call_id else: formatted.append({role: msg.role, content: msg.content}) return formatted def _format_tools_for_llm(self) - List[Dict]: 将所有Skill的描述符格式化为LLM可识别的tools格式 descriptors skill_registry.get_all_descriptors() tools [] for desc in descriptors: tool_def { type: function, function: { name: desc.name, description: desc.description, parameters: desc.parameters_schema, } } tools.append(tool_def) return tools async def _execute_skill(self, invocation: SkillInvocation) - str: 执行一个Skill调用并返回结果字符串 skill skill_registry.get(invocation.skill_name) if not skill: return fError: Skill {invocation.skill_name} not found. try: # 这里可以加入参数验证根据parameters_schema、超时控制、重试逻辑等 result await skill.execute(invocation.arguments) # 将结果转换为字符串便于放入对话历史 if isinstance(result, (dict, list)): result_str json.dumps(result, ensure_asciiFalse, indent2) else: result_str str(result) return fTool call {invocation.skill_name} succeeded. Result: {result_str} except Exception as e: # 捕获Skill执行中的异常返回错误信息而不是让整个进程崩溃 return fError executing tool {invocation.skill_name}: {str(e)} async def process_query(self, session_id: str, user_query: str, max_turns: int 5) - str: 处理用户查询的核心循环。 max_turns: 限制ReAct循环的最大轮数防止无限循环。 session self._get_or_create_session(session_id) session.message_history.append(Message(roleuser, contentuser_query)) final_response None for turn in range(max_turns): # 1. 准备LLM请求 messages self._format_messages_for_llm(session) tools self._format_tools_for_llm() # 2. 调用LLM llm_response await self.llm.generate(messages, tools) # 假设我们的LLMClient已经将响应解析为LLMResponse对象 # 3. 处理LLM响应 if llm_response.tool_calls: # LLM决定调用工具 tool_messages [] for tool_call in llm_response.tool_calls: # 执行每一个工具调用 tool_result await self._execute_skill(tool_call) # 将工具执行结果作为一条消息加入历史 tool_msg Message(roletool, contenttool_result) session.message_history.append(tool_msg) tool_messages.append(tool_result) # 如果有工具调用继续下一轮循环让LLM基于工具结果进行下一步思考 # 可以将本轮LLM的“思考”reasoning也加入历史帮助其保持连贯性 if llm_response.reasoning: session.message_history.append(Message(roleassistant, contentfThought: {llm_response.reasoning})) continue # 继续下一轮循环 else: # LLM直接给出了最终回答 final_response llm_response.content session.message_history.append(Message(roleassistant, contentfinal_response)) break # 跳出循环返回最终答案 if final_response is None: final_response f经过{max_turns}轮尝试未能得出最终结论。可能问题过于复杂或工具调用失败。 return final_response3.4 实现一个具体的Fat Skill示例最后我们来实现一个具体的“胖”Skill比如一个获取当前时间的Skill。import datetime class GetCurrentTimeSkill(BaseSkill): 一个获取当前日期和时间的Skill def _create_descriptor(self) - SkillDescriptor: return SkillDescriptor( nameget_current_time, description获取当前的日期和时间信息。可以指定时区例如Asia/Shanghai默认为系统本地时间。, parameters_schema{ type: object, properties: { timezone: { type: string, description: 可选的时区名称如 UTC, America/New_York, Asia/Shanghai。如果未提供使用系统默认时区。 } }, required: [] # 参数都不是必需的 } ) async def execute(self, arguments: Dict[str, Any]) - Dict[str, str]: timezone_str arguments.get(timezone) now datetime.datetime.now() if timezone_str: try: # 这里简化处理实际应用应使用pytz或zoneinfo库 # 假设我们只处理UTC偏移量格式如08:00或常见名称简化版 if timezone_str.upper() UTC: from datetime import timezone now datetime.datetime.now(timezone.utc) # 更复杂的时区处理在此省略... tz_info f(请求时区: {timezone_str}) except Exception: tz_info f(请求的时区{timezone_str}无效已使用系统时间) else: tz_info (系统默认时区) return { iso_format: now.isoformat(), readable: now.strftime(%Y-%m-%d %H:%M:%S), timezone_note: tz_info } # 注册这个Skill skill_registry.register(GetCurrentTimeSkill())实操心得在实现Harness时最容易犯的错误是过早优化和过度设计。我的建议是先让核心循环跑起来。先实现最基本的消息传递、Skill调用和单轮响应。等到你确实遇到了上下文管理问题、性能瓶颈或复杂的错误处理场景时再去迭代增强你的Harness。例如上面示例中的会话管理非常基础在实际生产中你可能需要引入LRU缓存来管理大量会话或者将对话历史持久化到数据库。但这些都是“胖”起来的部分应该在需求明确后再添加以保持Harness核心的“瘦”。4. 实战演练构建一个智能日程管理Agent现在让我们用一个更复杂的例子来串联所有概念构建一个能理解自然语言、并操作谷歌日历的智能日程管理Agent。这个例子将涉及多个Fat Skills和一个协调它们的Thin Harness。4.1 定义Agent的技能集Fat Skills我们的日程管理Agent需要以下核心技能create_calendar_event: 创建新日程。list_calendar_events: 列出特定时间段的日程。update_calendar_event: 更新已有日程。delete_calendar_event: 删除日程。get_current_time(复用之前的): 提供时间参考。每个Skill都需要与Google Calendar API进行交互。为了安全我们需要先处理OAuth2.0认证。这里我们假设已经有一个获取了有效凭证的CalendarService类。# calendar_skill.py import datetime from typing import List, Optional from .base_skill import BaseSkill, SkillDescriptor, skill_registry from .calendar_service import CalendarService # 假设的Google Calendar服务封装 class CreateCalendarEventSkill(BaseSkill): def __init__(self, calendar_service: CalendarService): super().__init__() self.service calendar_service def _create_descriptor(self): return SkillDescriptor( namecreate_calendar_event, description在日历中创建一个新事件。需要事件标题、开始时间、结束时间等详细信息。, parameters_schema{ type: object, properties: { summary: {type: string, description: 事件的标题/名称}, start_time: {type: string, description: 事件开始时间ISO 8601格式例如 2024-05-27T14:30:0008:00}, end_time: {type: string, description: 事件结束时间ISO 8601格式}, description: {type: string, description: 事件的详细描述可选}, location: {type: string, description: 事件地点可选}, attendees: { type: array, items: {type: string}, description: 参会者邮箱列表可选 } }, required: [summary, start_time, end_time] } ) async def execute(self, arguments): # 参数验证与转换 event_data { summary: arguments[summary], start: {dateTime: arguments[start_time]}, end: {dateTime: arguments[end_time]}, } if description in arguments: event_data[description] arguments[description] if location in arguments: event_data[location] arguments[location] if attendees in arguments: event_data[attendees] [{email: email} for email in arguments[attendees]] created_event await self.service.create_event(event_data) return { status: success, event_id: created_event[id], html_link: created_event.get(htmlLink, N/A), message: f日程 {arguments[summary]} 已创建成功。 } class ListCalendarEventsSkill(BaseSkill): def __init__(self, calendar_service: CalendarService): super().__init__() self.service calendar_service def _create_descriptor(self): return SkillDescriptor( namelist_calendar_events, description列出指定时间范围内的日历事件。, parameters_schema{ type: object, properties: { time_min: {type: string, description: 查询起始时间ISO 8601默认是现在。}, time_max: {type: string, description: 查询结束时间ISO 8601默认是24小时后。}, max_results: {type: integer, description: 返回的最大事件数量默认10。} }, required: [] } ) async def execute(self, arguments): time_min arguments.get(time_min) or datetime.datetime.utcnow().isoformat() Z time_max arguments.get(time_max) or (datetime.datetime.utcnow() datetime.timedelta(days1)).isoformat() Z max_results arguments.get(max_results, 10) events await self.service.list_events(time_min, time_max, max_results) simplified_events [] for event in events: start event[start].get(dateTime, event[start].get(date)) end event[end].get(dateTime, event[end].get(date)) simplified_events.append({ id: event[id], summary: event.get(summary, 无标题), start: start, end: end, status: event.get(status) }) return { count: len(simplified_events), time_range: f{time_min} to {time_max}, events: simplified_events } # 类似地实现 UpdateCalendarEventSkill 和 DeleteCalendarEventSkill...4.2 组装Agent与系统提示词工程有了Skills我们需要一个强大的系统提示词来引导LLM正确地使用它们。这个提示词是Harness“智能”的重要组成部分。CALENDAR_AGENT_SYSTEM_PROMPT 你是一个专业的日程管理助手。你的核心能力是调用工具来帮助用户管理他们的谷歌日历。 请遵循以下原则 1. **主动澄清**如果用户的请求模糊例如“明天下午开会”你需要主动询问具体的开始和结束时间、会议标题等缺失信息。 2. **分步执行**对于复杂请求例如“把我明天下午2点到4点的会议移到3点到5点并通知小王”将其分解为多个步骤先查询再更新最后可能需要发送邮件——如果存在邮件Skill。 3. **结果确认**在执行创建、更新、删除等写操作后用简洁的语言向用户确认操作结果并附上关键信息如事件ID或链接。 4. **安全第一**不要假设或猜测用户未明确提供的信息。对于删除操作可以要求二次确认。 5. **工具优先**尽可能使用工具来获取准确信息如当前时间或执行操作而不是依赖你自身的知识你的知识可能过时。 你可以使用的工具如下 {tools_descriptions} 每次思考请遵循以下格式 Thought: 在这里分析用户请求决定下一步是回复还是调用工具。如果需要调用工具说明原因和选择哪个工具。 Action: 如果需要调用工具这里是工具调用的JSON数组。如果不需要此项为null。 Observation: 工具执行的结果会放在这里供你下一轮思考使用。 ... (这个Thought/Action/Observation循环可以重复多次) Final Answer: 当你认为已经充分解决了用户的问题时给出最终的回答。 在Harness初始化时我们会用注册的所有Skill的描述符来填充{tools_descriptions}部分。4.3 运行一个完整对话示例假设我们已经初始化了HarnessThinHarness实例harness并注册了所有Calendar Skills。# 模拟用户对话 session_id user_123 # 第一轮用户提出模糊请求 query1 我明天有什么安排吗 response1 await harness.process_query(session_id, query1) print(f用户: {query1}) print(f助手: {response1}) # 助手可能会调用 list_calendar_events并设置 time_min 为明天零点time_max 为明晚23:59。 # 然后返回“您明天共有3个安排1. 上午10点团队站会2. 下午2点客户访谈3. 晚上7点健身。” # 第二轮用户提出具体操作 query2 帮我在后天下午3点到4点创建一个名为‘项目复盘会’的日程地点在301会议室。 response2 await harness.process_query(session_id, query2) print(f\n用户: {query2}) print(f助手: {response2}) # 助手会调用 get_current_time 来确认“后天”的具体日期然后调用 create_calendar_event。 # 返回“已为您在后天2024-05-29下午3点到4点创建了日程‘项目复盘会’地点301会议室。日历链接[链接]” # 第三轮复杂请求 query3 把刚才创建的那个复盘会提前半小时并加上描述‘讨论Q2项目进度’。 response3 await harness.process_query(session_id, query3) print(f\n用户: {query3}) print(f助手: {response3}) # 这是一个复杂请求。助手需要 # 1. Thought: 用户想修改一个已有事件。我需要先找到“刚才创建”的事件。我应该列出最近的事件来找到它。 # 2. Action: 调用 list_calendar_events设置 time_min 为今天max_results 为5。 # 3. Observation: 获取到事件列表从中识别出“项目复盘会”的事件ID。 # 4. Thought: 找到了事件ID ‘abc123’。现在需要更新它开始时间提前半小时从15:00变为14:30结束时间也提前半小时从16:00变为15:30并添加描述。 # 5. Action: 调用 update_calendar_event传入事件ID和新的开始、结束时间及描述。 # 6. Observation: 更新成功。 # 7. Final Answer: “已将‘项目复盘会’调整为后天下午2:30至3:30并添加了描述。”踩坑实录在实现此类Agent时最大的挑战之一是时间参数的解析和标准化。LLM可能输出“明天下午三点”、“next Monday 2pm”这样的自然语言但Google Calendar API需要ISO 8601格式。早期版本我试图让LLM直接输出ISO时间但格式错误率很高。解决方案是要么在Skill内部进行自然语言时间解析可以使用dateparser库要么设计更复杂的交互让Harness在发现时间参数不标准时触发一个专门的“时间澄清”子对话或者让LLM调用一个parse_natural_language_time的Skill来先进行转换。我最终选择了后者因为它更符合“Fat Skill”的理念——将复杂、确定性的解析任务交给专门的Skill处理。5. 高级话题与性能优化当你的Agent系统开始处理真实流量时性能、成本和稳定性问题就会浮现。以下是一些关键的优化方向。5.1 上下文管理与Token优化LLM的上下文窗口是宝贵且有限的资源。我们的对话历史会不断增长必须进行智能管理。选择性记忆不要无脑保存所有历史消息。可以只保留最近N轮对话如上述示例中的max_history_turns或者更智能地总结Summarize较早的对话内容。例如每经过5轮对话就用LLM将之前的对话压缩成一段摘要然后用摘要替换掉原始的长篇历史。LangChain等框架提供了多种对话记忆管理策略。Skill描述的精简在_format_tools_for_llm中我们发送了完整的Skill描述。当Skill很多时这会占用大量Token。可以考虑动态选择根据用户当前查询的意图只发送最相关的几个Skill的描述。这需要另一个轻量级模型或规则进行意图分类。描述压缩为每个Skill编写更简短、更精炼的描述同时不损失关键信息。系统提示词优化系统提示词往往很长且每次请求都会发送。确保它没有冗余信息。可以考虑将部分固定的指令放在LLM的“系统”角色消息中而将动态的如Skill列表放在“用户”或“助理”消息中部分模型对系统消息的Token计算方式可能不同。5.2 异步、超时与并发控制异步化如上文代码所示全程使用async/await。这允许你在等待LLM响应或某个Skill的I/O操作如网络请求时可以处理其他请求极大提高吞吐量。超时设置必须为LLM调用和每个Skill调用设置超时。使用asyncio.wait_for。try: llm_response await asyncio.wait_for(self.llm.generate(messages, tools), timeout30.0) except asyncio.TimeoutError: # 记录日志返回友好错误信息或尝试降级方案 return 请求超时请稍后再试。并发与限流如果你的服务会同时处理多个用户请求需要考虑对LLM API的调用进行限流Rate Limiting避免触发上游服务的限制。可以使用像asyncio.Semaphore或更高级的库如aiohttp的限流客户端来控制并发数。5.3 可观测性与调试一个黑盒的Agent是可怕的。必须建立强大的可观测性体系。结构化日志在Harness的每个关键步骤接收请求、调用LLM、调用Skill、返回响应记录结构化日志JSON格式。日志应包含session_id、turn_id、action、input、output、duration、error如果有等字段。链路追踪Tracing为每个用户请求分配一个唯一的trace_id并贯穿整个调用链Harness - LLM - Skill1 - Skill2 ...。这能让你在分布式系统中清晰地看到一个请求的完整路径和耗时瓶颈。可以使用OpenTelemetry等标准。监控与告警监控关键指标请求量/QPS平均响应时间 P95/P99延迟LLM调用耗时 Token消耗各Skill调用成功率与耗时错误率LLM错误、Skill错误、超时错误 当这些指标出现异常如错误率飙升、延迟增加时触发告警。5.4 测试策略测试Agent比测试普通软件更复杂因为涉及非确定性的LLM。Skill的单元测试这是最容易的。像测试普通函数一样为每个Skill的execute方法编写详尽的单元测试覆盖正常情况和各种边界、错误情况。Harness的集成测试Mock掉LLM客户端和具体的Skill测试Harness的流程逻辑是否正确。例如模拟LLM返回一个工具调用请求检查Harness是否能正确路由并调用Mock Skill然后模拟Skill返回结果检查Harness是否能正确继续循环。端到端E2E测试与评估这是最具挑战性的。需要准备一组具有标准答案的测试用例例如“创建明天下午3点的会议”然后运行整个Agent评估其最终输出是否正确。由于LLM输出的非确定性评估往往需要基于规则检查返回文本中是否包含特定关键词或使用另一个LLM作为裁判LLM-as-a-Judge来进行评估。这个过程对于保证Agent质量的稳定性至关重要。个人体会从“LLM裸奔”到“Thin Harness, Fat Skills”的转变不是一个一蹴而就的项目而是一个持续迭代的工程实践。最开始你的Harness可能只是一个简单的循环Skills也只有一两个。随着业务复杂度的增加你会不断地往Harness里添加新的“保障设施”如更好的错误处理、更智能的上下文管理同时也不断地开发出更多专精的“Fat Skills”。这个架构的魅力在于它的清晰度和可扩展性。无论业务需求如何变化你都能清楚地知道新的逻辑应该作为一个独立的Skill存在还是应该成为Harness核心流程的一部分。这种分离让你和你的团队能够并行开发高效协作最终构建出既强大又可靠的智能体应用。别再让LLM孤军奋战了是时候为它打造一支强大的“技能特种部队”和一个精干的“指挥框架”了。