LangChain实战：RAG与Agent技术高效开发指南

1. 项目概述

在当今AI技术快速发展的背景下，RAG(检索增强生成)和Agent技术已成为构建智能系统的两大核心支柱。本周我们将深入探讨如何通过LangChain框架高效调用聊天模型，并分享在实际开发中提升效率的简写技巧。

LangChain作为当前最流行的AI应用开发框架之一，其灵活性和扩展性使其成为连接大语言模型与实际业务场景的桥梁。特别是在RAG系统和Agent开发中，LangChain提供了标准化的接口和丰富的工具链，大大降低了开发门槛。

提示：本文假设读者已具备Python基础知识和LangChain基本概念，我们将聚焦于实战中的高级应用技巧和性能优化方法。

2. 核心组件解析

2.1 RAG系统架构

RAG系统的核心在于将检索(Retrieval)与生成(Generation)两个过程有机结合。典型架构包含以下组件：

1. 文档加载器：支持PDF、HTML、Markdown等多种格式
2. 文本分割器：按语义或固定长度切分文档
3. 向量数据库：存储文档的向量表示，常用Milvus、Weaviate等
4. 检索器：基于相似度搜索返回相关文档片段
5. 语言模型：基于检索结果生成最终响应

# 典型RAG流程示例 from langchain.document_loaders import WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma # 加载文档 loader = WebBaseLoader("https://example.com") docs = loader.load() # 分割文本 text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) splits = text_splitter.split_documents(docs) # 创建向量存储 vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())

2.2 Agent工作机制

Agent是能够自主决策和执行任务的智能体，其核心组件包括：

1. 工具(Tools)：Agent可调用的外部功能
2. 记忆(Memory)：保存对话历史和上下文
3. 策略(Policy)：决定下一步动作的逻辑
4. 执行器(Executor)：实际调用工具和模型

LangChain提供了多种预置Agent类型，如Zero-shot ReAct、Conversational等，开发者也可自定义Agent逻辑。

3. LangChain高级用法

3.1 聊天模型调用

LangChain支持多种聊天模型调用方式，从基础到高级依次为：

1. 直接调用：最基础的方式，适合简单场景

from langchain.chat_models import ChatOpenAI chat = ChatOpenAI(model="gpt-3.5-turbo") response = chat.invoke("Hello!")

2. 消息历史管理：维护对话上下文

from langchain.schema import HumanMessage, AIMessage messages = [ HumanMessage(content="你好！"), AIMessage(content="你好，有什么可以帮您的吗？"), HumanMessage(content="我想了解RAG技术") ] response = chat.invoke(messages)

3. 流式响应：处理长文本生成

for chunk in chat.stream("请解释RAG技术"): print(chunk.content, end="", flush=True)

3.2 简写形式优化

为提高开发效率，LangChain提供了多种简写形式：

1. 管道操作符：简化链式调用

from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser prompt = ChatPromptTemplate.from_template("总结以下文本：{text}") chain = prompt | chat | StrOutputParser() result = chain.invoke({"text": "长篇文章内容..."})

2. LCEL(LangChain Expression Language)：声明式定义复杂流程

from langchain.schema.runnable import RunnablePassthrough retriever = vectorstore.as_retriever() template = """基于以下上下文回答问题： {context} 问题：{question} """ prompt = ChatPromptTemplate.from_template(template) rag_chain = { "context": retriever, "question": RunnablePassthrough() } | prompt | chat | StrOutputParser()

3. 装饰器语法：快速定义工具

from langchain.agents import tool @tool def search_web(query: str) -> str: """执行网络搜索""" # 实际搜索实现 return "搜索结果..."

4. 实战案例：构建RAG增强的Agent

4.1 系统架构设计

我们将构建一个结合RAG和Agent技术的智能问答系统，架构如下：

