压力测试：让智能体在 1000 轮对话后仍保持精准

现在是周五下午 4 点 52 分，产品经理发来一条消息：“客户要求我们的客服智能体至少要能连续处理一个月的对话量。你们那个多智能体记忆方案，连续跑 1000 轮还能保持精准吗？周一给我报告。”

你不会等到周一才发现问题。接下来的 48 小时内，你将亲手把这个系统推向极限——制造记忆膨胀、注入噪声、触发冲突广播——看它究竟在哪个点开始退化，以及如何把它重新稳住。

你需要什么

注意：本章不关注特定模型厂商，代码以抽象接口编写。你用 Claude、GPT 或本地模型都可以接入。

最终成果

你将获得一套可复用的压力测试套件，包含：

1. 模拟长对话脚本：自动生成 1000 轮混合任务对话，插入噪声和无关历史。
2. 评估指标体系：关键信息保留率、幻觉率、上下文一致性——三项指标自动计算。
3. 瓶颈定位与调优记录：根据测试结果调整记忆压缩、检索和衰减参数，形成量化改进轨迹。

做这件事的核心逻辑是：一个没被压力测试打脸过的记忆系统，不值得被信任上生产。

步骤一：生成模拟长对话脚本

1.1 设计对话场景矩阵

真实生产环境不会是清一色的问答。你需要混合以下任务类型，才能模拟实际退化路径：

1.2 实现对话生成器

import json
import random
import asyncio
from datetime import datetime
from typing import List, Dict

# 定义任务模板库
TASK_TEMPLATES = {
    "query": [
        {"template": "订单 ORD-{order_id} 的当前状态是什么？", "key": "order_status"},
        {"template": "我在 {date} 购买的商品发货了吗？", "key": "shipment"},
        {"template": "我的账户余额还有多少？", "key": "balance"}
    ],
    "update": [
        {"template": "请将我的收货地址更新为 {new_address}", "key": "address_update"},
        {"template": "我需要把订单 ORD-{order_id} 的配送方式改成加急", "key": "shipping_change"},
        {"template": "我的手机号换成 {new_phone}", "key": "phone_update"}
    ],
    "noise": [
        {"template": "今天天气真不错，适合出门散步。", "key": None},
        {"template": "你觉得人工智能会取代人类工作吗？哈哈开玩笑的。", "key": None},
        {"template": "我午饭吃了太多，现在有点困。", "key": None}
    ],
    "cross_reference": [
        {"template": "我之前提到的那个客户，他的订单也发货了吗？", "key": "cross_ref_customer"},
        {"template": "上次你建议的那个方案，再帮我仔细说一下？", "key": "cross_ref_suggestion"}
    ]
}

# 生成 1000 轮混合对话
def generate_stress_conversation(
    num_turns: int = 1000,
    seed: int = 42
) -> List[Dict]:
    """
    生成包含噪声和逻辑断层的模拟长对话。
    关键设计：在特定轮次插入“探针问题”（probe questions），
    用于后续评估关键信息保留率。
    """
    random.seed(seed)
    conversation = []

    # 预先埋入的“关键信息”，将在后续被召回测试
    ground_truth = {
        "initial_address": "北京市海淀区中关村大街1号",
        "initial_order_id": "ORD-2024-0001",
        "updated_address": "上海市浦东新区张江高科技园区",
        "sensitive_fact": "客户明确要求周三前必须到货，否则退货"
    }

    for turn in range(num_turns):
        # 根据权重抽任务类型
        task_type = random.choices(
            ["query", "update", "noise", "cross_reference"],
            weights=[40, 25, 20, 15]
        )[0]

        template_pool = TASK_TEMPLATES[task_type]
        selected = random.choice(template_pool)

        # 填充模板变量
        user_message = selected["template"].format(
            order_id=f"ORD-2024-{random.randint(1000, 9999)}",
            date=datetime.now().strftime("%Y-%m-%d"),
            new_address="上海市浦东新区张江高科技园区",
            new_phone="13800000001"
        )

        turn_record = {
            "turn_id": turn,
            "task_type": task_type,
            "key": selected["key"],
            "user_message": user_message,
            "timestamp": datetime.now().isoformat()
        }
        conversation.append(turn_record)

        # 每 200 轮插入一个探针问题（记忆关键信息在早期被设置）
        if turn % 200 == 0 and turn > 0:
            probe = {
                "turn_id": turn,
                "task_type": "probe",
                "key": "memory_retention_check",
                "user_message": f"请回顾：我在最初设置里提到的收货地址是什么？",
                "expected_answer_contains": "北京市海淀区中关村大街1号",
                "timestamp": datetime.now().isoformat()
            }
            conversation.append(probe)

    return conversation

