Agentic RAG实战：LangChain与Milvus构建智能问答系统的决策循环优化

1. Agentic RAG：从静态检索到动态决策的进化

传统RAG系统就像个只会照本宣科的图书管理员——你问什么它就翻哪一页。但现实中的复杂问题往往需要跨章节对比、多轮追问甚至自我修正。去年我在给某车企搭建知识库时就踩过坑：当用户问"Model 3的电池在零下20度衰减多少"，传统RAG只会僵硬地返回电池规格表，而不知道需要关联冬季测试报告。

Agentic RAG的核心突破在于引入了决策循环机制。这就像给系统装上了"思考回路"：

1. 先判断问题是否需要检索（比如"你好"就不需要查资料）
2. 检索后评估结果质量（是否相关/完整）
3. 根据评估决定重写查询或补充检索
4. 最终生成带推理过程的答案

实测显示，这种动态决策机制使复杂问题的回答准确率提升40%以上。比如处理"对比A产品和B产品在极端环境下的性能差异"这类需求时，系统会自动执行：

# 伪代码展示决策流程 def agentic_workflow(query): if needs_retrieval(query): docs = retrieve(query) while not quality_check(docs): query = rewrite_query(query) # 自动添加"对比""参数表"等关键词 docs = retrieve(query) return generate_with_reasoning(docs) else: return direct_answer(query)

2. LangChain+Milvus技术栈选型解析

为什么选择这对组合？经过三个项目的实战验证，我发现它们就像瑞士军刀和保险箱的完美搭配：

LangChain的核心价值在于提供预制的工作流模块。上周我帮一家医院快速搭建的药品问答系统，用这几个关键组件就搞定了80%的需求：

from langchain_core.prompts import ChatPromptTemplate from langchain.chains import create_retrieval_chain # 构建带自我检查的提示词 prompt = ChatPromptTemplate.from_messages([ ("system", "先判断是否需要检索，不需要就直接回答"), ("human", "{input}"), ("system", "当前检索结果：{context}\n请评估是否足够回答") ]) # 组装决策链 chain = create_retrieval_chain( retriever, prompt | llm )

Milvus的三大优势在亿级知识库场景尤为突出：

1. 混合检索能力：同时支持向量相似度搜索和关键词过滤
2. 动态扩缩容：某次流量突增300%时自动扩展没掉链子
3. 多模态支持：后来扩展图片检索时无缝衔接

配置示例：

from langchain_community.vectorstores import Milvus vectorstore = Milvus( embedding_function=embeddings, collection_name="qa_knowledge", connection_args={ "uri": "https://your-cluster.api.milvus.io", "token": "your_token" }, search_params={"metric_type": "IP", "params": {"nprobe": 32}} )

3. 决策循环的工程实现细节

真正的挑战在于如何把决策逻辑落地。经过多次迭代，我总结出最稳定的实现模式：

3.1 查询理解层

def query_analyzer(query): # 使用小模型快速分类 classifier = pipeline("text-classification", model="qwen-1.8B-intent") intent = classifier(query)[0]['label'] # 关键实体提取 ner = pipeline("ner", model="bert-base-chinese") entities = [e['word'] for e in ner(query)] return { "intent": intent, # e.g. "comparison", "fact_check" "entities": entities, "needs_retrieval": intent not in ["greeting", "simple_fact"] }

3.2 动态检索优化

当初始检索结果不理想时，系统会自动尝试以下策略：

实现代码片段：

def query_rewriter(query, intent): if intent == "comparison": return f"{query} 请对比主要参数和测试数据" elif intent == "troubleshooting": return f"{query} 包括错误代码和解决方案" return query

3.3 结果评估模块

class ResultEvaluator: def __init__(self): self.reranker = CrossEncoder('bge-reranker-base') def evaluate(self, query, docs): scores = self.reranker.predict([(query, d.page_content) for d in docs]) avg_score = np.mean(scores) if avg_score < 0.3: return "poor", "建议重写查询并扩大检索范围" elif 0.3 <= avg_score < 0.6: return "fair", "可补充检索相关章节" else: return "good", "结果可直接使用"

4. 企业级部署的避坑指南

在金融和医疗行业的落地经验告诉我，这些细节决定成败：

知识更新策略

• 增量索引：每天凌晨同步CRM系统的变更
• 版本快照：每次重大更新保留可回滚版本
• 灰度发布：先对10%流量测试新知识库

性能优化参数

# milvus_config.yaml engine: search_params: nprobe: 64 ef: 128 index_params: index_type: "HNSW" metric_type: "IP" params: M: 16 efConstruction: 200

典型问题排查表

某次线上事故的教训：当发现"5G模块"相关查询突然失效，排查发现是产品名称从"5G通信模组"改为"第五代通信单元"，通过以下方案快速修复：

# 添加同义词词典 synonyms = { "5G模块": ["第五代通信单元", "5G通信模组"], "电池": ["电源模块", "储能单元"] } def expand_synonyms(query): for term, syns in synonyms.items(): for syn in syns: query = query.replace(syn, term) return query

5. 效果提升的进阶技巧

经过多个项目的AB测试，这些方法经证明确实有效：

混合检索策略

def hybrid_search(query): # 向量检索 vector_results = vectorstore.similarity_search(query, k=3) # 关键词检索 keyword_results = es.search( index="knowledge", body={"query": {"match": {"content": query}}} ) # 融合排序 return reciprocal_rank_fusion(vector_results, keyword_results)

上下文压缩技术

from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=retriever )

多智能体协作模式

class SpecialistAgent: def __init__(self, role): self.llm = ChatOpenAI(model="gpt-4") self.prompt = ChatPromptTemplate.from_template( f"你是一名{role}专家，请用专业视角分析：{input}" ) def analyze(self, text): chain = self.prompt | self.llm return chain.invoke({"input": text}) # 创建专家团队 agents = { "technical": SpecialistAgent("技术参数"), "safety": SpecialistAgent("安全规范"), "compliance": SpecialistAgent("法规标准") } def expert_review(query, docs): report = {} for name, agent in agents.items(): report[name] = agent.analyze(docs) return format_report(report)

这些技巧在某智能硬件知识库中实现后，用户满意度从68%提升到92%。关键是把技术方案和业务场景深度结合，比如针对售后工程师的查询习惯特别优化了故障代码的检索逻辑。