Agent上下文工程入门指南：从零构建高效对话系统

1次阅读

共计 2716 个字符，预计需要花费 7 分钟才能阅读完成。

传统对话系统在处理多轮对话时，常常会遇到这样的尴尬场景：

用户问 ” 北京的天气怎么样？”，系统回答 ” 北京今天晴转多云 ”
用户接着问 ” 那上海呢？”，系统却回答 ” 您想问什么？”

这种上下文丢失的问题，正是因为传统系统缺乏有效的对话状态管理机制。

让我们通过一个表格直观比较不同架构的表现：

指标	Rule-based	Intent-based	Agent 架构
上下文保持能力	❌ 差	⭕ 一般	✅ 优秀
内存占用 (MB/ 会话)	5-10	20-50	10-30
平均响应延迟 (ms)	50-100	200-500	100-300
开发维护成本	高	中	低

下面我们用 Python 构建一个最简单的上下文管理器：

from dataclasses import dataclass
from typing import List, Dict
import zlib  # 用于对话压缩

@dataclass
class DialogueTurn:
    speaker: str
    utterance: str
    timestamp: float

class ContextManager:
    def __init__(self, max_turns=10):
        self.history: List[DialogueTurn] = []
        self.max_turns = max_turns
        self.compression_threshold = 0.7  # 压缩阈值

    def add_turn(self, speaker: str, utterance: str) -> None:
        """添加对话轮次，自动执行压缩"""
        new_turn = DialogueTurn(
            speaker=speaker,
            utterance=utterance,
            timestamp=time.time())

        # 触发压缩的条件
        if len(self.history) >= self.max_turns:
            self._compress_history()

        self.history.append(new_turn)

    def _compress_history(self) -> None:
        """对话历史压缩算法"""
        # 时间复杂度 O(n)，n 为历史记录数
        compressed = []
        key_phrases = set()

        for turn in self.history:
            # 提取关键短语（简化版）phrases = [p for p in turn.utterance.split() if len(p) > 3]
            key_phrases.update(phrases)

        summary = ",".join(sorted(key_phrases))
        compressed.append(DialogueTurn(
            speaker="SYSTEM",
            utterance=f"[压缩上下文] 关键短语: {summary}",
            timestamp=time.time()))

        # 保留最近 30% 的对话
        keep_count = int(len(self.history) * self.compression_threshold)
        self.history = compressed + self.history[-keep_count:]

用 UML 描述的状态转移图：

[用户发起请求] --> [初始状态]
[初始状态] --> | 问候语 | [等待用户输入]
[等待用户输入] --> | 查询请求 | [信息收集]
[信息收集] --> | 参数完整 | [执行查询]
[执行查询] --> | 成功 | [结果展示]
[结果展示] --> [等待用户输入]

内存存储
优点：零延迟
缺点：重启丢失，不适合分布式
适用：开发测试环境
Redis
优点：高性能，支持 TTL
缺点：需要额外基础设施

示例配置：

import redis
r = redis.Redis(
    host='context-db',
    port=6379,
    db=0,
    socket_timeout=5
)

数据库
优点：持久可靠
缺点：性能较低
推荐：PostgreSQL 的 JSONB 类型

import re

def sanitize_input(text: str) -> str:
    """过滤身份证号、银行卡号等敏感信息"""
    patterns = [(r'\b\d{17}[\dXx]\b', '[ID_NUMBER]'),  # 身份证
        (r'\b\d{4}[-]?\d{4}[-]?\d{4}[-]?\d{4}\b', '[CARD_NUMBER]'),
        (r'\b\d{11}\b', '[PHONE]')  # 手机号
    ]

    for pattern, replacement in patterns:
        text = re.sub(pattern, replacement, text)
    return text

上下文膨胀
现象：内存占用飙升，响应变慢
解决：实现上面的压缩算法，设置 max_turns 限制
对话超时未处理
现象：用户隔天回来继续对话，上下文混乱

解决：添加会话 TTL（示例）：

class ContextManager:
    def __init__(self, ttl_minutes=30):
        self.ttl = ttl_minutes * 60

    def is_expired(self):
        return time.time() - self.history[-1].timestamp > self.ttl

跨会话状态污染
现象：用户 A 的偏好影响用户 B
解决：严格隔离会话 ID，使用 token 验证

基础框架已经准备好：

from flask import Flask, request
app = Flask(__name__)

# TODO: 在这里实现你的上下文管理器
contexts = {}  # 会话 ID 到上下文的映射

@app.route('/chat', methods=['POST'])
def chat():
    """带上下文缓存的对话端点"""
    data = request.json
    session_id = data.get('session_id')
    utterance = sanitize_input(data.get('text', ''))

    # 你的实现代码...
    # 需要处理:
    # 1. 新会话初始化
    # 2. 上下文检索与更新
    # 3. 响应生成

    return {
        'response': "这是一个占位响应",
        'session_id': session_id
    }

完成基础实现后，可以尝试以下进阶方向：