# 运行生成
stress_dialogue = generate_stress_conversation(num_turns=1000)
print(f"生成了 {len(stress_dialogue)} 轮对话记录")
# 预期输出：生成了 1005 轮对话记录（含 5 个探针）

踩坑经验
不要用纯随机字符串模拟对话。真实的记忆退化往往是“近因效应”（模型更关注最近的消息）和“关键信息被稀释”（大量噪声淹没早期事实）叠加造成的。你必须在脚本中预先埋入确定的 ground truth，到后面才可能用自动化手段衡量信息保留了多少。

步骤二：定义自动评估指标

2.1 指标设计原则

在 1000 轮对话后，你不可能人工逐一验收每轮的回忆是否准确。需要一套自动打分系统。

2.2 实现自动评分

from difflib import SequenceMatcher
from typing import List, Tuple

class StressTestEvaluator:
    """
    压力测试评估器：量化记忆系统在极端条件下的退化程度。
    """

    def __init__(self, ground_truth_facts: Dict[str, str]):
        self.ground_truth = ground_truth_facts
        self.probe_results = []  # 记录每次探针的回答

    def evaluate_single_response(
        self, 
        response: str, 
        expected_contains: str
    ) -> Tuple[bool, float, int]:
        """
        对单次回答打分。
        返回：(是否命中关键信息, 相似度得分, 新增实体数量)
        """
        # 关键信息命中检查
        hit = expected_contains.lower() in response.lower()

        # 相似度（衡量上下文一致性）
        similarity = SequenceMatcher(
            None, 
            response.lower(), 
            expected_contains.lower()
        ).ratio()

        # 简化的幻觉检测：统计专有名词
        # 实际生产环境应结合 NER 模型
        import re
        entities_found = re.findall(
            r'[\u4e00-\u9fa5]{2,}(?:大街|路|园区|科技园|区)', 
            response
        )
        novel_entities = len(entities_found)  # 简化版，实际需要减掉 ground truth 中的实体

        return hit, similarity, novel_entities

    def run_on_probes(
        self, 
        agent_responses: List[Dict]
    ) -> Dict[str, float]:
        """
        在所有探针回答上运行评估，汇总指标。
        """
        hits = 0
        total_similarity = 0.0
        total_novel_entities = 0
        total_probes = len(agent_responses)

        for resp in agent_responses:
            expected = resp.get("expected_answer_contains", "")
            response_text = resp.get("agent_response", "")

            hit, sim, novel = self.evaluate_single_response(
                response_text, expected
            )
            hits += 1 if hit else 0
            total_similarity += sim
            total_novel_entities += novel

        return {
            "关键信息保留率": hits / total_probes if total_probes > 0 else 0,
            "平均上下文一致性": total_similarity / total_probes,
            "平均幻觉实体数": total_novel_entities / total_probes
        }

# 示例：模拟对探针问题的回答（实际应由你的智能体生成）
mock_probe_responses = [
    {
        "turn_id": 201,
        "agent_response": "根据初始设置，您的收货地址是北京市海淀区中关村大街1号。",
        "expected_answer_contains": "北京市海淀区中关村大街1号"
    },
    {
        "turn_id": 401,
        "agent_response": "您的收货地址好像是上海市那边……浦东新区张江高科技园区。",
        "expected_answer_contains": "北京市海淀区中关村大街1号"  # 这里智能体记错了
    },
    {
        "turn_id": 601,
        "agent_response": "我记得您最早的地址是北京市海淀区中关村大街1号。",
        "expected_answer_contains": "北京市海淀区中关村大街1号"
    }
]

evaluator = StressTestEvaluator(ground_truth_facts={})
scores = evaluator.run_on_probes(mock_probe_responses)
print(scores)
# 预期输出：
# {'关键信息保留率': 0.67, '平均上下文一致性': 0.72, '平均幻觉实体数': 1.0}

注意
上述幻觉检测是极度简化的版本。在生产级系统中，你应当集成 NER（命名实体识别）来提取新出现的实体，并与已记录的事实库做差集，从而算出真正的幻觉率。本章聚焦在可用性评估，完整 NER 集成可参考调研素材中 LangChain 官方文档的工具调用日志追踪方案。

步骤三：定位瓶颈并调优

