AI Agent与RAG技术中的意图路由与查询重写实践
1. 项目概述：AI Agent与RAG技术落地的核心挑战
在当前的AI应用开发浪潮中，AI Agent和RAG（Retrieval-Augmented Generation）技术正成为企业级解决方案的热门选择。但很多开发团队在实际落地时都会遇到两个关键瓶颈：如何准确理解用户意图，以及如何优化查询语句以获得更精准的检索结果。这正是"意图路由+查询重写"技术组合要解决的核心问题。
我最近主导了一个金融领域的智能客服系统升级项目，就深刻体会到这两个技术点的重要性。当用户询问"我的信用卡账单怎么还"时，系统需要先判断这是查询账单（路由到查询模块）还是设置还款（路由到交易模块），然后还要将口语化表达重写为"信用卡账单还款流程"这样的结构化查询语句。没有这两个环节，再强大的LLM也容易给出偏离实际的回答。
2. 技术架构解析
2.1 意图路由的工作原理
意图路由的本质是一个分类器，其技术实现通常包含以下关键组件：
特征提取层：
- 词袋模型（TF-IDF）
- 词向量（Word2Vec/FastText）
- 上下文嵌入（BERT等预训练模型）
分类算法选型对比：
| 算法类型 | 准确率 | 训练速度 | 可解释性 | 适用场景 |
|---|---|---|---|---|
| 朴素贝叶斯 | 中等 | 快 | 高 | 小规模数据 |
| SVM | 高 | 慢 | 中 | 标注数据充足 |
| 神经网络 | 很高 | 很慢 | 低 | 大数据量 |
- 实际项目中的混合方案：
# 实际项目中我们采用的混合方案 def intent_classifier(query): # 第一层：快速规则匹配 if contains_keywords(query, ["账单","还款"]): return "finance_payment" # 第二层：模型预测 bert_embedding = bert_model.encode(query) return sklearn_svm.predict(bert_embedding)2.2 查询重写的技术实现
查询重写不只是简单的同义词替换，而是要考虑：
语义扩展：
- 同义词："手机"→"智能手机"
- 上下位词："苹果"→"水果"或"手机品牌"
- 关联概念："充电"→"电池续航"
结构优化：
- 去除停用词
- 标准化表达（"咋办"→"如何处理"）
- 添加领域限定词（"转账"→"银行转账"）
实战代码示例：
from transformers import pipeline rewriter = pipeline("text2text-generation", model="tscholak/cxmefzzi") def rewrite_query(query, domain="finance"): prompt = f"将以下{domain}领域查询改写为专业检索语句：{query}" return rewriter(prompt, max_length=50)[0]['generated_text']关键提示：重写模型需要针对特定领域微调，通用模型在专业场景效果会大打折扣
3. 完整实现流程
3.1 环境准备与工具链
推荐的技术栈组合：
基础框架：
- LangChain（Agent开发）
- LlamaIndex（RAG优化）
向量数据库选型：
- Milvus（大规模生产环境）
- FAISS（快速原型开发）
- Chroma（轻量级应用）
部署架构：
用户请求 → 意图分类服务 → 查询重写服务 → RAG引擎 → LLM生成 → 返回结果3.2 分步实现指南
- 构建意图分类器：
# 使用sentence-transformers构建 from sentence_transformers import SentenceTransformer, util model = SentenceTransformer('paraphrase-MiniLM-L6-v2') # 准备训练数据 intent_examples = { "account_query": ["我的余额是多少","查一下账单"], "payment": ["我要还款","怎么转账"] } # 生成意图嵌入 intent_embeddings = { intent: model.encode(examples) for intent, examples in intent_examples.items() } def classify_intent(query): query_embed = model.encode(query) scores = { intent: util.cos_sim(query_embed, embeds).mean() for intent, embeds in intent_embeddings.items() } return max(scores.items(), key=lambda x: x[1])[0]- 实现查询重写模块：
# 基于规则+模型混合方案 import re rewrite_rules = { r"\b怎么\b": "如何", r"\b办\b": "办理" } def rewrite_query(query): # 规则改写 for pattern, repl in rewrite_rules.items(): query = re.sub(pattern, repl, query) # 模型改写 if should_use_model(query): # 基于复杂度判断 return llm_rewrite(query) return query- RAG系统集成：
from langchain.chains import RetrievalQA from langchain.llms import OpenAI def build_rag_system(): # 1. 文档加载与分块 loader = DirectoryLoader('./docs') documents = loader.load() text_splitter = RecursiveCharacterTextSplitter() chunks = text_splitter.split_documents(documents) # 2. 向量化存储 embeddings = OpenAIEmbeddings() vectorstore = FAISS.from_documents(chunks, embeddings) # 3. 构建检索链 retriever = vectorstore.as_retriever() qa = RetrievalQA.from_chain_type( llm=OpenAI(), chain_type="stuff", retriever=retriever ) return qa4. 性能优化与生产级考量
4.1 延迟优化方案
缓存策略：
- 意图分类结果缓存（TTL 5分钟）
- 查询改写结果缓存（相似度>0.9时复用）
异步处理流程：
async def handle_request(query): # 并行执行耗时操作 intent_future = classify_intent_async(query) rewrite_future = rewrite_query_async(query) intent, rewritten = await asyncio.gather(intent_future, rewrite_future) return await generate_response(intent, rewritten)4.2 准确率提升技巧
bad case分析流程：
- 收集错误样本
- 人工标注修正
- 增量训练模型
AB测试方案：
# 在流量分流时进行版本对比 def route_request(query): if hash(query) % 10 < 3: # 30%流量走新算法 return new_pipeline(query) return old_pipeline(query)5. 典型问题排查指南
5.1 意图识别不准
现象：将"转账到支付宝"识别为账户查询
排查步骤：
- 检查训练数据是否包含类似样本
- 验证embedding模型是否适配领域
- 测试分类阈值是否合理（可调整score阈值）
5.2 检索结果偏离
现象：查询"信用卡年费"返回借记卡信息
解决方案：
- 检查重写后的查询语句
- 验证向量空间中的距离分布
- 添加负样本增强训练
经验之谈：当发现bad case时，最好的解决方式是在训练数据中添加3-5个相似样本，而不是立即调整模型参数
6. 进阶发展方向
- 动态路由机制：
def dynamic_router(query, history): context = analyze_conversation(history) if context.get('pending_task'): return "task_continuation" return intent_classifier(query)多模态扩展：
- 图像中的文字识别后路由
- 语音指令的意图识别
在线学习系统：
- 人工反馈闭环
- 自动数据增强
在实际项目迭代中，我们发现将用户修正行为（如重新提问）作为训练数据反馈给系统，能在2-3个迭代周期内显著提升准确率。例如某电商客服系统通过记录用户的"这不是我想要的"操作，使意图识别准确率从78%提升到了92%。