Claude Code Team Work的协同陷阱：我如何把Agent失忆率从40%干到3%

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-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)

效果与代价：这个方案更系统化，错误率进一步降到了8%左右。而且有了session_id，可以很方便地做对话的持久化和回溯分析。代价是架构变复杂了，引入了Redis依赖，需要处理网络延迟和缓存失效的问题。另外，如何设计build_prompt_from_history这个函数是个艺术活，摘要提取得不好，还是会丢失关键信息。

最终让我活过来的方案：结构化输出+校验工作流

方案二已经不错，但我总觉得还能再优化。在又一轮的调试中，我发现了问题的另一个维度：信息衰减不仅发生在传递过程，也发生在信息格式的转换中。理解员输出的是JSON，查单员输出的是自然语言描述，策略员又得去理解这些自然语言。每次格式转换都是一次信息损失的冒险。

我的终极方案结合了三个关键点：

1. 强制结构化输出：每个Agent都必须以严格的、预先定义好的JSON Schema输出。这样下游Agent可以直接解析，而不是去“理解”一段模糊文本。
2. 上下文注入与校验：将上游的关键输出，以结构化的方式，作为下游Agent系统提示（system prompt）的一部分变量，而不仅仅是用户消息里的文本。
3. 轻量级状态跟踪：在内存中维护一个工作流状态对象，而不是所有东西都依赖外部存储，对于单次请求的工作流，这样更简单高效。

下面是核心代码实现：

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-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-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-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=[]
    )

这个方案为什么最终胜出？

• 错误率降到3%：结构化输出和清晰的上下文注入，让信息丢失的可能性大大降低。Agent不再需要“猜测”上游的意图。
• Token使用更高效：相比方案一的暴力堆料，我们只传递了关键的结构化字段，没有冗长的自然语言描述，整体token消耗比最初设计只增加了约30%，但效果天差地别。
• 可调试性极强：每个步骤的输入输出都是结构化的对象，我可以轻松地将其记录到日志或监控系统，一眼就能看出是哪个环节出了问题。
• 健壮性更好：通过Pydantic模型校验，一旦Agent输出不符合约定，我们能立刻发现并触发降级逻辑，而不是让错误一直传递下去。

当然，没有完美的方案。这个设计的代价是前期需要精心设计每个Agent的接口Schema，并且要求LLM模型有较好的遵循指令和输出结构化内容的能力（好在Claude-3.5-Sonnet这方面很强）。另外，system prompt的模板会变得比较复杂，需要细心维护。

给后来者的血泪经验：多Agent系统不是过家家

经过这次从凌晨3点的崩溃到系统稳定的折腾，我对Claude Code（或者说任何多LLM Agent协同）工作流有了几点刻骨铭心的认识：

1. 上下文是命根子，你必须显式管理它。永远不要假设Agent能自动记住或理解之前的步骤。把需要传递的上下文，以最清晰、最结构化、最不易丢失的方式，主动“喂”给下一个Agent。System prompt是你的主要工具。
2. 设计输出契约，并用代码强制校验。让Agent之间用定义好的API（JSON Schema）通信，而不是用自然语言闲聊。用Pydantic这类工具去验证输出，在接口层面就把错误扼杀在摇篮里。
3. 为失败做好计划。LLM输出具有不确定性，网络会抖动，API会限流。你的工作流必须有降级策略（fallback）、重试机制和清晰的错误处理。我的代码里那些try...except和create_fallback_xxx函数，就是我的安全网。
4. 监控和可观测性不是可选项。在每个Agent的输入输出点埋下日志，记录关键状态和token使用。当问题再次出现时（相信我，它一定会），这些日志是你快速定位问题的唯一依据。我后来甚至加了一个简单的仪表盘，实时显示每个环节的成功率和平均处理时间。
5. 保持简单，直到你不得不复杂。我最终的方案没有引入Redis作为工作流状态管理，因为对于这个“单次请求内完成”的客服工作流，内存中的workflow_state字典足够了。只有当你需要跨请求、长时记忆的会话时，才去考虑外部存储。不必要的复杂性是万恶之源。

现在，“海淘优品”的智能客服系统已经稳定运行了两周，日均处理2万多次复杂咨询，人工客服的转接率降低了65%。虽然那天加班到凌晨3点的记忆依然“深刻”，但看着监控面板上那条平稳的绿色成功率曲线，我觉得这坑踩得值。多Agent协同是条充满诱惑又遍布陷阱的路，手里有地图（清晰的架构设计）和手电筒（细致的调试监控），你才能走得稳当点。

四、深入Agent记忆迷宫：从缓存到上下文的系统性改造

