1. 项目概述从“笔记”到“工程实践”最近在整理Harness Engineering相关的技术文档时我发现关于Agent运行时核心机制的讨论尤其是循环、路由与上下文这三块是很多开发者从“会用框架”到“理解框架”的关键门槛。网上能找到的资料要么过于零散只讲某个API怎么调用要么过于理论堆砌一堆架构图却不说清楚代码到底怎么跑起来的。这让我觉得有必要结合自己踩过的坑把这块“硬骨头”啃下来写一篇能直接指导编码的深度解析。所谓Harness Engineering你可以把它理解为一套用于构建、管理和控制智能体Agent的工程化框架与最佳实践。它关注的不是某个单一的算法模型而是如何让多个Agent协同、稳定、高效地完成复杂任务。在这个体系里Agent运行时就是那个让智能体“活”起来、能够感知、决策并执行的核心引擎。而驱动这个引擎的三大核心机制正是循环Loop、路由Routing和上下文Context。理解它们你才能真的驾驭Agent而不是被框架牵着鼻子走。这篇文章适合谁呢如果你正在或打算开发基于Agent的应用无论是自动化工作流、智能客服还是复杂的决策系统并且已经过了“Hello World”阶段开始头疼于Agent的状态管理、任务调度和信息流转问题那么这里面的内容就是为你准备的。我会尽量避开空泛的概念用具体的代码段、设计抉择背后的“为什么”以及我实际调试中总结出的“坑点”来展开。2. 核心机制深度拆解循环、路由与上下文的角色与联动在深入细节之前我们必须建立一个顶层的认知模型这三个机制不是孤立的它们共同构成了Agent运行时的一个完整“心跳周期”。想象一下一个处理用户咨询的客服Agent。它拿到用户问题上下文需要决定这个问题属于售前、售后还是技术故障路由然后根据决定调用相应的子能力或工具去执行执行后产生新的结果并评估是否还需要进一步追问或可以结束循环。这个过程周而复始。2.1 循环LoopAgent执行的状态机与驱动力循环机制是Agent运行的“节拍器”。它决定了Agent在单次触发后如何推进其内部状态何时停止。最简单的循环就是“执行一次就结束”One-Shot。但对于复杂任务我们需要更强大的循环模式。2.1.1 常见循环模式与实践选择While-Loop条件循环 这是最常用、最灵活的循环模式。它基于一个或多个条件来决定是否继续迭代。例如一个数据分析Agent可能会循环执行“查询-分析-提炼”步骤直到分析结果的置信度达到某个阈值或者用户明确要求停止。# 伪代码示例基于条件的While循环 context initialize_context(user_query) while not should_stop(context): # 1. 路由决策根据当前context决定下一步动作 action routing_engine.decide(context) # 2. 执行动作 result execute_action(action, context) # 3. 更新上下文 context.update(result, action) # 4. 评估停止条件例如任务完成、达到最大步数、用户中断 context.evaluate_stop_condition()实操心得should_stop函数的设计是核心。不要只依赖单一条件如任务标记完成最好结合多个维度最大迭代次数防止死循环、用户主动取消信号、连续多次迭代未产生有效进展等。我通常会设置一个默认的最大步数比如50步作为安全网。For-Loop有限循环 适用于步骤明确、次数固定的任务。比如一个Agent需要依次检查系统的A、B、C三个服务状态。checklist [检查数据库连接, 验证API网关, 扫描日志错误] for task in checklist: context.set_current_task(task) result execute_standard_check(task, context) context.record_check_result(task, result)注意事项即使在For-Loop中也要为每个步骤设计异常处理。某个步骤的失败不应导致整个Agent崩溃而应该被捕获并记录到上下文中供后续步骤或最终汇总时参考。递归循环Recursive Loop 当任务可以分解为同构的子任务时使用。例如一个文件整理Agent遇到文件夹就递归处理其中的文件。核心挑战上下文的管理。递归每一层都需要有自己的上下文“快照”同时又能访问到必要的父级信息。需要清晰界定上下文的作用域和继承关系否则很容易造成信息污染或丢失。2.1.2 循环控制的高级技巧暂停与恢复Pause/Resume对于长时任务Agent需要支持暂停。关键在于将完整的运行时状态包括循环索引、当前上下文、中间结果序列化并持久化。恢复时再反序列化从断点继续。这通常需要框架层面的支持。并行循环当多个子任务相互独立时可以使用并行循环来提升效率。但要注意共享上下文资源的线程安全问题。一个稳妥的做法是采用“复制-合并”策略每个并行任务处理上下文的一个副本执行完毕后再将结果合并回主上下文。循环超时与看门狗Watchdog必须为每个循环设置超时机制。一个独立的看门狗线程可以监控主循环的执行时间一旦超时就触发中断流程保存现场并上报错误避免Agent“卡死”。2.2 路由RoutingAgent的决策中枢与流量控制器如果说循环是“节奏”那么路由就是“方向”。它负责在运行时根据当前上下文动态决定下一步该执行哪个动作、调用哪个工具、或者将任务移交给哪个更专业的子Agent。2.2.1 路由策略解析路由的核心是一个决策函数Input(Current_Context) - Output(Next_Action/Agent)。基于规则的路由Rule-Based 最简单直接的方式。通过预定义的if-else或决策树来路由。def rule_based_router(context): if 退款 in context.user_intent: return refund_agent elif 技术故障 in context.sentiment and context.user_tier VIP: return premium_support_agent else: return general_support_agent优点确定性强易于调试和解释。缺点规则膨胀后难以维护灵活性差无法处理未预见的情况。基于模型的路由Model-Based 利用机器学习模型如分类器、甚至小型LLM来学习从上下文到最佳动作的映射。这是当前复杂Agent系统的趋势。# 使用嵌入Embedding和向量相似度进行路由 from sentence_transformers import SentenceTransformer import numpy as np class EmbeddingRouter: def __init__(self): self.model SentenceTransformer(all-MiniLM-L6-v2) # 预定义动作及其描述 self.action_descriptions { search_db: 在知识库中搜索相关信息, call_api: 调用外部API获取实时数据, generate_report: 生成分析总结报告 } # 预计算动作描述的向量 self.action_vectors {name: self.model.encode(desc) for name, desc in self.action_descriptions.items()} def route(self, context): # 将当前上下文如最新用户问题编码为向量 query_vector self.model.encode(context.latest_query) # 计算与所有动作的余弦相似度 similarities {} for action_name, action_vec in self.action_vectors.items(): cos_sim np.dot(query_vector, action_vec) / (np.linalg.norm(query_vector) * np.linalg.norm(action_vec)) similarities[action_name] cos_sim # 返回最相似的动作 return max(similarities, keysimilarities.get)实操心得模型路由的关键在于高质量的动作描述和上下文特征工程。动作描述要精准概括其功能和适用场景。上下文不能只扔进模型需要提炼出关键特征如用户意图、对话历史摘要、当前任务阶段等。混合路由Hybrid 结合规则和模型的优势。通常用规则处理明确、高优先级的场景如安全拦截、紧急转人工用模型处理复杂的、模糊的决策。也可以先用模型给出几个候选再用规则进行筛选和排序。2.2.2 路由表与策略链在工程实现上路由决策往往不是一步完成的。路由表Routing Table一个可配置的映射表将状态或意图映射到处理单元。适合规则路由便于运营人员修改。策略链Policy Chain按顺序执行多个路由策略。例如先经过一个“过滤器策略”排除非法请求再经过一个“优先级策略”识别VIP用户最后经过一个“负载均衡策略”分配Agent实例。每个策略都可以对路由结果进行修改或传递。注意路由决策本身也是有成本的。要避免在每次循环中都进行非常复杂的模型推理。可以考虑对路由结果进行缓存例如相同上下文指纹在短时间内路由到相同目标或者将路由决策的粒度放粗每N步或任务阶段变更时才重新路由。2.3 上下文ContextAgent的“记忆”与“工作台”上下文是贯穿循环和路由的“血液”。它承载了Agent任务执行过程中的所有状态和信息。一个设计良好的上下文管理机制是构建稳定、可维护Agent系统的基石。2.3.1 上下文的层次结构与内容上下文不是一个大杂烩字典而应该有清晰的结构会话上下文Session Context范围一次用户会话可能包含多轮对话的全局信息。内容用户ID、会话ID、创建时间、全局配置、用户长期偏好、安全令牌等。生命周期从会话开始到结束。任务上下文Task Context范围一个具体任务可能跨多个循环的信息。内容任务目标、任务参数、当前任务状态如进行中、暂停、完成、任务历史步骤记录、中间结果。生命周期从任务创建到任务终结成功、失败或取消。回合上下文Turn Context范围单次循环或单次用户交互的信息。内容本轮的用户输入、上轮Agent的输出、当前路由决策、本次执行的动作及其结果、临时变量。生命周期一次循环开始到结束。2.3.2 上下文的数据流与持久化上下文数据需要在循环、路由以及不同的处理单元间流动。数据流设计建议采用不可变Immutable或写时复制Copy-on-Write的理念。每次循环或动作执行产生的新结果应生成一个新的上下文版本或更新一个特定的分支而不是直接修改全局上下文。这有利于调试、回滚和实现“撤销”功能。class TurnContext: def __init__(self, previous_context, new_input): self.session_id previous_context.session_id # 继承 self.task_state previous_context.task_state.copy() # 浅拷贝或深拷贝视情况而定 self.current_input new_input self.actions_executed [] # 本轮新增 def record_action(self, action, result): self.actions_executed.append({action: action, result: result}) # 根据结果更新任务状态 self.task_state.update_from_result(result)持久化策略何时持久化关键节点必须持久化如任务开始、任务完成、每轮循环结束尤其是长循环、以及发生错误时。存储什么至少需要存储足以恢复任务状态的最小数据集。包括上下文的核心字段、循环的进度标识、路由决策历史等。存储后端根据延迟和一致性要求选择。Redis适合做高速缓存PostgreSQL/MongoDB适合做可靠存储。复杂的上下文对象可能需要序列化如JSON、MessagePack、Pickle后存储。实操心得为上下文对象设计一个版本号version字段。当你的Agent代码升级上下文结构可能变化版本号可以帮助你进行数据迁移或兼容性处理。2.3.3 上下文压缩与摘要随着对话或任务进行上下文会不断膨胀尤其是包含长对话历史或大量中间数据这会导致几个问题1超出模型token限制2降低路由和决策效率3增加存储和传输开销。因此上下文压缩Context Compression是高级Agent系统的必备技能。滑动窗口只保留最近N轮对话。最简单但可能丢失关键早期信息。关键信息提取使用一个轻量级模型或规则从历史上下文中提取出与当前任务最相关的实体、意图、事实结论丢弃冗余细节。增量式摘要在每轮或每N轮后动态生成一个不断更新的对话摘要。新的决策基于这个摘要和最新输入而不是全部原始历史。# 伪代码简单的增量摘要 class ConversationSummarizer: def __init__(self): self.summary def update_summary(self, new_dialogue_turn): # 将当前摘要和新对话轮次一起让LLM生成新的摘要 prompt f 现有摘要{self.summary} 最新一轮对话 用户{new_dialogue_turn.user} Agent{new_dialogue_turn.agent} 请基于以上信息更新对话摘要保留所有关键决策、事实和待办事项。 更新后的摘要 self.summary llm_invoke(prompt) return self.summary注意事项摘要的生成本身需要成本且可能存在信息损失。需要权衡摘要的更新频率和精度。对于关键业务信息即使被摘要了也应将其结构化后单独存储在任务上下文中。3. 实战构建一个具备核心运行机制的简易任务处理Agent理论说再多不如动手搭一个。我们来设计一个简易的“技术支持工单处理Agent”它需要理解用户问题自动尝试一些修复步骤如果不行则路由给人工客服。3.1 系统架构与组件设计我们将系统分为以下模块主循环引擎MainLoopEngine控制整体执行流程。上下文管理器ContextManager创建、更新、持久化上下文。路由决策器Router基于上下文决定下一步。动作执行器ActionExecutor执行具体的工具调用或子Agent任务。持久化存储Storage用于保存上下文和会话状态。3.2 核心代码实现解析3.2.1 上下文定义from dataclasses import dataclass, asdict, field from typing import Dict, List, Any, Optional from datetime import datetime import json dataclass class SessionContext: 会话级上下文 session_id: str user_id: str created_at: datetime attributes: Dict[str, Any] field(default_factorydict) # 存放用户偏好等 dataclass class TaskContext: 任务级上下文一个工单 task_id: str session_id: str description: str # 用户问题描述 status: str open # open, in_progress, resolved, escalated created_at: datetime field(default_factorydatetime.now) updated_at: datetime field(default_factorydatetime.now) max_steps: int 10 # 最大自动处理步数 current_step: int 0 history: List[Dict] field(default_factorylist) # 记录每一步操作 extracted_info: Dict[str, Any] field(default_factorydict) # 提取的关键信息如错误代码、设备型号 dataclass class TurnContext: 回合级上下文 turn_id: int task_context: TaskContext user_input: str router_decision: Optional[str] None action_result: Optional[Dict] None error: Optional[str] None def to_dict(self): # 方便序列化 return asdict(self)3.2.2 主循环引擎实现class MainLoopEngine: def __init__(self, router, action_executor, storage): self.router router self.action_executor action_executor self.storage storage def run_task(self, session_ctx: SessionContext, initial_query: str) - TaskContext: # 1. 初始化任务上下文 task_ctx TaskContext( task_idftask_{datetime.now().timestamp()}, session_idsession_ctx.session_id, descriptioninitial_query ) self.storage.save_task(task_ctx) # 2. 主循环 while task_ctx.status in [open, in_progress]: # 检查最大步数限制 if task_ctx.current_step task_ctx.max_steps: task_ctx.status escalated task_ctx.history.append({step: task_ctx.current_step, action: max_steps_reached, result: 自动处理步数超限转人工}) break # 创建本轮上下文 turn_ctx TurnContext( turn_idtask_ctx.current_step, task_contexttask_ctx, user_inputinitial_query if task_ctx.current_step 0 else # 后续轮次可能无新输入 ) # 3. 路由决策 try: next_action self.router.decide(turn_ctx) turn_ctx.router_decision next_action except Exception as e: turn_ctx.error f路由决策失败: {e} task_ctx.status escalated self._record_turn(task_ctx, turn_ctx) break # 4. 执行动作 try: result self.action_executor.execute(next_action, turn_ctx) turn_ctx.action_result result # 根据结果更新任务状态 self._update_task_from_result(task_ctx, result) except Exception as e: turn_ctx.error f动作执行失败: {e} # 执行失败可以考虑重试或直接升级 task_ctx.status escalated self._record_turn(task_ctx, turn_ctx) break # 5. 记录本轮并更新循环状态 self._record_turn(task_ctx, turn_ctx) task_ctx.current_step 1 task_ctx.updated_at datetime.now() # 6. 持久化任务状态检查点 if task_ctx.current_step % 3 0: # 每3步持久化一次 self.storage.save_task(task_ctx) # 循环结束最终持久化 self.storage.save_task(task_ctx) return task_ctx def _record_turn(self, task_ctx: TaskContext, turn_ctx: TurnContext): 记录单轮执行历史 record { step: turn_ctx.turn_id, decision: turn_ctx.router_decision, action_result: turn_ctx.action_result, error: turn_ctx.error, timestamp: datetime.now().isoformat() } task_ctx.history.append(record) # 也可以选择将详细的turn_ctx单独存储 self.storage.save_turn(turn_ctx) def _update_task_from_result(self, task_ctx: TaskContext, result: Dict): 根据动作执行结果更新任务状态 if result.get(status) resolved: task_ctx.status resolved elif result.get(suggest_escalation): task_ctx.status escalated # 更新提取的信息 if extracted_info in result: task_ctx.extracted_info.update(result[extracted_info])3.2.3 一个混合路由器的示例class HybridRouter: def __init__(self, rule_router, model_router, fallback_actionescalate_to_human): self.rule_router rule_router self.model_router model_router self.fallback_action fallback_action def decide(self, turn_ctx: TurnContext) - str: # 第一层安全与优先级规则硬性规则 if self._is_urgent_issue(turn_ctx): return immediate_human_escalation if self._contains_sensitive_info(turn_ctx): return mask_and_process # 第二层基于规则的快速路由明确场景 rule_based_action self.rule_router.decide(turn_ctx) if rule_based_action and rule_based_action ! unknown: return rule_based_action # 第三层基于模型的智能路由模糊场景 model_based_action self.model_router.decide(turn_ctx) if model_based_action: return model_based_action # 默认降级方案 return self.fallback_action def _is_urgent_issue(self, turn_ctx): # 判断是否为紧急问题例如包含“宕机”、“无法使用”等关键词且来自VIP用户 urgent_keywords [宕机, 崩溃, 完全无法] task_desc turn_ctx.task_context.description # 这里假设能从session_ctx获取用户等级简化处理 return any(kw in task_desc for kw in urgent_keywords) def _contains_sensitive_info(self, turn_ctx): # 简单示例检测是否包含疑似密码的信息 import re potential_pwd_pattern r(?i)(password|pwd|密码)[:]\s*\S return bool(re.search(potential_pwd_pattern, turn_ctx.user_input or turn_ctx.task_context.description))3.3 动作执行器的设计模式动作执行器负责将路由决策的“动作名称”转化为具体的操作。一个好的模式是使用“注册表”Registry。class ActionExecutor: def __init__(self): self._actions {} def register(self, name: str, action_func): self._actions[name] action_func def execute(self, action_name: str, turn_ctx: TurnContext) - Dict: if action_name not in self._actions: raise ValueError(f未知动作: {action_name}) try: # 执行动作并传入当前上下文 result self._actions[action_name](turn_ctx) # 确保返回结果包含标准字段 result.setdefault(status, success) return result except Exception as e: # 记录详细的执行错误 return {status: error, message: str(e), action: action_name} # 使用示例 executor ActionExecutor() executor.register(search_knowledge_base) def search_kb_action(turn_ctx): query turn_ctx.task_context.description # 模拟搜索知识库 search_results simulate_kb_search(query) # 尝试从结果中提取解决方案 solution extract_solution(search_results) return { status: resolved if solution else no_match, data: search_results, proposed_solution: solution, extracted_info: {query_topic: query[:50]} # 记录提取的信息 } executor.register(run_diagnostic_script) def run_diagnostic_action(turn_ctx): # 模拟运行一个诊断脚本 diagnostic_result run_script(standard_diagnostic) if diagnostic_result.get(issues_found): return {status: in_progress, fix_steps: diagnostic_result[suggested_fixes]} else: return {status: no_issue_detected, suggest_escalation: True}4. 生产环境下的挑战与调优实录把Demo跑起来只是第一步真正上线后各种意想不到的问题才会浮现。下面分享几个我实践中遇到的典型挑战和解决思路。4.1 循环失控与死锁预防问题场景一个用于处理文档的Agent在“解析-提取-验证”的循环中因为某个边缘文档格式解析异常导致验证永远无法通过循环卡死直到达到最大步数才超时退出浪费资源且体验差。根因分析循环的停止条件过于依赖业务结果如“验证通过”未考虑异常状态。缺乏对循环内“无进展”状态的检测。解决方案设置多维停止条件除了业务成功状态和最大步数增加“无进展检测”。def should_stop(task_ctx, turn_history): # 条件1: 业务成功 if task_ctx.status resolved: return True # 条件2: 达到最大步数 if task_ctx.current_step task_ctx.max_steps: return True # 条件3: 检测无进展循环 (最近N步结果高度相似或错误相同) recent_steps turn_history[-5:] if len(turn_history) 5 else turn_history if len(recent_steps) 3: last_three_actions [step.get(decision) for step in recent_steps[-3:]] last_three_errors [step.get(error) for step in recent_steps[-3:]] # 如果连续三步路由到同一个无效动作或产生相同错误 if (len(set(last_three_actions)) 1 and last_three_actions[0] in [action_a, action_b]) or \ (len(set(last_three_errors)) 1 and last_three_errors[0] is not None): return True # 触发停止并标记为需人工干预 return False引入看门狗Watchdog线程在主循环外启动一个监控线程检查单次循环的执行时间。如果某次循环执行时间远超历史平均时间例如3个标准差以外则向主线程发送中断信号保存当前状态并标记为“超时异常”。4.2 路由决策的准确性与效率平衡问题场景使用一个大型语言模型LLM作为路由决策器虽然准确率高但每个请求都调用LLM导致响应延迟高、成本昂贵。根因分析路由决策的粒度太细且未利用缓存。解决方案分层路由与缓存第一层意图快速分类。使用一个轻量级文本分类模型如FastText、小规模BERT或关键词匹配将请求分到几个大的意图桶如“查询”、“操作”、“故障”。这一步速度极快可以覆盖80%的常见请求。第二层桶内精细路由。只有进入“故障”等复杂桶的请求才触发更精细的LLM路由或规则路由。同时对路由结果进行缓存。缓存的Key可以是“意图桶用户输入文本的哈希”或“上下文特征指纹”。设置合理的TTL如5分钟在短时间内相同问题无需重复计算路由。路由决策预热与异步更新对于已知的、高频的任务模板可以在系统启动时或低峰期预计算其路由结果存入缓存。对于模型路由可以异步定期更新模型而不影响在线推理路径。4.3 上下文膨胀与信息丢失的权衡问题场景一个多轮对话Agent随着对话轮次增加上下文越来越长最终超出模型Token限制。使用简单的滑动窗口又丢失了对话开头约定的关键约束条件。根因分析上下文管理策略过于简单对所有信息一视同仁。解决方案结构化上下文与重要性标注不要将所有历史都作为非结构化的文本堆砌。设计结构化的上下文对象将信息分类系统指令高优先级Agent的角色、核心约束、对话目标。这部分应始终保留。关键事实与参数中优先级用户明确提供的实体、数字、选择等。可以提取出来放在一个“事实表”中。对话历史低优先级具体的每一轮问答。这部分是压缩的主要对象。动态摘要与关键信息提取在每轮对话后不是保存全部原文而是用一个小模型或提示词工程生成一个增量摘要。这个摘要会融合上一轮的摘要和本轮的新内容。同时运行一个命名实体识别NER或信息提取流程将本轮出现的新关键信息如订单号、日期、产品型号提取出来添加到“事实表”中。后续的路由和动作执行主要参考“系统指令”、“事实表”和“最新摘要”而非全部历史。只有当需要深度理解某段历史时才去查询详细的、经过索引的原始记录如果保存了的话。class CompressingContextManager: def __init__(self, max_raw_turns10): self.system_instructions self.fact_table {} # 键值对存储关键事实 self.dialogue_summary self.raw_turns [] # 保留最近N轮原始记录用于追溯 self.max_raw_turns max_raw_turns def add_turn(self, user_input, agent_response): # 1. 保存原始记录滑动窗口 self.raw_turns.append((user_input, agent_response)) if len(self.raw_turns) self.max_raw_turns: self.raw_turns.pop(0) # 2. 提取关键事实 new_facts extract_facts(user_input, agent_response) self.fact_table.update(new_facts) # 3. 更新摘要 update_prompt f当前摘要{self.dialogue_summary}\n新对话用户说{user_input}Agent回复{agent_response}\n请生成更新的摘要。 self.dialogue_summary call_llm_for_summary(update_prompt) def get_compressed_context_for_agent(self): 提供给Agent模型使用的压缩后上下文 return { system: self.system_instructions, facts: json.dumps(self.fact_table, ensure_asciiFalse), summary: self.dialogue_summary, recent_raw: self.raw_turns[-2:] if self.raw_turns else [] # 提供最近一两轮原文供参考 }4.4 调试与可观测性建设当Agent行为不符合预期时如何快速定位是循环、路由还是上下文的问题结构化日志不要在代码里随意打印print。为每个循环步骤、路由决策、上下文更新、动作执行记录结构化的日志。日志应包含时间戳、会话ID、任务ID、步骤ID、组件名如Router、ActionX、日志级别、关键数据如路由决策结果、上下文快照的哈希、动作执行耗时和错误信息。# 使用结构化日志库如structlog或json logger import structlog logger structlog.get_logger() def some_router_function(context): log logger.bind(task_idcontext.task_id, turncontext.current_step) log.info(router.started, input_snippetcontext.user_input[:100]) try: decision make_decision(context) log.info(router.decision_made, decisiondecision, confidence0.85) return decision except Exception as e: log.error(router.failed, errorstr(e), context_snapshotcontext.to_dict()) raise分布式追踪在微服务或分布式Agent架构中使用OpenTelemetry等工具为每个用户请求注入追踪ID。这个ID贯穿整个调用链从接收请求到循环的每一步到路由到子服务调用让你能在仪表盘上清晰地看到一个请求的完整生命周期和耗时瓶颈。上下文快照与回放将每个关键步骤尤其是路由决策前后和动作执行前后的完整上下文对象序列化后存储到可查询的存储中如Elasticsearch。当出现问题时你可以通过任务ID检索出完整的上下文历史在本地或测试环境“回放”该任务精确复现问题。路由决策的可解释性对于基于模型的路由不仅要输出决策还要输出置信度分数和如果可能主要依据的特征。这能帮助你在调试时判断是模型不准还是输入特征有问题。5. 性能优化与扩展性考量当你的Agent系统从原型走向生产服务大量并发用户时性能和扩展性就成为必须面对的问题。5.1 运行时性能优化循环异步化如果Agent的某些动作是I/O密集型的如调用外部API、查询数据库不要让主循环同步等待。可以将动作提交到任务队列如Celery、RabbitMQ主循环只负责生成任务和监听结果。这样单个Agent实例可以同时管理多个任务的执行流极大提高吞吐量。路由决策批处理对于请求量大的场景可以将短时间内的一批路由请求收集起来批量发送给模型进行推理如果模型支持批量推理这比逐个请求效率高得多。上下文缓存会话级和任务级的上下文是高频访问对象。使用内存缓存如Redis存储活跃的上下文对象避免每次循环都从数据库读取。注意设计合理的缓存失效和回写策略。动作执行结果缓存对于一些幂等的、结果相对稳定的动作如“根据城市名查询天气”可以对其结果进行缓存。路由决策后先查缓存命中则直接返回避免重复执行。5.2 系统扩展性设计无状态与有状态组件的分离无状态组件路由决策器特别是模型推理部分、某些工具函数。这些组件可以轻松地水平扩展用多个实例加负载均衡即可。有状态组件循环引擎和上下文管理器是典型的有状态组件。一个正在运行的任务其循环状态和上下文必须由同一个进程或线程来维护不能随意迁移。对此常见的模式是采用分片Sharding或基于会话的粘性Sticky Session。例如通过会话ID的哈希值决定由哪个后端实例来处理该会话的所有后续请求。水平扩展循环引擎由于循环引擎有状态扩展起来更复杂。一种架构是采用“主从”模式或“事件溯源”模式。事件溯源模式将Agent的每一次状态变更如“路由决策A”、“执行动作B成功”、“更新上下文C”都记录为一个不可变的事件Event并持久化到事件流如Kafka中。循环引擎本身可以设计得相对轻量它读取事件流根据当前状态和事件计算出新状态并生成新的事件。这样多个循环引擎实例可以消费同一个事件流的不同分区状态的计算逻辑是确定的最终状态由事件序列决定便于扩展和故障恢复。容错与高可用检查点Checkpointing循环引擎定期将任务上下文和循环进度持久化到可靠的存储中。当实例故障时调度器可以将该任务重新分配给另一个健康的实例新实例从最新的检查点加载状态并继续执行。优雅降级当路由模型服务或某个关键动作服务不可用时系统应能降级到备用方案。例如路由降级到基于规则的简单版本动作服务不可用则返回“服务暂不可用请稍后重试”并记录任务状态为“等待重试”。构建一个健壮、高效的Agent运行时系统是一个持续迭代和平衡的过程。从清晰理解循环、路由、上下文这三个核心机制开始在设计和编码时始终想着状态如何流转、决策如何做出、信息如何保存就能避开很多深坑。 智能体运行时核心机制：循环、路由与上下文的设计与工程实践
1. 项目概述从“笔记”到“工程实践”最近在整理Harness Engineering相关的技术文档时我发现关于Agent运行时核心机制的讨论尤其是循环、路由与上下文这三块是很多开发者从“会用框架”到“理解框架”的关键门槛。网上能找到的资料要么过于零散只讲某个API怎么调用要么过于理论堆砌一堆架构图却不说清楚代码到底怎么跑起来的。这让我觉得有必要结合自己踩过的坑把这块“硬骨头”啃下来写一篇能直接指导编码的深度解析。所谓Harness Engineering你可以把它理解为一套用于构建、管理和控制智能体Agent的工程化框架与最佳实践。它关注的不是某个单一的算法模型而是如何让多个Agent协同、稳定、高效地完成复杂任务。在这个体系里Agent运行时就是那个让智能体“活”起来、能够感知、决策并执行的核心引擎。而驱动这个引擎的三大核心机制正是循环Loop、路由Routing和上下文Context。理解它们你才能真的驾驭Agent而不是被框架牵着鼻子走。这篇文章适合谁呢如果你正在或打算开发基于Agent的应用无论是自动化工作流、智能客服还是复杂的决策系统并且已经过了“Hello World”阶段开始头疼于Agent的状态管理、任务调度和信息流转问题那么这里面的内容就是为你准备的。我会尽量避开空泛的概念用具体的代码段、设计抉择背后的“为什么”以及我实际调试中总结出的“坑点”来展开。2. 核心机制深度拆解循环、路由与上下文的角色与联动在深入细节之前我们必须建立一个顶层的认知模型这三个机制不是孤立的它们共同构成了Agent运行时的一个完整“心跳周期”。想象一下一个处理用户咨询的客服Agent。它拿到用户问题上下文需要决定这个问题属于售前、售后还是技术故障路由然后根据决定调用相应的子能力或工具去执行执行后产生新的结果并评估是否还需要进一步追问或可以结束循环。这个过程周而复始。2.1 循环LoopAgent执行的状态机与驱动力循环机制是Agent运行的“节拍器”。它决定了Agent在单次触发后如何推进其内部状态何时停止。最简单的循环就是“执行一次就结束”One-Shot。但对于复杂任务我们需要更强大的循环模式。2.1.1 常见循环模式与实践选择While-Loop条件循环 这是最常用、最灵活的循环模式。它基于一个或多个条件来决定是否继续迭代。例如一个数据分析Agent可能会循环执行“查询-分析-提炼”步骤直到分析结果的置信度达到某个阈值或者用户明确要求停止。# 伪代码示例基于条件的While循环 context initialize_context(user_query) while not should_stop(context): # 1. 路由决策根据当前context决定下一步动作 action routing_engine.decide(context) # 2. 执行动作 result execute_action(action, context) # 3. 更新上下文 context.update(result, action) # 4. 评估停止条件例如任务完成、达到最大步数、用户中断 context.evaluate_stop_condition()实操心得should_stop函数的设计是核心。不要只依赖单一条件如任务标记完成最好结合多个维度最大迭代次数防止死循环、用户主动取消信号、连续多次迭代未产生有效进展等。我通常会设置一个默认的最大步数比如50步作为安全网。For-Loop有限循环 适用于步骤明确、次数固定的任务。比如一个Agent需要依次检查系统的A、B、C三个服务状态。checklist [检查数据库连接, 验证API网关, 扫描日志错误] for task in checklist: context.set_current_task(task) result execute_standard_check(task, context) context.record_check_result(task, result)注意事项即使在For-Loop中也要为每个步骤设计异常处理。某个步骤的失败不应导致整个Agent崩溃而应该被捕获并记录到上下文中供后续步骤或最终汇总时参考。递归循环Recursive Loop 当任务可以分解为同构的子任务时使用。例如一个文件整理Agent遇到文件夹就递归处理其中的文件。核心挑战上下文的管理。递归每一层都需要有自己的上下文“快照”同时又能访问到必要的父级信息。需要清晰界定上下文的作用域和继承关系否则很容易造成信息污染或丢失。2.1.2 循环控制的高级技巧暂停与恢复Pause/Resume对于长时任务Agent需要支持暂停。关键在于将完整的运行时状态包括循环索引、当前上下文、中间结果序列化并持久化。恢复时再反序列化从断点继续。这通常需要框架层面的支持。并行循环当多个子任务相互独立时可以使用并行循环来提升效率。但要注意共享上下文资源的线程安全问题。一个稳妥的做法是采用“复制-合并”策略每个并行任务处理上下文的一个副本执行完毕后再将结果合并回主上下文。循环超时与看门狗Watchdog必须为每个循环设置超时机制。一个独立的看门狗线程可以监控主循环的执行时间一旦超时就触发中断流程保存现场并上报错误避免Agent“卡死”。2.2 路由RoutingAgent的决策中枢与流量控制器如果说循环是“节奏”那么路由就是“方向”。它负责在运行时根据当前上下文动态决定下一步该执行哪个动作、调用哪个工具、或者将任务移交给哪个更专业的子Agent。2.2.1 路由策略解析路由的核心是一个决策函数Input(Current_Context) - Output(Next_Action/Agent)。基于规则的路由Rule-Based 最简单直接的方式。通过预定义的if-else或决策树来路由。def rule_based_router(context): if 退款 in context.user_intent: return refund_agent elif 技术故障 in context.sentiment and context.user_tier VIP: return premium_support_agent else: return general_support_agent优点确定性强易于调试和解释。缺点规则膨胀后难以维护灵活性差无法处理未预见的情况。基于模型的路由Model-Based 利用机器学习模型如分类器、甚至小型LLM来学习从上下文到最佳动作的映射。这是当前复杂Agent系统的趋势。# 使用嵌入Embedding和向量相似度进行路由 from sentence_transformers import SentenceTransformer import numpy as np class EmbeddingRouter: def __init__(self): self.model SentenceTransformer(all-MiniLM-L6-v2) # 预定义动作及其描述 self.action_descriptions { search_db: 在知识库中搜索相关信息, call_api: 调用外部API获取实时数据, generate_report: 生成分析总结报告 } # 预计算动作描述的向量 self.action_vectors {name: self.model.encode(desc) for name, desc in self.action_descriptions.items()} def route(self, context): # 将当前上下文如最新用户问题编码为向量 query_vector self.model.encode(context.latest_query) # 计算与所有动作的余弦相似度 similarities {} for action_name, action_vec in self.action_vectors.items(): cos_sim np.dot(query_vector, action_vec) / (np.linalg.norm(query_vector) * np.linalg.norm(action_vec)) similarities[action_name] cos_sim # 返回最相似的动作 return max(similarities, keysimilarities.get)实操心得模型路由的关键在于高质量的动作描述和上下文特征工程。动作描述要精准概括其功能和适用场景。上下文不能只扔进模型需要提炼出关键特征如用户意图、对话历史摘要、当前任务阶段等。混合路由Hybrid 结合规则和模型的优势。通常用规则处理明确、高优先级的场景如安全拦截、紧急转人工用模型处理复杂的、模糊的决策。也可以先用模型给出几个候选再用规则进行筛选和排序。2.2.2 路由表与策略链在工程实现上路由决策往往不是一步完成的。路由表Routing Table一个可配置的映射表将状态或意图映射到处理单元。适合规则路由便于运营人员修改。策略链Policy Chain按顺序执行多个路由策略。例如先经过一个“过滤器策略”排除非法请求再经过一个“优先级策略”识别VIP用户最后经过一个“负载均衡策略”分配Agent实例。每个策略都可以对路由结果进行修改或传递。注意路由决策本身也是有成本的。要避免在每次循环中都进行非常复杂的模型推理。可以考虑对路由结果进行缓存例如相同上下文指纹在短时间内路由到相同目标或者将路由决策的粒度放粗每N步或任务阶段变更时才重新路由。2.3 上下文ContextAgent的“记忆”与“工作台”上下文是贯穿循环和路由的“血液”。它承载了Agent任务执行过程中的所有状态和信息。一个设计良好的上下文管理机制是构建稳定、可维护Agent系统的基石。2.3.1 上下文的层次结构与内容上下文不是一个大杂烩字典而应该有清晰的结构会话上下文Session Context范围一次用户会话可能包含多轮对话的全局信息。内容用户ID、会话ID、创建时间、全局配置、用户长期偏好、安全令牌等。生命周期从会话开始到结束。任务上下文Task Context范围一个具体任务可能跨多个循环的信息。内容任务目标、任务参数、当前任务状态如进行中、暂停、完成、任务历史步骤记录、中间结果。生命周期从任务创建到任务终结成功、失败或取消。回合上下文Turn Context范围单次循环或单次用户交互的信息。内容本轮的用户输入、上轮Agent的输出、当前路由决策、本次执行的动作及其结果、临时变量。生命周期一次循环开始到结束。2.3.2 上下文的数据流与持久化上下文数据需要在循环、路由以及不同的处理单元间流动。数据流设计建议采用不可变Immutable或写时复制Copy-on-Write的理念。每次循环或动作执行产生的新结果应生成一个新的上下文版本或更新一个特定的分支而不是直接修改全局上下文。这有利于调试、回滚和实现“撤销”功能。class TurnContext: def __init__(self, previous_context, new_input): self.session_id previous_context.session_id # 继承 self.task_state previous_context.task_state.copy() # 浅拷贝或深拷贝视情况而定 self.current_input new_input self.actions_executed [] # 本轮新增 def record_action(self, action, result): self.actions_executed.append({action: action, result: result}) # 根据结果更新任务状态 self.task_state.update_from_result(result)持久化策略何时持久化关键节点必须持久化如任务开始、任务完成、每轮循环结束尤其是长循环、以及发生错误时。存储什么至少需要存储足以恢复任务状态的最小数据集。包括上下文的核心字段、循环的进度标识、路由决策历史等。存储后端根据延迟和一致性要求选择。Redis适合做高速缓存PostgreSQL/MongoDB适合做可靠存储。复杂的上下文对象可能需要序列化如JSON、MessagePack、Pickle后存储。实操心得为上下文对象设计一个版本号version字段。当你的Agent代码升级上下文结构可能变化版本号可以帮助你进行数据迁移或兼容性处理。2.3.3 上下文压缩与摘要随着对话或任务进行上下文会不断膨胀尤其是包含长对话历史或大量中间数据这会导致几个问题1超出模型token限制2降低路由和决策效率3增加存储和传输开销。因此上下文压缩Context Compression是高级Agent系统的必备技能。滑动窗口只保留最近N轮对话。最简单但可能丢失关键早期信息。关键信息提取使用一个轻量级模型或规则从历史上下文中提取出与当前任务最相关的实体、意图、事实结论丢弃冗余细节。增量式摘要在每轮或每N轮后动态生成一个不断更新的对话摘要。新的决策基于这个摘要和最新输入而不是全部原始历史。# 伪代码简单的增量摘要 class ConversationSummarizer: def __init__(self): self.summary def update_summary(self, new_dialogue_turn): # 将当前摘要和新对话轮次一起让LLM生成新的摘要 prompt f 现有摘要{self.summary} 最新一轮对话 用户{new_dialogue_turn.user} Agent{new_dialogue_turn.agent} 请基于以上信息更新对话摘要保留所有关键决策、事实和待办事项。 更新后的摘要 self.summary llm_invoke(prompt) return self.summary注意事项摘要的生成本身需要成本且可能存在信息损失。需要权衡摘要的更新频率和精度。对于关键业务信息即使被摘要了也应将其结构化后单独存储在任务上下文中。3. 实战构建一个具备核心运行机制的简易任务处理Agent理论说再多不如动手搭一个。我们来设计一个简易的“技术支持工单处理Agent”它需要理解用户问题自动尝试一些修复步骤如果不行则路由给人工客服。3.1 系统架构与组件设计我们将系统分为以下模块主循环引擎MainLoopEngine控制整体执行流程。上下文管理器ContextManager创建、更新、持久化上下文。路由决策器Router基于上下文决定下一步。动作执行器ActionExecutor执行具体的工具调用或子Agent任务。持久化存储Storage用于保存上下文和会话状态。3.2 核心代码实现解析3.2.1 上下文定义from dataclasses import dataclass, asdict, field from typing import Dict, List, Any, Optional from datetime import datetime import json dataclass class SessionContext: 会话级上下文 session_id: str user_id: str created_at: datetime attributes: Dict[str, Any] field(default_factorydict) # 存放用户偏好等 dataclass class TaskContext: 任务级上下文一个工单 task_id: str session_id: str description: str # 用户问题描述 status: str open # open, in_progress, resolved, escalated created_at: datetime field(default_factorydatetime.now) updated_at: datetime field(default_factorydatetime.now) max_steps: int 10 # 最大自动处理步数 current_step: int 0 history: List[Dict] field(default_factorylist) # 记录每一步操作 extracted_info: Dict[str, Any] field(default_factorydict) # 提取的关键信息如错误代码、设备型号 dataclass class TurnContext: 回合级上下文 turn_id: int task_context: TaskContext user_input: str router_decision: Optional[str] None action_result: Optional[Dict] None error: Optional[str] None def to_dict(self): # 方便序列化 return asdict(self)3.2.2 主循环引擎实现class MainLoopEngine: def __init__(self, router, action_executor, storage): self.router router self.action_executor action_executor self.storage storage def run_task(self, session_ctx: SessionContext, initial_query: str) - TaskContext: # 1. 初始化任务上下文 task_ctx TaskContext( task_idftask_{datetime.now().timestamp()}, session_idsession_ctx.session_id, descriptioninitial_query ) self.storage.save_task(task_ctx) # 2. 主循环 while task_ctx.status in [open, in_progress]: # 检查最大步数限制 if task_ctx.current_step task_ctx.max_steps: task_ctx.status escalated task_ctx.history.append({step: task_ctx.current_step, action: max_steps_reached, result: 自动处理步数超限转人工}) break # 创建本轮上下文 turn_ctx TurnContext( turn_idtask_ctx.current_step, task_contexttask_ctx, user_inputinitial_query if task_ctx.current_step 0 else # 后续轮次可能无新输入 ) # 3. 路由决策 try: next_action self.router.decide(turn_ctx) turn_ctx.router_decision next_action except Exception as e: turn_ctx.error f路由决策失败: {e} task_ctx.status escalated self._record_turn(task_ctx, turn_ctx) break # 4. 执行动作 try: result self.action_executor.execute(next_action, turn_ctx) turn_ctx.action_result result # 根据结果更新任务状态 self._update_task_from_result(task_ctx, result) except Exception as e: turn_ctx.error f动作执行失败: {e} # 执行失败可以考虑重试或直接升级 task_ctx.status escalated self._record_turn(task_ctx, turn_ctx) break # 5. 记录本轮并更新循环状态 self._record_turn(task_ctx, turn_ctx) task_ctx.current_step 1 task_ctx.updated_at datetime.now() # 6. 持久化任务状态检查点 if task_ctx.current_step % 3 0: # 每3步持久化一次 self.storage.save_task(task_ctx) # 循环结束最终持久化 self.storage.save_task(task_ctx) return task_ctx def _record_turn(self, task_ctx: TaskContext, turn_ctx: TurnContext): 记录单轮执行历史 record { step: turn_ctx.turn_id, decision: turn_ctx.router_decision, action_result: turn_ctx.action_result, error: turn_ctx.error, timestamp: datetime.now().isoformat() } task_ctx.history.append(record) # 也可以选择将详细的turn_ctx单独存储 self.storage.save_turn(turn_ctx) def _update_task_from_result(self, task_ctx: TaskContext, result: Dict): 根据动作执行结果更新任务状态 if result.get(status) resolved: task_ctx.status resolved elif result.get(suggest_escalation): task_ctx.status escalated # 更新提取的信息 if extracted_info in result: task_ctx.extracted_info.update(result[extracted_info])3.2.3 一个混合路由器的示例class HybridRouter: def __init__(self, rule_router, model_router, fallback_actionescalate_to_human): self.rule_router rule_router self.model_router model_router self.fallback_action fallback_action def decide(self, turn_ctx: TurnContext) - str: # 第一层安全与优先级规则硬性规则 if self._is_urgent_issue(turn_ctx): return immediate_human_escalation if self._contains_sensitive_info(turn_ctx): return mask_and_process # 第二层基于规则的快速路由明确场景 rule_based_action self.rule_router.decide(turn_ctx) if rule_based_action and rule_based_action ! unknown: return rule_based_action # 第三层基于模型的智能路由模糊场景 model_based_action self.model_router.decide(turn_ctx) if model_based_action: return model_based_action # 默认降级方案 return self.fallback_action def _is_urgent_issue(self, turn_ctx): # 判断是否为紧急问题例如包含“宕机”、“无法使用”等关键词且来自VIP用户 urgent_keywords [宕机, 崩溃, 完全无法] task_desc turn_ctx.task_context.description # 这里假设能从session_ctx获取用户等级简化处理 return any(kw in task_desc for kw in urgent_keywords) def _contains_sensitive_info(self, turn_ctx): # 简单示例检测是否包含疑似密码的信息 import re potential_pwd_pattern r(?i)(password|pwd|密码)[:]\s*\S return bool(re.search(potential_pwd_pattern, turn_ctx.user_input or turn_ctx.task_context.description))3.3 动作执行器的设计模式动作执行器负责将路由决策的“动作名称”转化为具体的操作。一个好的模式是使用“注册表”Registry。class ActionExecutor: def __init__(self): self._actions {} def register(self, name: str, action_func): self._actions[name] action_func def execute(self, action_name: str, turn_ctx: TurnContext) - Dict: if action_name not in self._actions: raise ValueError(f未知动作: {action_name}) try: # 执行动作并传入当前上下文 result self._actions[action_name](turn_ctx) # 确保返回结果包含标准字段 result.setdefault(status, success) return result except Exception as e: # 记录详细的执行错误 return {status: error, message: str(e), action: action_name} # 使用示例 executor ActionExecutor() executor.register(search_knowledge_base) def search_kb_action(turn_ctx): query turn_ctx.task_context.description # 模拟搜索知识库 search_results simulate_kb_search(query) # 尝试从结果中提取解决方案 solution extract_solution(search_results) return { status: resolved if solution else no_match, data: search_results, proposed_solution: solution, extracted_info: {query_topic: query[:50]} # 记录提取的信息 } executor.register(run_diagnostic_script) def run_diagnostic_action(turn_ctx): # 模拟运行一个诊断脚本 diagnostic_result run_script(standard_diagnostic) if diagnostic_result.get(issues_found): return {status: in_progress, fix_steps: diagnostic_result[suggested_fixes]} else: return {status: no_issue_detected, suggest_escalation: True}4. 生产环境下的挑战与调优实录把Demo跑起来只是第一步真正上线后各种意想不到的问题才会浮现。下面分享几个我实践中遇到的典型挑战和解决思路。4.1 循环失控与死锁预防问题场景一个用于处理文档的Agent在“解析-提取-验证”的循环中因为某个边缘文档格式解析异常导致验证永远无法通过循环卡死直到达到最大步数才超时退出浪费资源且体验差。根因分析循环的停止条件过于依赖业务结果如“验证通过”未考虑异常状态。缺乏对循环内“无进展”状态的检测。解决方案设置多维停止条件除了业务成功状态和最大步数增加“无进展检测”。def should_stop(task_ctx, turn_history): # 条件1: 业务成功 if task_ctx.status resolved: return True # 条件2: 达到最大步数 if task_ctx.current_step task_ctx.max_steps: return True # 条件3: 检测无进展循环 (最近N步结果高度相似或错误相同) recent_steps turn_history[-5:] if len(turn_history) 5 else turn_history if len(recent_steps) 3: last_three_actions [step.get(decision) for step in recent_steps[-3:]] last_three_errors [step.get(error) for step in recent_steps[-3:]] # 如果连续三步路由到同一个无效动作或产生相同错误 if (len(set(last_three_actions)) 1 and last_three_actions[0] in [action_a, action_b]) or \ (len(set(last_three_errors)) 1 and last_three_errors[0] is not None): return True # 触发停止并标记为需人工干预 return False引入看门狗Watchdog线程在主循环外启动一个监控线程检查单次循环的执行时间。如果某次循环执行时间远超历史平均时间例如3个标准差以外则向主线程发送中断信号保存当前状态并标记为“超时异常”。4.2 路由决策的准确性与效率平衡问题场景使用一个大型语言模型LLM作为路由决策器虽然准确率高但每个请求都调用LLM导致响应延迟高、成本昂贵。根因分析路由决策的粒度太细且未利用缓存。解决方案分层路由与缓存第一层意图快速分类。使用一个轻量级文本分类模型如FastText、小规模BERT或关键词匹配将请求分到几个大的意图桶如“查询”、“操作”、“故障”。这一步速度极快可以覆盖80%的常见请求。第二层桶内精细路由。只有进入“故障”等复杂桶的请求才触发更精细的LLM路由或规则路由。同时对路由结果进行缓存。缓存的Key可以是“意图桶用户输入文本的哈希”或“上下文特征指纹”。设置合理的TTL如5分钟在短时间内相同问题无需重复计算路由。路由决策预热与异步更新对于已知的、高频的任务模板可以在系统启动时或低峰期预计算其路由结果存入缓存。对于模型路由可以异步定期更新模型而不影响在线推理路径。4.3 上下文膨胀与信息丢失的权衡问题场景一个多轮对话Agent随着对话轮次增加上下文越来越长最终超出模型Token限制。使用简单的滑动窗口又丢失了对话开头约定的关键约束条件。根因分析上下文管理策略过于简单对所有信息一视同仁。解决方案结构化上下文与重要性标注不要将所有历史都作为非结构化的文本堆砌。设计结构化的上下文对象将信息分类系统指令高优先级Agent的角色、核心约束、对话目标。这部分应始终保留。关键事实与参数中优先级用户明确提供的实体、数字、选择等。可以提取出来放在一个“事实表”中。对话历史低优先级具体的每一轮问答。这部分是压缩的主要对象。动态摘要与关键信息提取在每轮对话后不是保存全部原文而是用一个小模型或提示词工程生成一个增量摘要。这个摘要会融合上一轮的摘要和本轮的新内容。同时运行一个命名实体识别NER或信息提取流程将本轮出现的新关键信息如订单号、日期、产品型号提取出来添加到“事实表”中。后续的路由和动作执行主要参考“系统指令”、“事实表”和“最新摘要”而非全部历史。只有当需要深度理解某段历史时才去查询详细的、经过索引的原始记录如果保存了的话。class CompressingContextManager: def __init__(self, max_raw_turns10): self.system_instructions self.fact_table {} # 键值对存储关键事实 self.dialogue_summary self.raw_turns [] # 保留最近N轮原始记录用于追溯 self.max_raw_turns max_raw_turns def add_turn(self, user_input, agent_response): # 1. 保存原始记录滑动窗口 self.raw_turns.append((user_input, agent_response)) if len(self.raw_turns) self.max_raw_turns: self.raw_turns.pop(0) # 2. 提取关键事实 new_facts extract_facts(user_input, agent_response) self.fact_table.update(new_facts) # 3. 更新摘要 update_prompt f当前摘要{self.dialogue_summary}\n新对话用户说{user_input}Agent回复{agent_response}\n请生成更新的摘要。 self.dialogue_summary call_llm_for_summary(update_prompt) def get_compressed_context_for_agent(self): 提供给Agent模型使用的压缩后上下文 return { system: self.system_instructions, facts: json.dumps(self.fact_table, ensure_asciiFalse), summary: self.dialogue_summary, recent_raw: self.raw_turns[-2:] if self.raw_turns else [] # 提供最近一两轮原文供参考 }4.4 调试与可观测性建设当Agent行为不符合预期时如何快速定位是循环、路由还是上下文的问题结构化日志不要在代码里随意打印print。为每个循环步骤、路由决策、上下文更新、动作执行记录结构化的日志。日志应包含时间戳、会话ID、任务ID、步骤ID、组件名如Router、ActionX、日志级别、关键数据如路由决策结果、上下文快照的哈希、动作执行耗时和错误信息。# 使用结构化日志库如structlog或json logger import structlog logger structlog.get_logger() def some_router_function(context): log logger.bind(task_idcontext.task_id, turncontext.current_step) log.info(router.started, input_snippetcontext.user_input[:100]) try: decision make_decision(context) log.info(router.decision_made, decisiondecision, confidence0.85) return decision except Exception as e: log.error(router.failed, errorstr(e), context_snapshotcontext.to_dict()) raise分布式追踪在微服务或分布式Agent架构中使用OpenTelemetry等工具为每个用户请求注入追踪ID。这个ID贯穿整个调用链从接收请求到循环的每一步到路由到子服务调用让你能在仪表盘上清晰地看到一个请求的完整生命周期和耗时瓶颈。上下文快照与回放将每个关键步骤尤其是路由决策前后和动作执行前后的完整上下文对象序列化后存储到可查询的存储中如Elasticsearch。当出现问题时你可以通过任务ID检索出完整的上下文历史在本地或测试环境“回放”该任务精确复现问题。路由决策的可解释性对于基于模型的路由不仅要输出决策还要输出置信度分数和如果可能主要依据的特征。这能帮助你在调试时判断是模型不准还是输入特征有问题。5. 性能优化与扩展性考量当你的Agent系统从原型走向生产服务大量并发用户时性能和扩展性就成为必须面对的问题。5.1 运行时性能优化循环异步化如果Agent的某些动作是I/O密集型的如调用外部API、查询数据库不要让主循环同步等待。可以将动作提交到任务队列如Celery、RabbitMQ主循环只负责生成任务和监听结果。这样单个Agent实例可以同时管理多个任务的执行流极大提高吞吐量。路由决策批处理对于请求量大的场景可以将短时间内的一批路由请求收集起来批量发送给模型进行推理如果模型支持批量推理这比逐个请求效率高得多。上下文缓存会话级和任务级的上下文是高频访问对象。使用内存缓存如Redis存储活跃的上下文对象避免每次循环都从数据库读取。注意设计合理的缓存失效和回写策略。动作执行结果缓存对于一些幂等的、结果相对稳定的动作如“根据城市名查询天气”可以对其结果进行缓存。路由决策后先查缓存命中则直接返回避免重复执行。5.2 系统扩展性设计无状态与有状态组件的分离无状态组件路由决策器特别是模型推理部分、某些工具函数。这些组件可以轻松地水平扩展用多个实例加负载均衡即可。有状态组件循环引擎和上下文管理器是典型的有状态组件。一个正在运行的任务其循环状态和上下文必须由同一个进程或线程来维护不能随意迁移。对此常见的模式是采用分片Sharding或基于会话的粘性Sticky Session。例如通过会话ID的哈希值决定由哪个后端实例来处理该会话的所有后续请求。水平扩展循环引擎由于循环引擎有状态扩展起来更复杂。一种架构是采用“主从”模式或“事件溯源”模式。事件溯源模式将Agent的每一次状态变更如“路由决策A”、“执行动作B成功”、“更新上下文C”都记录为一个不可变的事件Event并持久化到事件流如Kafka中。循环引擎本身可以设计得相对轻量它读取事件流根据当前状态和事件计算出新状态并生成新的事件。这样多个循环引擎实例可以消费同一个事件流的不同分区状态的计算逻辑是确定的最终状态由事件序列决定便于扩展和故障恢复。容错与高可用检查点Checkpointing循环引擎定期将任务上下文和循环进度持久化到可靠的存储中。当实例故障时调度器可以将该任务重新分配给另一个健康的实例新实例从最新的检查点加载状态并继续执行。优雅降级当路由模型服务或某个关键动作服务不可用时系统应能降级到备用方案。例如路由降级到基于规则的简单版本动作服务不可用则返回“服务暂不可用请稍后重试”并记录任务状态为“等待重试”。构建一个健壮、高效的Agent运行时系统是一个持续迭代和平衡的过程。从清晰理解循环、路由、上下文这三个核心机制开始在设计和编码时始终想着状态如何流转、决策如何做出、信息如何保存就能避开很多深坑。