3.1 连接智能体运行完整测试

现在将生成的长对话喂给你的多智能体记忆中间层：

# 伪代码：实际运行你的智能体
from your_agent_module import AgentWithMemoryMiddleware

agent = AgentWithMemoryMiddleware(
    shared_board_config={"max_tokens": 4000},
    private_memory_config={"ttl_hours": 72},
    conflict_resolution="latest_wins"  # 默认策略
)

# 逐轮喂入对话
actual_responses = []
for turn in stress_dialogue:
    response = agent.process_message(
        user_id="customer_001",  # 模拟同一用户
        message=turn["user_message"],
        turn_id=turn["turn_id"]
    )
    actual_responses.append({
        "turn_id": turn["turn_id"],
        "agent_response": response,
        "expected_answer_contains": turn.get("expected_answer_contains", "")
    })

# 只取探针轮的响应做评估
probe_responses = [r for r in actual_responses if r["expected_answer_contains"]]
scores = evaluator.run_on_probes(probe_responses)
print(f"=== 初始配置评分 ===")
print(scores)

3.2 典型瓶颈与修复策略

根据你的测试结果，通常会遇见以下退化模式：

3.3 调优代码示例

# 调优配置示例
optimized_config = {
    "shared_board": {
        "max_tokens": 8000,  # 从 4000 提升到 8000
        "protected_labels": ["customer_preferences", "critical_commitments"],  # 受保护事实不参与压缩
        "compression_ratio": 0.6  # 未保护区域在达到阈值时压缩到原来的 60%
    },
    "private_memory": {
        "ttl_hours": 168,  # 重要信息保持一周
        "conflict_policy": "timestamp_priority",  # 更新时如果新值的时间戳更新，才覆盖
        "decay_factor": 0.95  # 每次检索时，非关键记忆的权重乘以此系数
    }
}

agent_v2 = AgentWithMemoryMiddleware(**optimized_config)

# 重新运行压力测试
actual_responses_v2 = []
for turn in stress_dialogue:
    response = agent_v2.process_message(
        user_id="customer_001",
        message=turn["user_message"],
        turn_id=turn["turn_id"]
    )
    actual_responses_v2.append({
        "turn_id": turn["turn_id"],
        "agent_response": response,
        "expected_answer_contains": turn.get("expected_answer_contains", "")
    })

probe_responses_v2 = [r for r in actual_responses_v2 if r["expected_answer_contains"]]
scores_v2 = evaluator.run_on_probes(probe_responses_v2)

print("=== 优化后评分 ===")
print(scores_v2)
print(f"关键信息保留率提升：{scores_v2['关键信息保留率'] - scores['关键信息保留率']:.2%}")
# 预期：关键信息保留率从 60% 左右提升到 85%+

踩坑经验
调优配置时，不要同时改多个参数。每次只改一处，重新跑一遍完整压力测试，对比分数变化。否则你无法判断到底是哪个改动起了作用，最终会陷入“参数体操”的泥潭。

回顾

你刚刚走完了从“担心系统不可靠”到“有信心交报告”的完整闭环：

1. 生成模拟长对话：用加权随机和探针机制构建了 1000 轮混合任务流。
2. 定义自动评估：关键信息保留率、幻觉率、上下文一致性——三项可量化指标。
3. 定位并调优：根据测试结果调整了共享板容量、衰减策略和冲突解决逻辑。

从头到尾大约需要 2 小时。此后，每次你的智能体记忆架构有变动，你都可以重新跑这个套件，看一眼三个数字就知道是变好了还是变坏了。

行动清单

• [ ] 按你的业务场景改写 TASK_TEMPLATES，让对话模板反映真实用户问题。
• [ ] 在 generate_stress_conversation 中埋入自己场景的关键事实（如用户偏好、历史承诺）。
• [ ] 将 StressTestEvaluator 的幻觉检测接上 NER 模型，替换当前的简化正则。
• [ ] 运行一次基线测试，记录初始的三项指标数值。
• [ ] 一次只改一个记忆参数，重跑测试，形成量化的调优日志。

当你的智能体在 1000 轮压力测试中站稳脚跟，下一步的挑战就是如何精准找出那些仍然存在、但不是压力测试直接暴露的隐藏 bug。在下一章《记忆问题的排查是智能体调试中最困难的环节》中，你会学到一个系统化的排错方法——从“它为什么记错了”这个模糊的症状出发，一步步追溯到具体的代码行或参数值。我们下一章见。