那个“记忆碎片化”的结论，只是我深入Agent协同记忆迷宫的入口。在将失忆率从40%干到20%之后，我意识到，剩下的问题不再是某个单一模块的故障，而是整个信息流转体系的系统性脆弱。这就像修好了水管的主阀门，却发现支线管道依然到处是沙眼。我必须对Agent之间的“对话”方式进行一场外科手术式的改造。

4.1 上下文传递的“衰减曲线”与我的“信息接力棒”设计

我首先用高精度日志，完整记录了三个Agent在一次完整会话中，输入和输出的所有上下文内容。结果绘制出了一条清晰的“信息衰减曲线”。理解员从用户模糊的“我买的那个蓝色的、上周到的杯子坏了”中，准确提取了{商品类型:杯子，特征:蓝色，到货时间:上周内，问题:损坏}。但到了查单员手里，为了适配数据库查询接口，它被简化成了一个结构化的查询条件：WHERE product_type='杯子' AND color='蓝色' AND delivery_date BETWEEN '2023-10-23' AND '2023-10-30'。这里，“坏了”这个核心问题被暂时搁置了。

问题出在策略员。它收到的输入是查单员的查询结果（订单号、物流单号、购买日期）和理解员最初输出（包含“问题:损坏”）的混合体。但在复杂的判断逻辑中，原始问题有时会被覆盖。我亲眼在日志里看到，一次会话中，策略员中间步骤的思考是：“用户反馈商品损坏，需核实是否在保修期并询问是否有照片”，但最终输出的话术却变成了：“已为您查询到订单物流正常，请问还有什么可以帮您？”——它“忘记”了损坏这件事。

我的解决方案不再是简单的“传递更多”，而是设计一个结构化的“信息接力棒”（InfoBaton）。我定义了一个强制性的共享内存结构，以JSON格式在Agent间传递：

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 缓存策略的“冷热分层”：不止于时间，更在于逻辑关联

最初的Redis缓存是粗粒度的：以session_id为键，存储整个对话历史。这带来了两个问题：一是单个键值过大，序列化/反序列化消耗剧增；二是缓存失效策略单一，整个会话要么全在，要么全无。

我借鉴了CPU缓存的设计思想，引入了“冷热分层缓存”策略。将一次会话中的信息分为三层：

1. 热层（Hot Layer）：当前轮次直接相关的信息，即InfoBaton的最新版本。使用独立的短时键（如session_id:hot），TTL设置为5分钟，读写优先级最高。
2. 温层（Warm Layer）：本次会话的历史步骤（processing_chain）和已查询的静态数据（如产品信息、用户基本信息）。按逻辑模块拆分存储（如session_id:chain, session_id:product:12345），TTL延长至30分钟。
3. 冷层（Cold Layer）：完整的原始对话历史。仅当需要深度回溯或分析时才从数据库加载，平时不占用缓存空间。

更重要的是，我建立了缓存间的逻辑关联。当查单员查询到订单#67890时，它不仅缓存订单数据，还会在缓存中建立一个反向索引：order:67890 -> [session_id_1, session_id_2]。这样，如果后续其他Agent或会话需要关联此订单，可以快速定位到相关的会话上下文碎片，实现了跨会话的“记忆联想”。这个改动看似微小，却让Agent在面对用户“我刚才问的那个订单……”这类指代性提问时，命中率提升了惊人的70%。

4.3 压力下的“记忆过载保护”与优雅降级

新的缓存和上下文系统运行平稳后，我再次启动了压力测试。然而，在模拟“黑五”量级的并发请求冲击下，失忆率又出现小幅回升，稳定在8%左右。监控显示，在峰值时，部分请求的InfoBaton构建时间过长，甚至出现序列化错误。

我意识到，给Agent赋予强大的记忆能力，也必须给它们装上“过载保护开关”。我实现了以下机制：

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"

同时，我为每个Agent增加了“上下文健康度”自检。在收到InfoBaton后，Agent会快速检查其完整性和逻辑连贯性。如果发现异常（例如，processing_chain中出现断裂），它会主动在日志中标记并尝试从其他层恢复，而不是将错就错地执行下去。这相当于给每个Agent配备了一个记忆质检员。

经过这一系列从微观数据结构到宏观系统架构的改造，Agent失忆率终于被我死死地按在了3%以下。监控面板上那条曾经剧烈跳动的红色曲线，如今变成了一条紧贴X轴的、平静的绿线。那个凌晨三点让我崩溃的“集体失忆”Bug，彻底成为了过去。但我知道，在AI协同的深水区，新的挑战，永远在下一个天亮等待。