利用langchain创建信息检索链对话应用

以下内容有AI生成内容，请注意区分

信息检索链

信息检索链三步流程走向图

flowchart TDA[用户输入问题] --> B[第一步: 查询优化]subgraph B [查询优化]B1[原始用户问题] --> B2[LLM分析并优化]B2 --> B3[生成多个搜索查询]endB --> C[第二步: 信息检索]subgraph C [信息检索]C1[执行多个搜索查询] --> C2[获取相关文档]C2 --> C3[按相关性排序文档]C3 --> C4[选择最相关结果]endC --> D[第三步: 答案生成]subgraph D [答案生成]D1[组合问题与检索结果] --> D2[LLM生成最终答案]D2 --> D3[返回结构化响应]endD --> E[输出最终答案]style A fill:#e1f5festyle E fill:#e8f5e8style B fill:#fff3e0style C fill:#fce4ecstyle D fill:#e8f5e8

流程详解

第一步：查询优化

1. 输入：用户原始问题（自然语言）
2. 处理：LLM分析问题意图，生成更有效的搜索查询
3. 输出：多个优化的搜索查询（通常为2-3个）
4. 目的：将模糊的用户问题转换为搜索引擎友好的查询

第二步：信息检索

1. 输入：优化的搜索查询
2. 处理：
   • 执行多个搜索查询
   • 获取相关文档/片段
   • 按相关性对结果排序
   • 选择最相关的结果
3. 输出：排序后的相关文档片段
4. 目的：从知识库中查找与问题最相关的信息

第三步：答案生成

1. 输入：
   • 原始用户问题
   • 检索到的相关信息
2. 处理：
   • 将问题和检索结果组合成提示词
   • LLM基于提供的上下文生成答案
   • 确保答案基于检索到的信息，不编造内容
3. 输出：结构化、基于证据的最终答案
4. 目的：基于可靠信息生成准确、全面的答案

关键特点

1. 迭代优化：通过多个搜索查询提高找到相关信息的概率
2. 证据基础：所有答案都基于检索到的信息，提高可靠性
3. 可解释性：可以追溯答案的来源（检索到的文档）
4. 灵活性：可以适配不同的搜索引擎和知识库

这个流程是RAG（检索增强生成）架构的核心，广泛应用于问答系统、研究助手和知识库应用。

应用实现

信息检索链是LangChain中非常强大的模式，它模拟了人类获取信息的过程：先搜索相关信息，再基于这些信息生成答案。下面是一个完整的学习示例，使用虚拟的搜索结果来演示整个流程。

完整代码示例

