30秒速览
- 多Agent协同别指望它们自动共享记忆，上下文说丢就丢，我40%的错误率就是这么来的。
- 暴力把历史全塞进prompt能缓解，但token费用爆炸，不是长久之计。
- 最终靠结构化输出（Pydantic Schema）+ 系统提示模板注入上下文，把失忆率压到了3%。
- 给Agent设计好通信契约并校验，比指望它们自觉可靠一万倍。
- 监控必须做，日志必须细，不然出问题你连北都找不到。
那个凌晨3点的Bug：Agent们突然集体失忆
上周三，我为一个中型跨境电商平台“海淘优品”做的智能客服系统上线了压力测试。这个系统用了Claude的Code Workflow功能，搞了个三Agent协同架构：一个理解员负责解析用户模糊的售后问题，一个查单员去数据库里捞订单和物流，最后一个策略员根据前两个的结果生成具体的解决方案话术。理论上很美，分工明确，各司其职。白天小流量测试时，一切正常，Agent们配合得跟老友记似的。
结果晚上10点，模拟真实用户高峰（日活大概8万）的压测一开始，系统就开始抽风。监控大屏上，错误率像坐火箭一样往上窜。我一看日志，全是“策略员”Agent在胡言乱语，给出的方案跟用户问题风马牛不相及。比如用户问“我上周买的咖啡机刀头坏了怎么办？”，理解员正确解析出了“产品问题：咖啡机刀头损坏”，查单员也找到了对应的订单。但策略员给出的回复竟然是“您的物流包裹预计明天送达，请保持手机畅通”。这他娘的是完全失忆了，根本不记得前面两个兄弟干了啥。
我一开始以为是Claude的API限流或者网络抖动，加了重试和降级。没用。又怀疑是Agent的system prompt没写清楚，连夜改了十几版，把职责描述得跟法律条文一样精确。还是没用。错误率稳定在40%左右，这意味着将近一半的复杂售后请求会得到完全错误的答案。客户那边的技术负责人已经在钉钉群里@我问“王工，这系统是不是还没睡醒？”。我盯着满屏的ERROR日志，咖啡灌到第三杯，知道今晚是别想睡了。问题的核心很明确：在高压、多轮的工作流中，Agent之间的“对话上下文”像断线的风筝一样，说丢就丢。
别天真地以为Agent会自动共享记忆
这是我踩的第一个，也是最深的坑。我最初的设计简单得有点天真，直接用Claude提供的Python SDK串起了三个Agent，代码大概长这样：
# 初始的错误设计：以为把上一个Agent的回复传给下一个就行
from anthropic import Anthropic
import os
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
def naive_agent_workflow(user_query):
# Agent 1: 理解员
understanding_response = 应将 model 参数设为单一的正确模型 ID，例如 "claude-3-5-sonnet-20241022" 或 "claude-3-opus-20240229"。
understanding_result = understanding_response.content[0].text
print(f"理解员输出: {understanding_result}")
# Agent 2: 查单员 - 直接把理解员的文本输出作为输入
query_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=1000,
system="你是一个数据查询专家，根据理解专家提供的JSON，生成数据库查询语句并虚拟执行，返回订单和物流信息。",
messages=[{"role": "user", "content": understanding_result}] # 问题在这里！
)
query_result = query_response.content[0].text
print(f"查单员输出: {query_result}")
# Agent 3: 策略员 - 同样，只看到查单员的输出
strategy_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=800,
system="你是一个客服策略专家，根据查询到的订单数据和用户问题，生成具体的解决方案和话术。",
messages=[{"role": "user", "content": query_result}] # 上下文严重缺失！
)
return strategy_response.content[0].text看出来问题了吗？每个Agent都是一个独立的client.messages.create调用。对于Claude API来说，这就是三次完全独立的对话会话（session）。查单员根本不知道原始的user_query是什么，它只看到了理解员输出的那个JSON字符串。策略员更惨，它只看到了查单员输出的查询结果，既不知道原始问题，也不知道用户的意图。这就好比让三个人传话，第一个人听到“我想喝咖啡”，传给第二个人时可能变成“他要个杯子”，第三个人听到后直接给了个碗。信息在传递过程中被不断剥离上下文，最后的结果必然失真。
在低流量下，这个问题不明显，因为模型或许能从残存的信息中“猜”出一点上下文。并发请求不会导致模型内部注意力机制不稳定，性能下降通常因为系统资源瓶颈或上下文管理问题。。我犯了一个基础错误：把LLM的对话上下文（Conversation Context）和工作流的状态上下文（Workflow State Context）混为一谈了。LLM的对话上下文只在单次messages数组的传递中有效，而我需要的是一个跨多个独立API调用的、持久化的状态管理。
我像法医一样解剖了这次“对话”
找到方向后，我开始像破案一样收集证据。首先，我在每个Agent的调用前后加上了详细的日志，不仅打印输入输出，还把完整的messages结构、token消耗都记下来。然后，我写了一个小脚本，从错误日志中随机采样了50个失败案例，把这三个Agent的输入输出像拼图一样拼回完整的链条。
分析结果让我后背发凉。下面是一个典型的“失忆链条”：
| 阶段 | 输入内容（简化） | 输出内容（简化） | 丢失的关键信息 |
|---|---|---|---|
| 原始用户输入 | “我上周买的SK-II神仙水，瓶子漏了，能换货吗？” | – | – |
| 理解员 | 用户输入原文 | {“产品”: “SK-II神仙水”, “问题”: “包装泄漏”, “诉求”: “换货”, “时间”: “一周内”} | 无 |
| 查单员 | 理解员的JSON字符串 | “订单号: ORD20241028XXX， 状态: 已签收， 金额: 1580元” | 丢失了“包装泄漏”和“换货”诉求 |
| 策略员 | 查单员的查询结果 | “您好，您订单已签收，如有质量问题请在7天内申请售后。” | 丢失了原始问题、具体产品、具体问题，只剩下一个通用的售后话术 |
看到没？信息就像沙漏里的沙子，每经过一个Agent就漏掉一大半。查单员只从JSON里提取了“产品”和“时间”去查订单，完全忽略了“问题”和“诉求”。策略员拿到一个干巴巴的订单信息，它就算再聪明，也不可能无中生有出“瓶子漏了”这个事实。
更坑爹的是，我发现即使我把理解员的完整输出传给查单员，在高压下，Claude模型有时也会“选择性阅读”，只关注它认为重要的字段（比如产品名、订单号），而忽略其他（比如问题描述）。这跟模型的注意力机制和 prompt 的设计都有关。我需要一个强制性的、结构化的方式来传递完整上下文，而不是依赖模型的“自觉”。
我的武器库：从土法炼钢到系统化方案
搞清楚病因，我开始尝试治疗。前前后后折腾了四种方案，整个过程就像在给一个失忆症患者做康复训练。
方案一：暴力堆料——把所有历史信息都塞进prompt
这是最直接的想法：既然你记不住，我就把前面所有人的话都念给你听。我修改了查单员和策略员的messages参数，把之前所有步骤的输入输出都拼接起来作为当前步骤的用户输入。
def brute_force_context_agent_workflow(user_query):
# 初始化一个上下文字符串
full_context = f"原始用户问题: {user_query}nn"
# Agent 1: 理解员
understanding_response = client.messages.create(...) # 同前
understanding_result = understanding_response.content[0].text
full_context += f"===理解员分析结果===n{understanding_result}nn"
# Agent 2: 查单员 - 看到所有之前的信息
query_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=1000,
system="你是一个数据查询专家，请仔细阅读以下全部上下文，然后执行你的任务。",
messages=[{"role": "user", "content": full_context + "请根据上述信息生成查询语句并返回虚拟数据。"}]
)
query_result = query_response.content[0].text
full_context += f"===查单员查询结果===n{query_result}nn"
# Agent 3: 策略员 - 看到从头到尾的所有信息
strategy_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=800,
system="你是一个客服策略专家，请仔细阅读以下全部上下文，然后生成最终答复。",
messages=[{"role": "user", "content": full_context + "请根据以上所有信息生成给用户的最终答复。"}]
)
return strategy_response.content[0].text效果与代价：错误率从40%降到了15%左右，有效果！但代价巨大：每个后续Agent的prompt变得无比冗长，token消耗飙升了差不多3倍。单个请求的成本和时间都上去了。而且，我观察到在超长上下文下，模型对最开头和最后面的信息更敏感，中间部分还是有被忽略的风险。这方案救急可以，但绝不是长久之计。
方案二：外部记忆库——用Redis当Agent的共享大脑
我琢磨着，既然LLM自己的对话上下文靠不住，那我就给它造一个外部的。思路是给每个用户会话分配一个唯一ID（session_id），每个Agent把自己的产出以结构化的方式存到一个共享存储（比如Redis）里，下一个Agent在执行前，先去Redis里把当前session_id下的所有历史记录读出来，作为自己上下文的一部分。 效果与代价：这个方案更系统化，错误率进一步降到了8%左右。而且有了session_id，可以很方便地做对话的持久化和回溯分析。代价是架构变复杂了，引入了Redis依赖，需要处理网络延迟和缓存失效的问题。另外，如何设计 方案二已经不错，但我总觉得还能再优化。在又一轮的调试中，我发现了问题的另一个维度：信息衰减不仅发生在传递过程，也发生在信息格式的转换中。理解员输出的是JSON，查单员输出的是自然语言描述，策略员又得去理解这些自然语言。每次格式转换都是一次信息损失的冒险。 我的终极方案结合了三个关键点： 下面是核心代码实现： 这个方案为什么最终胜出？ 当然，没有完美的方案。这个设计的代价是前期需要精心设计每个Agent的接口Schema，并且要求LLM模型有较好的遵循指令和输出结构化内容的能力（好在Claude-3.5-Sonnet这方面很强）。另外，system prompt的模板会变得比较复杂，需要细心维护。 经过这次从凌晨3点的崩溃到系统稳定的折腾，我对Claude Code（或者说任何多LLM Agent协同）工作流有了几点刻骨铭心的认识： 现在，“海淘优品”的智能客服系统已经稳定运行了两周，日均处理2万多次复杂咨询，人工客服的转接率降低了65%。虽然那天加班到凌晨3点的记忆依然“深刻”，但看着监控面板上那条平稳的绿色成功率曲线，我觉得这坑踩得值。多Agent协同是条充满诱惑又遍布陷阱的路，手里有地图（清晰的架构设计）和手电筒（细致的调试监控），你才能走得稳当点。 那个“记忆碎片化”的结论，只是我深入Agent协同记忆迷宫的入口。在将失忆率从40%干到20%之后，我意识到，剩下的问题不再是某个单一模块的故障，而是整个信息流转体系的系统性脆弱。这就像修好了水管的主阀门，却发现支线管道依然到处是沙眼。我必须对Agent之间的“对话”方式进行一场外科手术式的改造。 我首先用高精度日志，完整记录了三个Agent在一次完整会话中，输入和输出的所有上下文内容。结果绘制出了一条清晰的“信息衰减曲线”。理解员从用户模糊的“我买的那个蓝色的、上周到的杯子坏了”中，准确提取了{商品类型:杯子，特征:蓝色，到货时间:上周内，问题:损坏}。但到了查单员手里，为了适配数据库查询接口，它被简化成了一个结构化的查询条件： 问题出在策略员。它收到的输入是查单员的查询结果（订单号、物流单号、购买日期）和理解员最初输出（包含“问题:损坏”）的混合体。但在复杂的判断逻辑中，原始问题有时会被覆盖。我亲眼在日志里看到，一次会话中，策略员中间步骤的思考是：“用户反馈商品损坏，需核实是否在保修期并询问是否有照片”，但最终输出的话术却变成了：“已为您查询到订单物流正常，请问还有什么可以帮您？”——它“忘记”了损坏这件事。 我的解决方案不再是简单的“传递更多”，而是设计一个结构化的“信息接力棒”（InfoBaton）。我定义了一个强制性的共享内存结构，以JSON格式在Agent间传递： 这个 最初的Redis缓存是粗粒度的：以 我借鉴了CPU缓存的设计思想，引入了“冷热分层缓存”策略。将一次会话中的信息分为三层： 更重要的是，我建立了缓存间的逻辑关联。当查单员查询到订单#67890时，它不仅缓存订单数据，还会在缓存中建立一个反向索引： 新的缓存和上下文系统运行平稳后，我再次启动了压力测试。然而，在模拟“黑五”量级的并发请求冲击下，失忆率又出现小幅回升，稳定在8%左右。监控显示，在峰值时，部分请求的 我意识到，给Agent赋予强大的记忆能力，也必须给它们装上“过载保护开关”。我实现了以下机制： 同时，我为每个Agent增加了“上下文健康度”自检。在收到 经过这一系列从微观数据结构到宏观系统架构的改造，Agent失忆率终于被我死死地按在了3%以下。监控面板上那条曾经剧烈跳动的红色曲线，如今变成了一条紧贴X轴的、平静的绿线。那个凌晨三点让我崩溃的“集体失忆”Bug，彻底成为了过去。但我知道，在AI协同的深水区，新的挑战，永远在下一个天亮等待。import json
import redis
import uuid
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
class AgentMemory:
def __init__(self, session_id=None):
self.session_id = session_id or str(uuid.uuid4())
self.memory_key = f"agent_workflow:{self.session_id}"
def save_step(self, agent_name, input_data, output_data):
"""保存一个Agent步骤的完整信息"""
step = {
"agent": agent_name,
"input": input_data,
"output": output_data,
"timestamp": time.time()
}
# 使用列表存储步骤，保持顺序
redis_client.rpush(self.memory_key, json.dumps(step, ensure_ascii=False))
# 设置过期时间，例如1小时，防止内存泄漏
redis_client.expire(self.memory_key, 3600)
def get_full_history(self):
"""获取当前会话的完整历史"""
history_json = redis_client.lrange(self.memory_key, 0, -1)
return [json.loads(h) for h in history_json]
def get_condensed_history(self, max_steps=5):
"""获取精简版历史，只保留最近几步和关键信息，用于控制token"""
full = self.get_full_history()
if len(full) <= max_steps:
return full
# 保留最近几步，并尝试总结更早的步骤（这里简化处理，只取最近）
return full[-max_steps:]
def workflow_with_memory(user_query, session_id):
memory = AgentMemory(session_id)
# 1. 理解员
understanding_response = client.messages.create(...)
understanding_result = understanding_response.content[0].text
memory.save_step("理解员", user_query, understanding_result)
# 2. 查单员 - 从memory中构建上下文
history_for_query = memory.get_condensed_history()
# 构建一个给查单员的提示，包含精简历史和明确指令
query_prompt = build_prompt_from_history(history_for_query, current_task="查询订单")
query_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=1000,
system="你是一个数据查询专家。以下是与当前用户会话相关的历史记录，请参考它们来理解当前任务。",
messages=[{"role": "user", "content": query_prompt}]
)
query_result = query_response.content[0].text
memory.save_step("查单员", query_prompt, query_result)
# 3. 策略员 - 同样从memory构建更丰富的上下文
history_for_strategy = memory.get_condensed_history()
strategy_prompt = build_prompt_from_history(history_for_strategy, current_task="生成最终答复")
strategy_response = client.messages.create(...)
final_result = strategy_response.content[0].text
memory.save_step("策略员", strategy_prompt, final_result)
return final_result, memory.session_id
def build_prompt_from_history(history_steps, current_task):
"""一个将历史步骤构建成提示文本的辅助函数"""
prompt_lines = [f"当前任务：{current_task}"]
prompt_lines.append("以下是本次会话中已完成的步骤摘要：")
for step in history_steps:
prompt_lines.append(f"- [{step['agent']}] 输入：{step.get('input', '')[:200]}...")
prompt_lines.append(f" 输出：{step.get('output', '')[:300]}...")
应改为 prompt_lines.append("n请基于以上信息，完成你的当前任务")，或者直接使用正确拼接。。")
return "n".join(prompt_lines)build_prompt_from_history这个函数是个艺术活，摘要提取得不好，还是会丢失关键信息。最终让我活过来的方案：结构化输出+校验工作流
from pydantic import BaseModel, Field
from typing import Optional, List
import json
# 第一步：用Pydantic定义所有Agent的输入输出Schema
class UnderstandingOutput(BaseModel):
product_name: str = Field(description="产品名称")
problem_type: str = Field(description="问题类型，如：质量、物流、使用等")
user_demand: str = Field(description="用户诉求，如：退货、换货、维修、补偿")
order_time_range: Optional[str] = Field(None, description="订单时间范围")
extracted_keywords: List[str] = Field(default_factory=list, description="提取的关键词")
class QueryOutput(BaseModel):
order_id: str = Field(description="订单号")
order_status: str = Field(description="订单状态")
product_info_match: bool = Field(description="查询到的产品是否与问题产品一致")
logistics_info: Optional[str] = Field(None, description="物流信息")
# 关键：这里直接引用上游的字段，建立连接
relates_to_problem: str = Field(description="此订单信息如何关联到用户报告的问题")
class StrategyOutput(BaseModel):
solution: str = Field(description="解决方案，如：同意换货、补偿优惠券、提供维修指南")
reply_template: str = Field(description="给用户的回复话术模板")
internal_note: str = Field(description="给客服人员的内部备注，说明处理依据")
confidence: float = Field(ge=0, le=1, description="策略置信度")
def robust_agent_workflow(user_query: str):
"""最终版的工作流，结构化上下文传递"""
workflow_state = {
"original_query": user_query,
"understanding": None,
"query": None,
"strategy": None
}
# --- Agent 1: 理解员 ---
# System Prompt里强调了结构化输出
understand_system = """
你是一个客服问题理解专家。你的任务是将用户模糊的描述转化为结构化的信息。
你必须严格按照提供的JSON格式输出，仅输出JSON对象，不要有任何额外解释。
"""
understand_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=500,
system=understand_system,
messages=[
{"role": "user", "content": f"请分析以下用户问题，并输出结构化信息：n{user_query}"}
]
)
# 解析并验证输出
try:
understanding_dict = json.loads(understand_response.content[0].text)
workflow_state["understanding"] = UnderstandingOutput(**understanding_dict)
except (json.JSONDecodeError, ValidationError) as e:
# 如果Agent不听话，没输出JSON，这里可以重试或降级处理
print(f"理解员输出解析失败: {e}")
# 降级：用一个简单规则提取关键信息
workflow_state["understanding"] = create_fallback_understanding(user_query)
# --- Agent 2: 查单员 ---
# 关键技巧：将上游的结构化对象，以清晰的方式注入system prompt
query_system_template = """
你是一个数据查询专家。基于以下已经确认的用户问题分析结果，生成查询并返回虚拟数据。
【已确认的问题分析】
产品名称：{product_name}
问题类型：{problem_type}
用户诉求：{user_demand}
关键词：{keywords}
原始问题：{original_query}
你的输出必须是严格的JSON格式，包含order_id, order_status等字段。
"""
# 格式化system prompt，直接嵌入上游数据
query_system = query_system_template.format(
product_name=workflow_state["understanding"].product_name,
problem_type=workflow_state["understanding"].problem_type,
user_demand=workflow_state["understanding"].user_demand,
keywords=", ".join(workflow_state["understanding"].extracted_keywords),
original_query=workflow_state["original_query"] # 再次注入原始问题！
)
query_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=1000,
system=query_system,
messages=[{
"role": "user",
"content": "请根据系统提示中已确认的信息，生成订单查询结果（虚拟数据）。"
}]
)
try:
query_dict = json.loads(query_response.content[0].text)
workflow_state["query"] = QueryOutput(**query_dict)
except (json.JSONDecodeError, ValidationError) as e:
print(f"查单员输出解析失败: {e}")
workflow_state["query"] = create_fallback_query(workflow_state["understanding"])
# --- Agent 3: 策略员 ---
# 同样，将所有上游结构化数据注入system prompt
strategy_system_template = """
你是一个客服策略专家。请基于以下完整的、结构化的会话信息生成最终策略。
【完整会话上下文】
1. 原始用户问题：{original_query}
2. 问题分析结果：
- 产品：{product_name}
- 问题：{problem_type}
- 诉求：{user_demand}
3. 订单查询结果：
- 订单号：{order_id}
- 状态：{order_status}
- 关联性说明：{relates_to_problem}
请综合以上所有信息，生成解决方案。输出必须是严格的JSON格式。
"""
strategy_system = strategy_system_template.format(
original_query=workflow_state["original_query"],
product_name=workflow_state["understanding"].product_name,
problem_type=workflow_state["understanding"].problem_type,
user_demand=workflow_state["understanding"].user_demand,
order_id=workflow_state["query"].order_id,
order_status=workflow_state["query"].order_status,
relates_to_problem=workflow_state["query"].relates_to_problem
)
strategy_response = client.messages.create(
model="claude-3-opus-20240229 或 claude-3-sonnet-20240229 等具体模型 ID-5-sonnet-20241022",
max_tokens=800,
system=strategy_system,
messages=[{
"role": "user",
"content": "请生成最终客服策略。"
}]
)
try:
strategy_dict = json.loads(strategy_response.content[0].text)
workflow_state["strategy"] = StrategyOutput(**strategy_dict)
return workflow_state["strategy"].reply_template
except (json.JSONDecodeError, ValidationError) as e:
print(f"策略员输出解析失败: {e}")
return "抱歉，系统暂时无法处理您的问题，已转接人工客服。"
# 辅助的降级函数
def create_fallback_understanding(query: str) -> UnderstandingOutput:
# 简单的规则或关键词匹配来创建降级输出
return UnderstandingOutput(
product_name="未知产品",
problem_type="其他",
user_demand="咨询",
extracted_keywords=[]
)
给后来者的血泪经验：多Agent系统不是过家家
try...except和create_fallback_xxx函数，就是我的安全网。workflow_state字典足够了。只有当你需要跨请求、长时记忆的会话时，才去考虑外部存储。不必要的复杂性是万恶之源。四、深入Agent记忆迷宫：从缓存到上下文的系统性改造
4.1 上下文传递的“衰减曲线”与我的“信息接力棒”设计
WHERE product_type='杯子' AND color='蓝色' AND delivery_date BETWEEN '2023-10-23' AND '2023-10-30'。这里，“坏了”这个核心问题被暂时搁置了。class InfoBaton:
def __init__(self, session_id):
self.session_id = session_id
self.core_user_problem = None # 用户核心问题，永不删除
self.extracted_entities = {} # 理解员提取的实体
self.retrieved_data = {} # 查单员查询的结果
self.processing_chain = [] # 记录处理步骤和关键决策
self.flagged_attention = [] # 需要后续Agent特别注意的事项
def set_core_problem(self, problem):
"""由理解员调用，设定本次会话不可遗忘的核心"""
self.core_user_problem = problem
self.flagged_attention.append(f"核心待解决问题: {problem}")
def add_processing_step(self, agent_name, decision, reason):
"""每个Agent在处理后必须添加步骤记录"""
self.processing_chain.append({
'agent': agent_name,
'step': len(self.processing_chain) + 1,
'decision': decision,
'reason': reason,
'timestamp': time.time()
})
InfoBaton对象在每个工作流的开头被创建，并作为必须参数传递给每一个Agent。策略员在生成最终话术前，会被强制要求检查core_user_problem和flagged_attention，并在processing_chain中记录其回应方案是否覆盖了所有要点。这一设计，将隐式的、容易丢失的上下文，变成了显式的、结构化的、可追踪的数据对象。4.2 缓存策略的“冷热分层”：不止于时间，更在于逻辑关联
session_id为键，存储整个对话历史。这带来了两个问题：一是单个键值过大，序列化/反序列化消耗剧增；二是缓存失效策略单一，整个会话要么全在，要么全无。
InfoBaton的最新版本。使用独立的短时键（如session_id:hot），TTL设置为5分钟，读写优先级最高。processing_chain）和已查询的静态数据（如产品信息、用户基本信息）。按逻辑模块拆分存储（如session_id:chain, session_id:product:12345），TTL延长至30分钟。order:67890 -> [session_id_1, session_id_2]。这样，如果后续其他Agent或会话需要关联此订单，可以快速定位到相关的会话上下文碎片，实现了跨会话的“记忆联想”。这个改动看似微小，却让Agent在面对用户“我刚才问的那个订单……”这类指代性提问时，命中率提升了惊人的70%。4.3 压力下的“记忆过载保护”与优雅降级
InfoBaton构建时间过长，甚至出现序列化错误。def get_context_with_fallback(session_id, required_fields):
"""
智能获取上下文，带有降级逻辑
"""
try:
# 尝试从热缓存获取完整InfoBaton
baton = redis.get(f"{session_id}:hot")
if baton and all(field in baton for field in required_fields):
return baton, "full"
# 降级1：尝试从温层缓存拼凑关键字段
fallback_data = {}
for field in required_fields:
value = redis.get(f"{session_id}:{field}")
if value:
fallback_data[field] = value
if len(fallback_data) >= len(required_fields) * 0.6: # 关键字段命中率超过60%
logger.warning(f"Session {session_id} using fallback context.")
return fallback_data, "partial"
# 降级2：最小化上下文，仅包含核心问题（从更持久的存储获取）
core_problem = get_core_problem_from_db(session_id) # 极简查询
return {"core_user_problem": core_problem}, "minimal"
except Exception as e:
# 降级3：紧急情况，启动无上下文模式
logger.error(f"Context fetch failed for {session_id}: {e}")
return {"error": "system_busy"}, "degraded"
InfoBaton后，Agent会快速检查其完整性和逻辑连贯性。如果发现异常（例如，processing_chain中出现断裂），它会主动在日志中标记并尝试从其他层恢复，而不是将错就错地执行下去。这相当于给每个Agent配备了一个记忆质检员。
