引言在人工智能快速发展的今天，大语言模型（LLM）已经渗透到各个行业和领域。然而，单纯使用LLM往往难以满足复杂的业务需求，尤其是在需要多步骤推理、数据检索、外部工具调用等场景下。LangChain作为一个强大的框架，正是为了解决这些问题而生。本文将深入探讨如何利用LangChain搭建高效的工作流，帮助开发者充分利用LLM的能力，构建复杂的AI应用。什么是LangChain工作流LangChain是一个用于开发由语言模型驱动的应用程序的框架。它提供了一系列工具和抽象，使开发者能够轻松地将LLM与其他计算资源、数据源和API连接起来。工作流（Workflow）是LangChain的核心概念之一，指的是将多个组件（如提示模板、模型、数据检索器、工具等）组合成一个完整的处理流程。工作流的核心组件Prompt Templates：动态生成提示词，确保输入格式一致Language Models：核心的LLM调用，支持多种模型提供商Output Parsers：将模型输出解析为结构化数据Memory：维护对话历史或状态信息Tools：外部API、数据库、计算引擎等Chains：将上述组件串联成完整流程设计高效工作流的关键原则在开始编码之前，理解工作流设计的基本原则至关重要。以下是几个关键点：1. 模块化设计将复杂任务拆分为独立的、可复用的模块。每个模块应专注于单一功能，便于测试和维护。2. 错误处理与回退机制LLM调用可能失败或返回不理想的结果。设计工作流时应考虑：重试策略备选模型或提示降级处理方案3. 性能优化减少不必要的API调用使用缓存机制并行执行独立任务合理设置超时和并发限制4. 可观测性记录关键步骤的输入输出监控执行时间和资源消耗支持调试模式实践：搭建一个智能客服工作流让我们通过一个实际案例来展示如何搭建高效工作流。假设我们需要构建一个智能客服系统，能够回答产品相关问题、处理订单查询，并在必要时转接人工客服。步骤1：环境准备首先安装必要的依赖：pip install langchain langchain-openai langchain-community步骤2：定义基础组件from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.output_parsers import PydanticOutputParser from langchain.memory import ConversationBufferMemory from langchain.tools import tool from pydantic import BaseModel, Field # 初始化模型 llm = ChatOpenAI(model="gpt-4", temperature=0.2) # 定义输出解析器 class CustomerQuery(BaseModel): intent: str = Field(description="用户意图：product_query/order_query/complaint/other") urgency: int = Field(description="紧急程度：1-5") requires_human: bool = Field(description="是否需要人工介入") parser = PydanticOutputParser(pydantic_object=CustomerQuery) # 创建提示模板 intent_prompt = ChatPromptTemplate.from_messages([ ("system", "分析用户查询，提取意图和紧急程度。\n{format_instructions}"), ("human", "{input}") ]).partial(format_instructions=parser.get_format_instructions())步骤3：创建工具函数@tool def search_product(query: str) -> str: """搜索产品信息""" # 模拟产品搜索 products = { "笔记本电脑": "高性能笔记本电脑，配备i7处理器和16GB内存", "手机": "最新款智能手机，支持5G网络" } return products.get(query, "未找到相关产品信息") @tool def check_order(order_id: str) -> str: """查询订单状态""" # 模拟订单查询 return f"订单{order_id}状态：已发货，预计3天内到达" @tool def escalate_to_human(reason: str) -> str: """转接人工客服""" return f"已将问题转接至人工客服，原因：{reason}"步骤4：构建工作流链from langchain.chains import LLMChain from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain.prompts import MessagesPlaceholder # 创建意图识别链 intent_chain = LLMChain( llm=llm, prompt=intent_prompt, output_parser=parser ) # 创建主处理链 system_prompt = """你是一个专业的客服助手。根据用户查询： 1. 如果是产品查询，使用search_product工具 2. 如果是订单查询，使用check_order工具 3. 如果用户情绪激动或问题复杂，使用escalate_to_human工具 4. 始终保持友好和专业的语气""" agent = create_openai_tools_agent( llm=llm, tools=[search_product, check_order, escalate_to_human], prompt=ChatPromptTemplate.from_messages([ ("system", system_prompt), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ]) ) agent_executor = AgentExecutor( agent=agent, tools=[search_product, check_order, escalate_to_human], verbose=True, max_iterations=3, handle_parsing_errors=True )步骤5：集成记忆功能from langchain.memory import ConversationSummaryMemory memory = ConversationSummaryMemory( llm=llm, memory_key="chat_history", return_messages=True ) # 完整的工作流函数 async def process_customer_query(query: str): # 1. 意图识别 intent_result = await intent_chain.arun(input=query) # 2. 根据意图决定处理方式 if intent_result.requires_human or intent_result.urgency >= 4: return await escalate_to_human.arun( f"紧急查询（等级{intent_result.urgency}）：{query}" ) # 3. 使用agent处理 response = await agent_executor.arun( input=query, chat_history=memory.buffer ) # 4. 更新记忆 await memory.save_context( {"input": query}, {"output": response} ) return response步骤6：添加缓存和监控from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache import time import logging # 启用缓存 set_llm_cache(InMemoryCache()) # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 带监控的包装函数 async def monitored_process(query: str): start_time = time.time() try: result = await process_customer_query(query) elapsed = time.time() - start_time logger.info(f"查询处理完成，耗时：{elapsed:.2f}秒") return result except Exception as e: logger.error(f"处理失败：{str(e)}") return "抱歉，系统出现异常，请稍后再试或联系人工客服。"高级工作流模式1. 并行处理对于需要同时查询多个数据源的任务，可以使用并行处理：from langchain.callbacks import AsyncCallbackHandler import asyncio async def parallel_search(queries: list): tasks = [search_product.arun(q) for q in queries] return await asyncio.gather(*tasks)2. 条件分支根据中间结果动态调整处理路径：from langchain.chains import SequentialChain, TransformChain def check_sentiment(text: str) -> dict: # 简单的情感分析 negative_words = ["差", "糟糕", "投诉", "不满意"] is_negative = any(word in text for word in negative_words) return {"is_negative": is_negative, "text": text} sentiment_chain = TransformChain( input_variables=["text"], output_variables=["is_negative", "text"], transform=check_sentiment ) # 条件路由 if sentiment_result["is_negative"]: # 转接优先处理 pass else: # 正常处理 pass3. 循环与迭代对于需要多轮对话或逐步推理的任务：from langchain.chains import LLMChain class IterativeRefinement: def __init__(self, max_iterations=3): self.max_iterations = max_iterations self.refinement_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_template( "当前答案：{current_answer}\n" "用户反馈：{feedback}\n" "请改进答案：" ) ) async def refine(self, initial_answer, get_feedback): answer = initial_answer for i in range(self.max_iterations): feedback = await get_feedback(answer) if feedback.get("is_satisfied"): break answer = await self.refinement_chain.arun( current_answer=answer, feedback=feedback["message"] ) return answer性能优化技巧1. 提示词优化使用具体的指令而非模糊的描述提供示例以引导模型输出格式限制输出长度以避免不必要的token消耗2. 模型选择策略# 简单任务使用快速模型 simple_tasks_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 复杂任务使用强大模型 complex_tasks_llm = ChatOpenAI(model="gpt-4", temperature=0.3) # 根据任务复杂度动态选择 def get_appropriate_model(task_complexity: int): if task_complexity < 3: return simple_tasks_llm else: return complex_tasks_llm3. 批量处理将多个独立请求合并处理，减少API调用次数：from langchain.schema import HumanMessage, SystemMessage async def batch_process(queries: list): # 合并多个查询 combined_prompt = "\n".join([f"{i+1}. {q}" for i, q in enumerate(queries)]) messages = [ SystemMessage(content="请依次回答以下问题，用序号标记每个答案。"), HumanMessage(content=combined_prompt) ] response = await llm.agenerate([[messages]]) # 解析响应，分割成独立答案 answers = response.generations[0][0].text.split("\n") return answers测试与部署单元测试import pytest from unittest.mock import AsyncMock @pytest.mark.asyncio async def test_intent_recognition(): mock_llm = AsyncMock() mock_llm.arun.return_value = '{"intent": "product_query", "urgency": 2, "requires_human": false}' chain = intent_chain chain.llm = mock_llm result = await chain.arun(input="你们有笔记本电脑吗？") assert result.intent == "product_query" assert result.urgency == 2部署建议使用异步框架：如FastAPI构建REST API添加速率限制：防止API滥用实现健康检查：监控服务状态配置环境变量：管理API密钥等敏感信息使用容器化部署：确保环境一致性结语LangChain为开发者提供了强大的工作流搭建能力，但高效的实现需要深入理解其核心概念和最佳实践。本文通过一个智能客服的案例，展示了从基础组件搭建到高级模式应用的完整过程。关键要点包括：模块化设计是构建可维护系统的基石错误处理和缓存机制显著提升系统稳定性并行处理和条件分支扩展了工作流的能力边界性能优化需要在开发阶段就纳入考虑随着LangChain生态的不断发展，新的组件和模式将持续涌现。建议开发者保持学习，关注官方文档和社区实践，同时根据实际业务需求灵活调整工作流设计。记住，最好的工作流不是最复杂的，而是最贴合业务需求、最易于维护和扩展的那个。未来，随着多模态模型和Agent技术的发展，LangChain工作流将能处理更加复杂的任务，如视频分析、实时决策等。掌握这些核心方法，将为你在AI应用开发领域奠定坚实的基础。
引言在人工智能飞速发展的今天，大语言模型（LLM）如 GPT-4、Claude 等已经展现出惊人的能力。然而，单独使用这些模型往往面临诸多限制：无法访问实时数据、缺乏记忆能力、难以与外部工具交互。LangChain 应运而生，它是一个强大的框架，旨在帮助开发者将 LLM 与外部数据源、API 和工具无缝集成，构建复杂的 AI 应用。对于零基础的开发者来说，LangChain 可能听起来有些复杂，但事实上，它的设计理念非常直观——通过“链”（Chain）的方式将不同组件连接起来，让大模型能够完成更复杂的任务。本文将带你从零开始，逐步掌握 LangChain 的核心概念和开发技巧。什么是 LangChain？LangChain 是一个开源的 Python 框架，于 2022 年 10 月由 Harrison Chase 创建。它的核心目标是解决大语言模型在实际应用中的三个关键问题：数据连接：让模型能够访问私有数据或实时信息记忆管理：使对话具备上下文记忆能力工具集成：允许模型调用外部 API 和执行代码简单来说，LangChain 就像是一个“胶水”，把大模型、数据库、搜索引擎、计算器等组件粘合在一起，形成一个功能强大的 AI 应用。环境搭建与安装基本安装首先，确保你的 Python 版本在 3.8 以上。通过 pip 安装 LangChain：pip install langchain安装必要的依赖根据你的需求，你可能还需要安装以下组件：# OpenAI 集成 pip install openai # 文档加载器 pip install pypdf python-docx # 向量数据库 pip install chromadb faiss-cpu # 其他工具 pip install tiktoken配置 API 密钥以 OpenAI 为例，你需要设置 API 密钥：import os os.environ["OPENAI_API_KEY"] = "your-api-key-here"核心概念详解1. 模型（Models）LangChain 支持多种大语言模型，包括：LLM：基础的语言模型，如 GPT-3.5Chat Models：对话模型，如 GPT-4、ClaudeText Embedding Models：文本嵌入模型示例代码：from langchain.llms import OpenAI from langchain.chat_models import ChatOpenAI # 使用 LLM llm = OpenAI(model_name="gpt-3.5-turbo-instruct") response = llm("解释什么是机器学习") print(response) # 使用 Chat Model chat_model = ChatOpenAI(model_name="gpt-4") response = chat_model.predict("用简单的语言解释量子计算") print(response)2. 提示词模板（Prompts）提示词模板是 LangChain 的核心组件之一，它允许你创建可复用的提示结构：from langchain.prompts import PromptTemplate # 创建模板 template = """ 你是一位{role}专家。 请用{language}回答以下问题： {question} """ prompt = PromptTemplate( input_variables=["role", "language", "question"], template=template ) # 使用模板 formatted_prompt = prompt.format( role="人工智能", language="中文", question="什么是深度学习？" )3. 链（Chains）链是 LangChain 的核心概念，它将多个步骤串联起来：from langchain.chains import LLMChain from langchain.llms import OpenAI # 创建简单的链 llm = OpenAI(temperature=0.7) chain = LLMChain( llm=llm, prompt=prompt ) # 运行链 result = chain.run( role="数据科学", language="英文", question="什么是过拟合？" )4. 记忆（Memory）记忆组件让 LLM 能够记住对话历史：from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain # 创建带记忆的对话链 memory = ConversationBufferMemory() conversation = ConversationChain( llm=llm, memory=memory ) # 进行对话 response1 = conversation.predict(input="你好，我叫小明") response2 = conversation.predict(input="你还记得我的名字吗？")5. 文档加载与分割处理大量文本时，需要将文档分割成合适的块：from langchain.document_loaders import TextLoader from langchain.text_splitter import CharacterTextSplitter # 加载文档 loader = TextLoader("document.txt") documents = loader.load() # 分割文本 text_splitter = CharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) chunks = text_splitter.split_documents(documents)6. 向量存储与检索向量存储是实现 RAG（检索增强生成）的关键：from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma # 创建嵌入 embeddings = OpenAIEmbeddings() # 创建向量存储 vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings ) # 相似性搜索 query = "什么是机器学习" results = vectorstore.similarity_search(query, k=3)实战项目：构建知识问答系统现在，让我们用所学知识构建一个完整的知识问答系统。项目结构from langchain.chains import RetrievalQA from langchain.llms import OpenAI from langchain.document_loaders import PyPDFLoader from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter class KnowledgeQA: def __init__(self): self.llm = OpenAI(temperature=0) self.embeddings = OpenAIEmbeddings() self.vectorstore = None def load_documents(self, file_path): """加载文档""" loader = PyPDFLoader(file_path) documents = loader.load() # 分割文档 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=100 ) chunks = text_splitter.split_documents(documents) # 创建向量存储 self.vectorstore = Chroma.from_documents( documents=chunks, embedding=self.embeddings ) def ask_question(self, question): """回答问题""" if not self.vectorstore: return "请先加载文档" qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=self.vectorstore.as_retriever() ) return qa_chain.run(question) # 使用示例 qa_system = KnowledgeQA() qa_system.load_documents("ai_handbook.pdf") answer = qa_system.ask_question("什么是神经网络？") print(answer)高级特性探索1. 代理（Agents）代理可以自主决定使用哪些工具：from langchain.agents import initialize_agent, Tool from langchain.tools import WikipediaQueryRun # 定义工具 tools = [ Tool( name="Wikipedia", func=WikipediaQueryRun().run, description="搜索维基百科" ), Tool( name="Calculator", func=lambda x: eval(x), description="执行数学计算" ) ] # 创建代理 agent = initialize_agent( tools=tools, llm=llm, agent="zero-shot-react-description", verbose=True ) # 使用代理 agent.run("计算 2的10次方 并查找其历史背景")2. 回调（Callbacks）回调机制让你可以监控和调试链的执行过程：from langchain.callbacks import StdOutCallbackHandler handler = StdOutCallbackHandler() chain.run(input="问题", callbacks=[handler])最佳实践与注意事项性能优化合理设置 chunk 大小：过大会导致上下文溢出，过小会丢失语义使用缓存：对重复查询使用缓存减少 API 调用选择合适的模型：简单任务用 GPT-3.5，复杂任务用 GPT-4错误处理from langchain.schema import LLMResult try: result = chain.run(input) except Exception as e: print(f"错误发生: {e}") # 实现降级策略安全考虑不要在提示词中暴露敏感信息对用户输入进行过滤和验证设置合理的 token 限制总结通过本文的学习，你已经掌握了 LangChain 的核心概念和基本开发技能。从环境搭建到构建完整的知识问答系统，我们一步步探索了这个强大框架的方方面面。LangChain 的价值在于它让开发者能够：快速原型开发：无需从头实现复杂的 AI 逻辑组件化开发：各个模块可以独立测试和替换灵活扩展：支持自定义组件和第三方集成作为零基础的入门者，建议你按照以下路径继续学习：熟练掌握本文中的基础组件尝试构建更多实际应用（如聊天机器人、文档分析工具）深入理解 RAG、Agent 等高级概念关注社区动态，学习最新的最佳实践记住，AI 开发不仅仅是技术实现，更重要的是理解业务需求，设计合理的交互流程。LangChain 为你提供了强大的工具，真正的价值在于你如何运用它们解决实际问题。现在，打开你的编辑器，开始你的第一个 LangChain 项目吧！
引言在人工智能技术飞速发展的今天，大语言模型（LLM）已经成为推动应用创新的核心引擎。然而，单纯调用API远远无法满足复杂业务场景的需求——我们需要将LLM与外部数据源、工具和逻辑流程进行深度整合。这正是LangChain诞生的背景。LangChain是一个专为构建基于大语言模型的应用而设计的开源框架。它提供了一套标准化的接口和组件，帮助开发者快速搭建从简单聊天机器人到复杂智能代理的各种应用。截至2024年，LangChain在GitHub上已获得超过80,000颗星，成为AI应用开发领域最受欢迎的工具之一。本文将带你从零开始，系统掌握LangChain的核心概念、关键组件和实战技巧。无论你是AI初学者还是经验丰富的开发者，都能从中获得实用的知识和技能。一、LangChain核心概念解析1.1 什么是LangChain？LangChain本质上是一个抽象层，它封装了与LLM交互的复杂性，提供了模块化的构建块。它的设计哲学可以概括为三个关键词：组件化：将复杂功能拆解为可复用的独立模块可组合：通过链式调用将多个组件串联成完整流程可扩展：支持自定义组件和第三方集成1.2 LangChain的核心架构LangChain的架构可以分为四个层次：应用层（Applications） ↑ 链层（Chains） ↑ 代理层（Agents） ↑ 基础组件（Models, Prompts, Memory, Indexes）Models：封装各种LLM的调用接口Prompts：管理和优化提示词模板Memory：维护对话历史状态Indexes：管理和检索外部知识库Chains：将多个组件串联成工作流Agents：赋予LLM使用工具和执行决策的能力二、环境搭建与基础配置2.1 安装LangChain首先，确保你的Python版本在3.8以上，然后通过pip安装：pip install langchain pip install langchain-community # 社区贡献的集成 pip install langchain-openai # OpenAI集成2.2 配置API密钥推荐使用环境变量管理敏感信息：import os from dotenv import load_dotenv load_dotenv() # 设置OpenAI API密钥 os.environ["OPENAI_API_KEY"] = "your-api-key-here" # 或者直接在代码中设置 # import openai # openai.api_key = "your-api-key"2.3 创建第一个LangChain应用让我们从最简单的LLM调用开始：from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage # 初始化模型 llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0.7, max_tokens=500 ) # 发送消息 response = llm.invoke([HumanMessage(content="你好，请用中文介绍一下LangChain")]) print(response.content)三、核心组件深度实践3.1 Prompt模板管理好的提示词是LLM应用成功的关键。LangChain提供了强大的提示词管理工具：from langchain.prompts import PromptTemplate from langchain.prompts.chat import ChatPromptTemplate # 基础模板 template = """ 你是一位{role}专家。 请用{language}回答以下问题： {question} """ prompt = PromptTemplate( input_variables=["role", "language", "question"], template=template ) # 使用模板 formatted_prompt = prompt.format( role="Python", language="中文", question="如何优化列表推导式？" ) # 对话式模板 chat_template = ChatPromptTemplate.from_messages([ ("system", "你是一位专业的编程导师，擅长用通俗易懂的方式解释技术概念。"), ("human", "请解释{concept}"), ("ai", "好的，我来解释{concept}："), ("human", "能再深入一点吗？"), ]) messages = chat_template.format_messages(concept="递归函数")3.2 输出解析器为了从LLM的响应中提取结构化数据，我们可以使用输出解析器：from langchain.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field from typing import List # 定义数据结构 class CodeReview(BaseModel): issues: List[str] = Field(description="代码问题列表") suggestions: List[str] = Field(description="改进建议") score: int = Field(description="代码质量评分，1-10分") # 创建解析器 parser = PydanticOutputParser(pydantic_object=CodeReview) # 构建提示 template = """ 请审查以下代码并返回JSON格式的审查结果： {code} {format_instructions} """ prompt = PromptTemplate( template=template, input_variables=["code"], partial_variables={"format_instructions": parser.get_format_instructions()} ) # 使用 code = """ def add(a,b): return a+b """ _formatted_prompt = prompt.format_prompt(code=code) output = llm.invoke(_formatted_prompt.to_messages()) parsed_result = parser.parse(output.content)3.3 记忆系统记忆系统让LLM能够记住对话历史，实现上下文感知：from langchain.memory import ConversationBufferMemory from langchain.memory import ConversationSummaryMemory from langchain.memory import VectorStoreRetrieverMemory # 基础缓冲记忆 buffer_memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True ) # 摘要记忆（适用于长对话） summary_memory = ConversationSummaryMemory( llm=llm, memory_key="history", return_messages=True ) # 向量存储记忆（适用于海量历史） from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings vector_memory = VectorStoreRetrieverMemory( retriever=FAISS.from_texts( ["历史对话内容"], OpenAIEmbeddings() ).as_retriever(search_kwargs={"k": 3}), memory_key="relevant_history" )3.4 文档加载与索引处理外部文档是LangChain的核心能力之一：from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings # 加载文档 loader = PyPDFLoader("document.pdf") documents = loader.load() # 文本分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", " ", ""] ) chunks = text_splitter.split_documents(documents) # 创建向量存储 vectorstore = Chroma.from_documents( documents=chunks, embedding=OpenAIEmbeddings(), persist_directory="./chroma_db" ) # 检索 retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 3} )四、构建复杂工作流4.1 链（Chains）的使用链是LangChain的核心抽象，它允许我们将多个组件组合成完整的工作流：from langchain.chains import LLMChain from langchain.chains import SimpleSequentialChain from langchain.chains import RetrievalQA # 基础链 qa_chain = LLMChain( llm=llm, prompt=prompt, verbose=True ) # 检索增强生成（RAG）链 rag_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 或 "map_reduce", "refine", "map_rerank" retriever=retriever, return_source_documents=True, verbose=True ) # 顺序链 chain1 = LLMChain(llm=llm, prompt=prompt1) chain2 = LLMChain(llm=llm, prompt=prompt2) overall_chain = SimpleSequentialChain( chains=[chain1, chain2], verbose=True )4.2 智能代理（Agents）代理让LLM能够自主决定使用哪些工具来完成任务：from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool, tool from langchain_community.tools import DuckDuckGoSearchRun import requests from datetime import datetime # 定义工具 @tool def get_current_time(format: str = "%Y-%m-%d %H:%M:%S") -> str: """获取当前时间""" return datetime.now().strftime(format) @tool def search_web(query: str) -> str: """搜索网络信息""" search = DuckDuckGoSearchRun() return search.run(query) @tool def calculate(expression: str) -> str: """执行数学计算""" try: return str(eval(expression)) except Exception as e: return f"计算错误: {str(e)}" # 创建工具列表 tools = [ Tool( name="CurrentTime", func=get_current_time, description="获取当前时间" ), Tool( name="WebSearch", func=search_web, description="搜索网络信息" ), Tool( name="Calculator", func=calculate, description="执行数学计算" ) ] # 创建代理 from langchain.agents import create_openai_functions_agent from langchain_openai import ChatOpenAI agent = create_openai_functions_agent( llm=ChatOpenAI(model="gpt-4", temperature=0), tools=tools, prompt=prompt # 需要定义合适的提示模板 ) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=5, early_stopping_method="generate" ) # 运行代理 response = agent_executor.invoke({ "input": "现在几点了？帮我搜索一下今天的天气，然后计算7的阶乘" })五、高级应用场景5.1 构建RAG系统检索增强生成（RAG）是目前最流行的LLM应用模式：from langchain.chains import RetrievalQAWithSourcesChain from langchain.chains.combine_documents import create_stuff_documents_chain # 高级RAG配置 rag_chain = RetrievalQAWithSourcesChain.from_chain_type( llm=ChatOpenAI(model="gpt-4", temperature=0), chain_type="stuff", retriever=retriever, return_source_documents=True, chain_type_kwargs={ "prompt": CUSTOM_PROMPT, # 自定义提示 "document_variable_name": "context" } ) # 多查询检索 from langchain.retrievers import MultiQueryRetriever multi_query_retriever = MultiQueryRetriever.from_llm( retriever=retriever, llm=llm )5.2 流式输出与回调提升用户体验的关键技术：from langchain.callbacks import StreamingStdOutCallbackHandler from langchain.callbacks.base import BaseCallbackHandler class CustomCallbackHandler(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): print(f"开始处理: {prompts[0][:50]}...") def on_llm_end(self, response, **kwargs): print(f"处理完成，生成了{len(response.generations[0][0].text)}个字符") def on_llm_error(self, error, **kwargs): print(f"发生错误: {error}") # 流式输出 llm_stream = ChatOpenAI( model="gpt-3.5-turbo", streaming=True, callbacks=[StreamingStdOutCallbackHandler()] ) # 使用自定义回调 llm_with_callback = ChatOpenAI( model="gpt-3.5-turbo", callbacks=[CustomCallbackHandler()] )六、最佳实践与性能优化6.1 成本控制策略# 令牌限制 from langchain.callbacks import get_openai_callback with get_openai_callback() as cb: result = llm.invoke("生成一篇长文章") print(f"消耗令牌: {cb.total_tokens}") print(f"成本: ${cb.total_cost:.4f}") # 缓存机制 from langchain.cache import InMemoryCache import langchain langchain.llm_cache = InMemoryCache()6.2 错误处理与重试from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) def robust_llm_call(prompt): return llm.invoke(prompt)6.3 多模型策略# 模型路由 from langchain.chains import RouterChain router_chain = RouterChain.from_llm( llm=llm, chains={ "编程问题": code_chain, "文学创作": creative_chain, "数据分析": analysis_chain }, router_prompt=ROUTER_PROMPT )七、项目实战：智能客服系统让我们将所学知识整合成一个完整的智能客服系统：class SmartCustomerService: def __init__(self): self.llm = ChatOpenAI(model="gpt-4", temperature=0.3) self.memory = ConversationSummaryMemory( llm=self.llm, memory_key="history", return_messages=True ) self.retriever = self._init_knowledge_base() self.tools = self._init_tools() self.agent = self._init_agent() def _init_knowledge_base(self): # 加载产品文档 loader = TextLoader("product_docs.txt") docs = loader.load() splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100) chunks = splitter.split_documents(docs) vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings()) return vectorstore.as_retriever() def _init_tools(self): return [ Tool(name="KnowledgeBase", func=self.retriever.get_relevant_documents, description="查询产品知识库"), Tool(name="OrderLookup", func=self.lookup_order, description="查询订单状态"), Tool(name="RefundProcess", func=self.process_refund, description="处理退款请求") ] def handle_request(self, user_input): response = self.agent_executor.invoke({ "input": user_input, "chat_history": self.memory.load_memory_variables({}) }) self.memory.save_context({"input": user_input}, {"output": response["output"]}) return response["output"]八、部署与监控8.1 API服务部署from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Query(BaseModel): text: str session_id: str = None @app.post("/chat") async def chat(query: Query): # 处理请求 response = agent_executor.invoke({"input": query.text}) return {"response": response["output"]} # 使用 uvicorn main:app --host 0.0.0.0 --port 8000 启动8.2 性能监控from langchain.callbacks.tracers import LangChainTracer from langsmith import Client # 配置LangSmith监控 tracer = LangChainTracer( project_name="customer-service", client=Client(api_key="your-langsmith-api-key") ) # 在链或代理中使用 agent_executor = AgentExecutor( agent=agent, tools=tools, callbacks=[tracer] )九、常见陷阱与解决方案9.1 令牌溢出问题：当处理长文档时，容易超出模型令牌限制。解决方案：使用RecursiveCharacterTextSplitter合理分割文档采用map_reduce或refine链类型实施滑动窗口策略9.2 幻觉控制问题：LLM可能生成不准确或虚构的信息。解决方案：使用RAG架构，强制基于检索内容生成设置较低的温度参数（0-0.3）添加验证步骤，交叉检查关键信息9.3 性能瓶颈问题：链式调用导致响应延迟增加。解决方案：实现异步调用（asyncio）使用缓存减少重复计算考虑使用本地模型（如Llama）处理简单任务结论LangChain为构建基于LLM的应用提供了一个强大而灵活的框架。通过本文的实战指南，你已经掌握了从基础组件到高级应用的全栈技能：核心组件：模型、提示、记忆、索引的配置与使用工作流构建：链式调用、智能代理的创建与管理高级应用：RAG系统、流式输出、错误处理生产实践：部署、监控、性能优化记住，LangChain的真正价值不在于单个组件的使用，而在于如何创造性地组合这些组件来解决实际问题。建议你从简单的项目开始，逐步增加复杂度，在实践中深化理解。随着AI技术的快速发展，LangChain也在持续进化。保持学习，关注官方文档和社区动态，你就能始终站在AI应用开发的前沿。现在，是时候将你的创意付诸实践了——开始构建你的第一个LangChain应用吧！