Marco Liu1. 项目概述：当LangChain遇上RAG与Agent
三年前我第一次接触LangChain时，它还是个刚诞生不久的框架。如今这个工具链已经成长为连接大模型与现实业务的关键桥梁。最近在金融知识问答系统项目中，我们通过LangChain实现的RAG+Agent架构，将GPT-4的准确率从63%提升到了89%。这让我意识到，掌握这套技术栈正在成为AI工程师的必备技能。
本文要探讨的"消息简写"技术，源于我们在处理证券行业客户需求时的实际痛点。当Agent需要同时处理多个工具调用和知识库查询时，上下文窗口经常被冗长的中间过程占满。通过开发消息压缩模块，我们成功将对话token消耗降低了40%，这直接关系到API调用成本和响应速度。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 技术架构解析
2.1 RAG与Agent的协同机制
典型的RAG流程包含知识库构建（embedding+向量存储）和检索增强生成两个阶段。而在Agent体系中，RAG作为工具（Tool）存在，其特殊性在于：
- 动态检索：不同于传统问答系统的一次性检索，Agent可能在多轮对话中触发多次RAG调用
- 上下文感知：每次检索都携带当前对话历史，需要智能过滤无关信息
- 结果后处理：检索到的文档需要经过重排序、截断等操作才能输入LLM
我们实现的混合架构如下图所示（伪代码表示）：
python复制class HybridAgent:
def __init__(self):
self.retriever = FAISSVectorStore(retriever) # RAG模块
self.tools = [StockQueryTool(), Calculator()] # 其他工具
def run(self, query):
# 决策路由
if needs_knowledge(query):
docs = self.retriever.search(query)
return self._compress_context(docs)
else:
return self._call_tools(query)
2.2 消息简写的技术实现
消息压缩的核心在于保持语义完整性的同时减少token占用。我们对比了三种方案：
| 方法 | 压缩率 | 语义保留度 | 适用场景 |
|---|---|---|---|
| 关键词提取 | 60-70% | ★★☆ | 简单问答 |
| 摘要生成 | 50-60% | ★★★ | 复杂文档 |
| 结构化表示（推荐） | 40-50% | ★★★★ | 多工具调用场景 |
最终采用的结构化表示方案示例：
python复制def compress_message(history):
compressed = []
for msg in history:
if msg['role'] == 'tool':
# 将工具返回结果转换为结构化摘要
compressed.append({
'tool': msg['metadata']['name'],
'status': 'success' if msg['success'] else 'failed',
'key_data': extract_key_fields(msg['content'])
})
else:
compressed.append(msg) # 保留原始对话
return compressed
3. 关键实现细节
3.1 动态上下文管理
在实时对话中，我们开发了滑动窗口策略来平衡历史保留与token消耗：
- 重要性评分：基于TF-IDF和注意力权重计算每段话的重要性
- 分层存储：
- 核心上下文（完整保留最近3轮）
- 摘要上下文（保留前10轮的压缩版本）
- 长期记忆（存储到向量数据库供检索）
实现代码片段：
python复制class ContextManager:
def update_context(self, new_message):
# 计算消息重要性
importance = self._calculate_importance(new_message)
if importance > self.HIGH_THRESHOLD:
self.core_context.append(new_message)
elif importance > self.LOW_THRESHOLD:
self.summary_context.append(
self.compressor.summarize(new_message))
# 执行窗口滑动
if len(self.core_context) > 3:
oldest = self.core_context.pop(0)
self.summary_context.append(
self.compressor.summarize(oldest))
3.2 工具调用优化
当Agent需要链式调用多个工具时，传统方式会产生大量中间输出。我们的解决方案：
- 并行执行：对无依赖关系的工具调用使用asyncio
- 结果预过滤：在工具返回时立即应用压缩规则
- 元编程技巧：通过装饰器自动生成工具描述
python复制@tool(compress_level=2)
def stock_query(symbol: str):
"""查询股票实时数据
Args:
symbol: 股票代码
Returns:
dict格式的市场数据
"""
data = yfinance.Ticker(symbol).info
return {
'price': data['currentPrice'],
'change': data['regularMarketChange']
} # 仅返回关键字段
4. 性能优化实战
4.1 Token消耗对比测试
在证券客服场景下的基准测试结果：
| 场景 | 原始token | 压缩后token | 节省比例 |
|---|---|---|---|
| 单轮简单问答 | 1,024 | 812 | 20.7% |
| 多工具调用 | 3,572 | 2,101 | 41.2% |
| 长对话（10+轮） | 7,845 | 4,326 | 44.9% |
4.2 质量评估指标
引入压缩后需要监控的关键指标：
- 意图识别准确率（Intent Accuracy）
- 信息完整度（Information Preservation Score）
- 响应延迟（P99 Latency）
我们的AB测试显示，适度压缩（30-40%）反而能提升指标：
原因在于：
- 减少了噪声干扰
- 突出了关键信息
- 降低了模型处理长文本的负担
5. 典型问题排查
5.1 信息丢失问题
症状：压缩后回答出现关键事实错误
解决方案：
- 建立关键实体白名单
- 添加校验层：
python复制def validate_compression(original, compressed):
entity_diff = set(extract_entities(original)) - set(extract_entities(compressed))
if len(entity_diff) > 0:
raise CriticalEntityMissingError(list(entity_diff))
5.2 工具兼容性问题
不同工具的输出格式差异会导致压缩失败。我们制定的规范：
- 强制要求工具返回JSON格式
- 提供标准的元数据字段：
json复制{
"_meta": {
"compress_priority": 1,
"key_fields": ["price", "volume"]
},
"data": {...}
}
6. 进阶技巧
6.1 自适应压缩策略
根据对话阶段动态调整压缩强度：
python复制def get_compression_level(turn_count):
if turn_count < 3:
return 1 # 轻度压缩
elif turn_count < 10:
return 2 # 中度压缩
else:
return 3 # 深度压缩
6.2 记忆增强方案
对于需要长期记忆的场景，我们设计了两级缓存：
- 短期缓存：Redis存储最近1小时对话的压缩版本
- 长期记忆：通过向量存储关键事实，使用以下映射规则：
python复制class MemoryMapper:
def to_memory(self, message):
return {
"embedding": get_embedding(message),
"facts": extract_facts(message),
"timestamp": time.time()
}
在实际项目中，这种优化使得系统能够处理超过50轮的复杂对话，同时保持token消耗在8k以内。一个意外的收获是，压缩过程本身也成为了信息重要性评估的过滤器，反而提升了回答质量。