import refrom langchain_openai import ChatOpenAI
from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate
from langchain.schema import Document
from typing import List, Dict# 1. 配置模型
llm = ChatOpenAI(model_name="xop3qwen1b7",openai_api_base="https://域名/v1",openai_api_key="sk-kBa9GlpIxpWX281iA35a2...",temperature=0.3,  # 降低温度以获得更确定的答案max_tokens=1024
)# 2. 改进的查询解析函数
def parse_search_queries(llm_response: str) -> List[str]:"""从LLM响应中解析出搜索查询，处理多种可能的格式"""# 移除可能的前导文本response = llm_response.strip()# 匹配用逗号分隔的查询（可能带有编号）comma_pattern = r'(?:\d+[\.\:]?\s*)?([^,\n]+?)(?=,|$|\n)'comma_matches = re.findall(comma_pattern, response)if comma_matches and len(comma_matches) >= 2:# 清理结果queries = [q.strip() for q in comma_matches if q.strip()]return queries[:3]  # 返回最多3个查询# 匹配用换行符分隔的查询（可能带有编号）line_pattern = r'(?:\d+[\.\:]?\s*)?([^\n]+)'line_matches = re.findall(line_pattern, response)if line_matches and len(line_matches) >= 2:queries = [q.strip() for q in line_matches if q.strip()]return queries[:3]# 如果以上方法都失败，尝试按句号分割fallback_queries = [q.strip() for q in response.split('.') if q.strip()]if fallback_queries and len(fallback_queries) >= 2:return fallback_queries[:3]# 最后的手段：返回原始响应作为单个查询return [response]# 2. 模拟搜索引擎函数（实际项目中可替换为真实搜索引擎API）
def mock_search_api(query: str, num_results: int = 3) -> List[Document]:"""模拟搜索引擎API，返回虚拟的搜索结果"""# 这里是硬编码的模拟数据，实际应用中会调用真实APIsearch_results_db = {"langchain 是什么": [Document(page_content="LangChain是一个用于开发由语言模型驱动的应用程序的框架。它提供了一套工具、组件和接口，简化了构建基于LLM的应用过程。",metadata={"source": "官方文档", "relevance": 0.9}),Document(page_content="LangChain由Harrison Chase于2022年创建，已经成为最流行的LLM应用开发框架之一，拥有活跃的开源社区。",metadata={"source": "技术博客", "relevance": 0.7}),Document(page_content="LangChain的核心概念包括模型I/O、检索、链、代理、内存等模块。",metadata={"source": "教程", "relevance": 0.8})],"langchain 如何安装": [Document(page_content="安装LangChain: pip install langchain。如果需要使用OpenAI模型，还需安装pip install langchain-openai。",metadata={"source": "官方文档", "relevance": 0.95}),Document(page_content="LangChain支持Python 3.8及以上版本。建议使用虚拟环境进行安装。",metadata={"source": "GitHub README", "relevance": 0.6}),Document(page_content="对于开发大型应用，可以考虑安装LangChain的扩展包，如langchain-community等。",metadata={"source": "社区论坛", "relevance": 0.5})],"langchain 的应用场景": [Document(page_content="LangChain可用于构建聊天机器人、问答系统、文档分析工具、代码生成器等应用。",metadata={"source": "用例文档", "relevance": 0.9}),Document(page_content="企业使用LangChain构建客户支持系统、内部知识库检索工具和自动化报告生成器。",metadata={"source": "案例研究", "relevance": 0.8}),Document(page_content="研究人员使用LangChain进行学术文献分析和科学发现辅助工具开发。",metadata={"source": "研究论文", "relevance": 0.7})],"默认": [Document(page_content="很抱歉，没有找到与您查询直接相关的结果。请尝试使用更具体的关键词重新搜索。",metadata={"source": "搜索引擎", "relevance": 0.1})]}# 返回匹配的搜索结果或默认结果return search_results_db.get(query, search_results_db["默认"])[:num_results]# 3. 第一步：生成优化搜索查询的链
query_generation_template = """
你是一个搜索查询优化专家。请将用户的问题转换为更有效、更具体的搜索引擎查询。请使用中文只返回3个搜索查询，用逗号分隔，不要有任何额外的解释、编号或文本。原始问题: {user_question}搜索查询:
"""query_generation_prompt = PromptTemplate(input_variables=["user_question"],template=query_generation_template
)query_chain = LLMChain(llm=llm,prompt=query_generation_prompt,output_key="search_queries"
)# 4. 第二步：执行搜索并处理结果
def search_and_process(search_queries: str) -> str:"""执行搜索并格式化结果"""# 使用改进的解析函数queries = parse_search_queries(search_queries)print(f"解析后的搜索查询: {queries}")all_results = []for query in queries[:2]:  # 取前两个查询results = mock_search_api(query)all_results.extend(results)# 按相关性排序（假设metadata中有relevance字段）all_results.sort(key=lambda x: x.metadata.get("relevance", 0), reverse=True)# 格式化搜索结果formatted_results = ""for i, doc in enumerate(all_results[:4]):  # 取前4个最相关结果formatted_results += f"结果 {i + 1} (来源: {doc.metadata.get('source', '未知')}):\n{doc.page_content}\n\n"return formatted_results# 5. 第三步：基于搜索结果生成最终答案的链
answer_generation_template = """
你是一个专业的研究助手。请根据以下搜索结果为用户的问题提供更全面、准确的答案。如果搜索结果中没有相关信息或搜索结果不相关，请如实告知用户，不能编造信息。用户问题: {user_question}搜索结果:
{search_results}请基于以上信息，用中文提供详细的回答:
"""answer_generation_prompt = PromptTemplate(input_variables=["user_question", "search_results"],template=answer_generation_template
)answer_chain = LLMChain(llm=llm,prompt=answer_generation_prompt,output_key="final_answer"
)# 6. 组合成顺序链
overall_chain = SequentialChain(chains=[query_chain],input_variables=["user_question"],output_variables=["search_queries"],verbose=True
)def run_information_retrieval():print("=" * 60)print("🔍 信息检索链演示系统")print("💡 我会先优化你的问题，然后搜索相关信息，最后生成答案")print("📝 输入'退出'结束程序")print("=" * 60)while True:user_question = input("\n请输入你的问题: ").strip()if user_question.lower() in ['退出', 'quit', 'exit', 'q']:print("感谢使用，再见！")breakif not user_question:continuetry:print("\n🔄 正在优化搜索查询...")# 第一步：生成优化的搜索查询query_result = overall_chain.invoke({"user_question": user_question})search_queries = query_result["search_queries"]print(f"生成的搜索查询: {search_queries}")# 第二步：执行搜索print("🔍 正在搜索相关信息...")search_results = search_and_process(search_queries)print("找到的相关信息:")print(search_results)# 第三步：生成最终答案print("🤖 正在生成最终答案...")final_answer = answer_chain.invoke({"user_question": user_question,"search_results": search_results})print("\n" + "=" * 40)print("✅ 最终答案:")print(final_answer.get("final_answer"))print("=" * 40)except Exception as e:print(f"❌ 出错了: {e}")import tracebacktraceback.print_exc()# 7. 演示函数
def demonstrate_retrieval_chain():"""演示信息检索链的工作流程"""print("\n" + "=" * 60)print("🎯 信息检索链工作流程演示")print("=" * 60)test_questions = ["LangChain是什么?","如何安装LangChain?","LangChain有什么应用场景?"]for question in test_questions:print(f"\n📋 测试问题: {question}")# 生成搜索查询query_result = query_chain.run({"user_question": question})print(f"🔍 生成的搜索查询: {query_result}")# 执行搜索search_results = search_and_process(query_result)# 生成最终答案final_answer = answer_chain.run({"user_question": question,"search_results": search_results})print(f"✅ 最终答案: {final_answer}")print("-" * 40)if __name__ == "__main__":# 运行主程序run_information_retrieval()# 取消注释运行演示# demonstrate_retrieval_chain()