1. 前端接收用户问题
2. Agent决策是否需要检索外部知识
3. 如需检索，调用RAG流程获取相关信息
4. 语言模型综合所有信息生成回答
5. 记录交互历史供后续参考

4.2 关键代码实现

from langchain.agents import AgentExecutor, Tool, create_openai_tools_agent from langchain import hub # 定义工具 tools = [ Tool( name="Knowledge Base", func=rag_chain.invoke, description="用于查询产品知识库" ), search_web # 前面定义的搜索工具 ] # 创建Agent prompt = hub.pull("hwchase17/openai-tools-agent") agent = create_openai_tools_agent(chat, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 执行查询 result = agent_executor.invoke({ "input": "你们产品的最新版本有什么新功能？", "chat_history": [] # 可传入历史消息 })

4.3 性能优化技巧

1. 批量处理：对多个查询同时执行检索

from langchain.schema.runnable import RunnableParallel parallel = RunnableParallel({ "query1": rag_chain, "query2": rag_chain }) parallel.invoke({ "query1": {"question": "问题1"}, "query2": {"question": "问题2"} })

2. 缓存机制：减少重复计算

from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())

3. 异步调用：提高吞吐量

async def run_queries(): results = await agent_executor.abatch([ {"input": "问题1"}, {"input": "问题2"} ])

5. 常见问题与解决方案

5.1 检索质量不佳

症状：返回的文档片段与问题无关解决方案：

• 调整文本分割策略（chunk_size和overlap）
• 尝试不同嵌入模型（如text-embedding-3-large）
• 添加元数据过滤条件

text_splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=150, separators=["\n\n", "\n", "。", "！", "？"] )

5.2 Agent决策错误

症状：该检索时未检索，或错误调用工具解决方案：

• 优化工具描述（清晰说明适用场景）
• 调整提示词模板
• 设置温度参数（temperature=0可获得更稳定结果）

agent = create_openai_tools_agent( ChatOpenAI(temperature=0.2), tools, prompt )

5.3 响应速度慢

优化策略：

1. 预加载向量数据库
2. 使用更轻量级的嵌入模型
3. 实现分级缓存（内存+Redis）
4. 限制检索文档数量

retriever = vectorstore.as_retriever( search_kwargs={"k": 3} # 只返回最相关的3个片段 )

6. 进阶技巧与最佳实践

6.1 多Agent协作

对于复杂任务，可通过LangGraph实现多Agent协作：

from langgraph.graph import Graph from langgraph.prebuilt import ToolNode workflow = Graph() workflow.add_node("research_agent", research_agent) workflow.add_node("writing_agent", writing_agent) workflow.add_edge("research_agent", "writing_agent") workflow.set_entry_point("research_agent") chain = workflow.compile()

6.2 评估与监控

建立评估体系确保系统质量：

1. 检索评估：命中率、相关性评分
2. 生成评估：事实准确性、流畅度
3. 端到端评估：用户满意度、任务完成率

from langchain.evaluation import load_evaluator evaluator = load_evaluator("labeled_score_string", criteria="correctness") eval_result = evaluator.evaluate_strings( prediction=agent_response, input=user_question, reference=golden_answer )

6.3 生产环境部署

关键考虑因素：

1. 容器化：使用Docker打包应用
2. 弹性伸缩：根据负载自动调整实例数
3. 监控告警：跟踪延迟、错误率等指标
4. 回滚机制：快速恢复到稳定版本

注意：生产环境建议使用专门的向量数据库服务（如Milvus集群），而非本地Chroma

在实际项目中，我发现合理设置超时参数可以显著提高系统稳定性。以下是我的推荐配置：

agent_executor = AgentExecutor( agent=agent, tools=tools, max_iterations=5, early_stopping_method="generate", handle_parsing_errors=True )

对于需要长期维护的项目，建议建立完整的CI/CD流程，包括自动化测试、性能基准和版本管理。特别要注意模型版本固化，避免因模型更新导致意外行为变化。