LangChain与AI Agent开发实战：从工具调用到企业级RAG系统构建 LangChain和Agent开发已经成为当前AI应用开发的核心技术栈无论是构建智能问答系统、自动化工作流还是企业级AI助手都离不开这套强大的工具链。这次我们深入探讨如何从零开始构建功能完整的AI Agent涵盖天气查询、RAG知识库、记忆系统等实战案例帮你快速掌握企业级AI应用开发能力。从实际项目经验来看一个成熟的AI Agent需要具备工具调用、知识检索、状态记忆等核心能力。LangChain作为最流行的AI应用开发框架提供了完整的组件库和开发范式而Agent模式则让AI具备了自主决策和执行任务的能力。本文将带你从环境搭建到项目实战完整走通AI Agent的开发全流程。1. LangChain与Agent开发核心能力速览能力项技术说明应用价值工具调用集成外部API、数据库、计算工具让AI具备执行实际任务的能力RAG检索向量数据库语义搜索为企业知识库提供智能问答记忆系统对话历史管理和状态保持实现多轮连贯对话子Agent协调任务分解和并行处理处理复杂工作流多模型支持OpenAI、Anthropic、Google、本地模型灵活适配不同场景需求部署方式本地服务、API接口、云部署支持生产环境使用硬件要求大多数LangChain应用可以在CPU环境下运行涉及大模型推理时建议8GB以上显存。本文案例均支持CPU推理适合个人开发者和小团队验证。2. 环境准备与依赖安装开始前需要准备Python 3.8环境建议使用conda或venv创建独立的Python环境。2.1 基础环境配置# 创建并激活虚拟环境 conda create -n langchain-agent python3.10 conda activate langchain-agent # 安装核心依赖 pip install langchain langchain-community langchain-core pip install openai anthropic google-generativeai2.2 向量数据库选择根据项目需求选择合适的向量数据库# 轻量级选择 - Chroma本地部署 pip install chromadb langchain-chroma # 生产级选择 - Qdrant支持分布式 pip install qdrant-client langchain-qdrant # 云服务选择 - Pinecone免运维 pip install pinecone-client langchain-pinecone2.3 模型API配置配置各大模型平台的API密钥import os from getpass import getpass # OpenAI配置 os.environ[OPENAI_API_KEY] getpass(Enter OpenAI API key: ) # 备用方案 - 本地模型Ollama pip install ollama ollama pull llama2 # 下载本地模型3. 基础Agent开发天气查询助手我们先从最简单的工具调用开始构建一个天气查询Agent。3.1 工具函数定义import requests from langchain.tools import tool from langchain.agents import AgentType, initialize_agent from langchain.chat_models import ChatOpenAI tool def get_weather(city: str) - str: 获取指定城市的天气信息 try: # 模拟天气API调用实际项目中替换为真实API response requests.get(fhttps://api.weather.com/{city}) return f{city}天气晴25°C # 模拟返回 except: return f无法获取{city}的天气信息 # 初始化模型和工具 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [get_weather] # 创建Agent weather_agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue )3.2 测试天气查询功能# 测试Agent result weather_agent.run(今天北京天气怎么样) print(result) # 预期输出 # 我将为您查询北京的天气信息... # 北京天气晴25°C3.3 工具调用原理分析Agent执行流程理解意图LLM分析用户问题识别需要调用天气工具参数提取从问题中提取城市名称北京工具执行调用get_weather函数获取天气数据结果整合LLM将工具返回结果组织成自然语言响应4. RAG系统构建智能文档问答AgentRAG检索增强生成是LangChain最核心的应用场景下面构建一个文档问答系统。4.1 文档加载和预处理from langchain.document_loaders import TextLoader, WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma # 文档加载示例 def load_documents(source_typeweb, source_pathNone): if source_type web: loader WebBaseLoader([https://langchain.com/docs]) else: loader TextLoader(source_path) documents loader.load() # 文本分割 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) return text_splitter.split_documents(documents) # 创建向量数据库 def create_vector_store(docs, embedding_modelopenai): embeddings OpenAIEmbeddings() vector_store Chroma.from_documents(docs, embeddings) return vector_store4.2 RAG检索工具实现from langchain.tools import tool from langchain.schema import Document tool def search_documentation(query: str) - str: 在文档库中搜索相关信息 # 相似度搜索 docs vector_store.similarity_search(query, k3) # 格式化结果 result 检索到的相关信息\n for i, doc in enumerate(docs, 1): result f{i}. {doc.page_content[:200]}...\n return result # 更新Agent工具集 rag_tools [search_documentation] rag_agent initialize_agent( toolsrag_tools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue )4.3 高级RAG模式Deep Agent架构基于LangChain官方文档的Deep Agent模式实现更复杂的文档处理流程import uuid from deepagents import create_deep_agent from deepagents.backends import StateBackend class AdvancedRAGSystem: def __init__(self, vector_store, llm): self.vector_store vector_store self.llm llm self.backend StateBackend() def search_and_store(self, query: str, k: int 4) - list: 搜索文档并存储到临时文件系统 retrieved_docs self.vector_store.similarity_search(query, kk) batch_id uuid.uuid4().hex[:8] saved_paths [] for index, doc in enumerate(retrieved_docs, start1): path f/retrieved/{batch_id}/chunk_{index}.md content f# Source: {doc.metadata.get(source, unknown)}\n\n{doc.page_content} # 存储到文件系统 self.backend.upload_files([(path, content.encode(utf-8))]) saved_paths.append(path) return saved_paths def create_rag_agent(self): 创建具备子Agent协调能力的RAG系统 # 子Agent定义 - 文档分析专家 chunk_analyst { name: chunk-analyst, description: 分析单个文档片段提取关键信息, system_prompt: 你负责分析检索到的文档片段。提取与用户问题相关的事实、API说明和配置细节。保持回答简洁专业。 } # 主Agent系统提示词 system_prompt # 文档智能问答工作流 1. **规划**分析用户问题生成精准的搜索查询 2. **检索**搜索文档库获取相关片段 3. **分析**委托子Agent并行分析各个文档片段 4. **综合**整合所有分析结果生成完整答案 5. **验证**检查答案是否完整必要时进行二次检索 return create_deep_agent( modelself.llm, tools[self.search_and_store], backendself.backend, system_promptsystem_prompt, subagents[chunk_analyst] )5. 记忆系统实现多轮对话状态管理记忆系统是Agent实现连贯对话的关键下面实现两种记忆模式。5.1 基于对话历史的记忆from langchain.memory import ConversationBufferWindowMemory from langchain.agents import AgentExecutor # 创建带记忆的Agent memory ConversationBufferWindowMemory( memory_keychat_history, k5, # 保留最近5轮对话 return_messagesTrue ) agent_with_memory AgentExecutor.from_agent_and_tools( agentweather_agent.agent, toolstools, memorymemory, verboseTrue ) # 测试多轮对话 result1 agent_with_memory.run(北京天气怎么样) result2 agent_with_memory.run(那上海呢) # Agent能理解上海指代天气查询5.2 实体记忆和状态跟踪from langchain.memory import EntityMemory from langchain.schema import BaseMemory class CustomEntityMemory(BaseMemory): 自定义实体记忆系统跟踪重要实体信息 def __init__(self): self.entity_store {} def save_context(self, inputs, outputs): # 从对话中提取实体信息并存储 user_input inputs.get(input, ) ai_output outputs.get(output, ) # 简单的实体提取逻辑实际项目可用NER模型 if 北京 in user_input: self.entity_store[current_city] 北京 elif 上海 in user_input: self.entity_store[current_city] 上海 def load_memory_variables(self, inputs): return self.entity_store property def memory_variables(self): return [current_city]6. 完整案例企业知识库问答系统整合以上技术构建一个完整的企业级问答系统。6.1 系统架构设计class EnterpriseQASystem: def __init__(self, knowledge_base_path, llm_modelgpt-3.5-turbo): self.llm ChatOpenAI(modelllm_model, temperature0) self.knowledge_base self.setup_knowledge_base(knowledge_base_path) self.memory ConversationBufferWindowMemory(k10) self.tools self.setup_tools() def setup_knowledge_base(self, path): 初始化知识库 documents load_documents(local, path) return create_vector_store(documents) def setup_tools(self): 配置工具集 tool def knowledge_search(query: str) - str: 企业知识库搜索 docs self.knowledge_base.similarity_search(query, k5) return \n\n.join([doc.page_content for doc in docs]) tool def escalation_procedure(issue_type: str) - str: 问题升级流程 procedures { technical: 联系技术支持团队supportcompany.com, billing: 联系财务部门billingcompany.com, general: 联系客服中心helpcompany.com } return procedures.get(issue_type, 请联系管理员) return [knowledge_search, escalation_procedure] def create_agent(self): 创建企业级问答Agent return initialize_agent( toolsself.tools, llmself.llm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, memoryself.memory, verboseTrue, handle_parsing_errorsTrue )6.2 系统测试和验证# 初始化系统 qa_system EnterpriseQASystem(./company_docs/) agent qa_system.create_agent() # 测试用例 test_cases [ 我们公司的请假流程是什么, 技术问题应该联系谁, 上次你告诉我请假需要审批具体要多久 ] for case in test_cases: print(f用户: {case}) response agent.run(case) print(fAgent: {response}\n)7. 性能优化和最佳实践7.1 向量检索优化# 优化检索效果 def optimized_search(query, vector_store, k3, score_threshold0.7): 带分数阈值的检索 results vector_store.similarity_search_with_score(query, kk) # 过滤低质量结果 filtered [doc for doc, score in results if score score_threshold] return filtered if filtered else [doc for doc, score in results[:1]] # 多路检索策略 def multi_retrieval(query, vector_store): 结合多种检索策略 # 1. 直接相似度搜索 direct_results vector_store.similarity_search(query, k2) # 2. 最大边际相关性避免重复 mmr_results vector_store.max_marginal_relevance_search(query, k2) # 3. 结合两种结果 combined list({doc.page_content: doc for doc in direct_results mmr_results}.values()) return combined7.2 Agent响应优化# 定制化提示词模板 CUSTOM_PROMPT 你是一个专业的企业知识库助手。请遵循以下准则 1. 基于检索到的信息回答问题不要编造 2. 如果信息不足明确说明需要补充什么 3. 回答要简洁专业避免冗长 4. 涉及流程的问题提供具体步骤 5. 记得利用对话历史提供连贯服务 对话历史 {chat_history} 问题{input} 思考过程{agent_scratchpad} # 配置优化后的Agent from langchain.agents import AgentExecutor, create_structured_chat_agent def create_optimized_agent(tools, llm, memory): prompt CustomPromptTemplate( templateCUSTOM_PROMPT, toolstools, input_variables[input, chat_history, agent_scratchpad] ) agent create_structured_chat_agent(llm, tools, prompt) return AgentExecutor.from_agent_and_tools( agentagent, toolstools, memorymemory, verboseTrue )8. 部署和生产环境考虑8.1 本地API服务部署from fastapi import FastAPI from pydantic import BaseModel import uvicorn app FastAPI() class QueryRequest(BaseModel): question: str user_id: str None app.post(/api/ask) async def ask_question(request: QueryRequest): 提供问答服务的API接口 try: # 根据user_id获取对应的Agent实例 agent get_agent_for_user(request.user_id) response agent.run(request.question) return { success: True, answer: response, timestamp: datetime.now().isoformat() } except Exception as e: return { success: False, error: str(e), timestamp: datetime.now().isoformat() } if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)8.2 监控和日志记录import logging from langchain.callbacks import FileCallbackHandler # 配置详细日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(agent_operations.log), logging.StreamHandler() ] ) # LangChain回调处理器 file_callback FileCallbackHandler(langchain_trace.json) def create_agent_with_monitoring(tools, llm): 创建带监控的Agent return initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, callbacks[file_callback] )9. 常见问题排查指南9.1 工具调用失败排查# 工具调用调试函数 def debug_tool_execution(agent, query): 调试Agent的工具调用过程 print(f调试查询: {query}) try: # 分步执行查看中间状态 agent.agent.llm_chain.verbose True result agent.run(query) print(f执行结果: {result}) except Exception as e: print(f错误信息: {e}) # 检查工具配置 print(可用工具:, [tool.name for tool in agent.tools])9.2 内存管理问题# 记忆系统诊断 def diagnose_memory_issues(agent): 诊断记忆相关的问题 if hasattr(agent, memory): history agent.memory.load_memory_variables({}) print(当前记忆内容:, history) else: print(Agent未配置记忆系统) # 检查记忆长度限制 if hasattr(agent.memory, k): print(f记忆窗口大小: {agent.memory.k})9.3 性能瓶颈分析import time from functools import wraps def timing_decorator(func): 执行时间测量装饰器 wraps(func) def wrapper(*args, **kwargs): start time.time() result func(*args, **kwargs) end time.time() print(f{func.__name__} 执行时间: {end - start:.2f}秒) return result return wrapper # 应用性能监控 timing_decorator def monitored_agent_run(agent, query): return agent.run(query)10. 进阶开发方向10.1 多Agent协作系统from langchain.agents import AgentExecutor, Tool from langchain import LLMChain class MultiAgentSystem: 多Agent协作系统 def __init__(self): self.specialist_agents { technical: self.create_technical_agent(), billing: self.create_billing_agent(), general: self.create_general_agent() } self.router_agent self.create_router_agent() def route_question(self, query): 路由问题到合适的专家Agent # 使用router Agent判断问题类型 analysis self.router_agent.run(f分析问题类型: {query}) if 技术 in analysis or 故障 in analysis: return self.specialist_agents[technical] elif 账单 in analysis or 支付 in analysis: return self.specialist_agents[billing] else: return self.specialist_agents[general]10.2 持续学习和知识更新class SelfImprovingAgent: 具备自我学习能力的Agent def __init__(self, vector_store): self.vector_store vector_store self.feedback_log [] def learn_from_feedback(self, query, response, user_feedback): 从用户反馈中学习 self.feedback_log.append({ query: query, response: response, feedback: user_feedback, timestamp: datetime.now() }) # 基于负面反馈更新知识库 if user_feedback.get(rating) negative: self.update_knowledge_base(query, user_feedback.get(correction)) def update_knowledge_base(self, query, correction): 用正确信息更新知识库 if correction: new_doc Document( page_contentfQ: {query}\nA: {correction}, metadata{source: user_feedback, type: correction} ) self.vector_store.add_documents([new_doc])通过本文的完整学习路径你已经掌握了LangChain和Agent开发的核心技术。从简单的工具调用到复杂的RAG系统从基础记忆管理到多Agent协作这套技术栈能够支撑起企业级的AI应用需求。实际项目中建议从简单用例开始逐步增加复杂度。重点关注工具设计的合理性、检索效果优化和用户体验打磨。随着项目经验积累可以进一步探索Agent的自动化测试、性能监控和持续学习等高级特性。