代码解析与学习要点

1. 信息检索链的三步流程

这个示例展示了信息检索链的典型三步流程：

1. 查询优化：将用户的自然语言问题转换为有效的搜索查询
2. 信息检索：使用搜索查询获取相关信息
3. 答案生成：基于检索到的信息生成最终答案

2. 关键组件

• LLMChain：用于处理每个步骤的LLM调用
• SequentialChain：将多个链组合成顺序工作流（这里只用于演示第一步）
• PromptTemplate：定义每个步骤的提示词模板
• 模拟搜索API：在实际项目中可替换为真实搜索引擎（如Google Serper API、SerpAPI等）

3. 提示词工程

注意两个提示词模板的设计：

1. 查询生成提示：要求LLM生成多个不同的搜索查询
2. 答案生成提示：要求LLM基于搜索结果回答问题，并明确指示不要编造信息

4. 错误处理与可靠性

• 提供了默认搜索结果，避免空结果导致的错误
• 添加了异常处理机制
• 对搜索结果进行排序，优先使用最相关的内容

进阶学习建议

1. 集成真实搜索引擎

将mock_search_api函数替换为真实的搜索引擎API：

# 使用Serper API示例（需要注册获取API key）
from langchain_community.utilities import GoogleSerperAPIWrapperdef real_search_api(query: str, num_results: int = 3):search = GoogleSerperAPIWrapper()results = search.results(query)# 处理结果格式...return processed_results

2. 添加结果评估步骤

在搜索和答案生成之间添加结果评估步骤，过滤低质量或无关的结果：

evaluation_template = """
请评估以下搜索结果与问题的相关性（0-1分），并简要说明理由。问题: {user_question}搜索结果:
{search_results}请给出分数和理由:
"""evaluation_prompt = PromptTemplate(...)
evaluation_chain = LLMChain(...)

3. 实现多查询融合

对多个搜索查询的结果进行去重、排序和融合，提高答案质量：

def merge_search_results(results_list):"""合并和去重多个查询的结果"""all_docs = []seen_content = set()for results in results_list:for doc in results:content_hash = hash(doc.page_content[:100])  # 简单去重if content_hash not in seen_content:seen_content.add(content_hash)all_docs.append(doc)# 按相关性排序all_docs.sort(key=lambda x: x.metadata.get("relevance", 0), reverse=True)return all_docs[:5]  # 返回前5个最相关结果

4. 添加引用来源

在最终答案中注明信息来源，提高可信度：

citation_template = """
...（其他内容保持不变）请基于以上信息，用中文提供详细的回答，并在末尾注明信息来源。答案:
"""# 在答案生成后处理来源标注
def add_citations(answer, sources):citation_text = "\n\n信息来源: "citation_text += ", ".join(set(sources))return answer + citation_text

运行与测试

运行程序后，尝试以下问题：

1. "LangChain是什么？"
2. "如何安装LangChain？"
3. "LangChain有什么应用场景？"