LangChain 实操指南：从零构建生产级 LLM 应用

文章导航前言为什么你需要 LangChain第一章环境搭建与核心概念第二章Model I/O —— 与大模型对话的基础第三章检索增强生成RAG—— 让LLM访问你的私有数据第四章Chain —— 构建多步骤工作流第五章Agent —— 让LLM自主使用工具第六章生产部署要点避坑指南终章一个完整的端到端示例前言为什么你需要 LangChain2026年大语言模型已经无处不在。但一个残酷的现实是仅靠API调用GPT你只能做Demo做不了产品。真正的价值在于——让LLM与外部工具交互、访问私有数据、执行多步骤推理。这正是LangChain登场的理由。LangChain不是又一个SDK而是一套构建LLM驱动应用的完整框架。它抽象了Prompt管理、输出解析、检索增强、Agent决策等核心环节让你从调API工程师进化到AI应用架构师。本文不会面面俱到地罗列API文档而是从一个真实的企业知识库问答系统需求出发带你走通从环境搭建到生产部署的全链路。每一段代码都经过验证可直接运行。第一章环境搭建与核心概念1.1 安装与环境准备# 推荐使用Python 3.11 pip install langchain langchain-openai langchain-community 如需本地模型本文后期会用到 pip install langchain-ollama 向量数据库用于RAG pip install chromadb设置你的OpenAI API Key或使用国内代理import os os.environ[OPENAI_API_KEY] your-api-key # 如使用代理 os.environ[OPENAI_API_BASE] https://your-proxy.com/v11.2 LangChain 的三大核心抽象在动手写代码前先建立一张认知地图抽象层作用类比Model I/O管理LLM的输入输出类似数据库的CRUDRetrieval从外部数据源检索信息类似搜索引擎的索引与查询CompositionChain/Agent串联多个步骤形成工作流类似Airflow的DAG或微服务编排LangChain表达式语言LCEL是贯穿所有组件的通用语它用|管道操作符将各环节串联起来简洁且可观测。第二章Model I/O —— 与大模型对话的基础2.1 统一多模型调用接口LangChain最实用的价值之一一套代码切换GPT、Claude、Llama或国产模型。from langchain_openai import ChatOpenAI from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage, SystemMessage 云端模型 gpt ChatOpenAI(modelgpt-4o-mini, temperature0.7) 本地模型需先运行 ollama pull llama3.2 llama ChatOllama(modelllama3.2, temperature0) 统一的消息格式 messages [ SystemMessage(content你是一位精通Python的性能优化专家), HumanMessage(content如何快速判断一个列表是否包含重复元素给出三种方法并比较时间复杂度) ] response gpt.invoke(messages) print(response.content)关键点invoke()是LangChain所有组件的统一入口未来还会用到stream()流式输出和batch()批量处理。2.2 PromptTemplate —— 告别字符串拼接from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一位{role}擅长用{style}风格解释复杂概念), (human, 请用{word_count}个字以内解释{topic}) ]) chain prompt | gpt result chain.invoke({ role: 物理老师, style: 幽默风趣, word_count: 50, topic: 量子纠缠 }) print(result.content)2.3 输出解析器 —— 让AI输出JSON/结构化数据很多初学者会踩的坑让AI输出JSON结果它返回了Markdown代码块。用输出解析器解决from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field 定义你期望的强类型结构 class CodeReview(BaseModel): bug_risk: int Field(description潜在Bug风险评分1-10) suggestions: list[str] Field(description改进建议列表) refactored_code: str Field(description重构后的代码片段) parser PydanticOutputParser(pydantic_objectCodeReview) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严格的代码评审专家。{format_instructions}), (human, 请评审以下Python代码\npython\n{code}\n) ]).partial(format_instructionsparser.get_format_instructions()) chain prompt | gpt | parser # 自动解析为CodeReview对象 review chain.invoke({code: def add(a,b): return ab}) print(f风险评分: {review.bug_risk}) print(f建议: {review.suggestions[0]})第三章检索增强生成RAG—— 让LLM访问你的私有数据这是LangChain最广泛的应用场景。我们以公司内部文档问答为例。3.1 数据加载与切分from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter 加载指定目录下的所有txt文件 loader DirectoryLoader(./docs/, glob**/*.txt, loader_clsTextLoader) docs loader.load() 智能切分按段落、句子递归切割保留语义完整性 splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, # 重叠部分防止语义断层 separators[\n\n, \n, 。, , , , ] ) chunks splitter.split_documents(docs) print(f原始文档{len(docs)}份切分为{len(chunks)}个片段)3.2 向量化与存储from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma 生成向量嵌入也可用本地模型如BGE embeddings OpenAIEmbeddings(modeltext-embedding-3-small) 存入向量数据库 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db ) 持久化保存下次直接加载 vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings)3.3 构建RAG链 —— 检索 生成from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough 创建检索器 retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 返回最相似的4个片段 模板让LLM基于检索到的上下文回答 template 你是一位专业助手。请基于以下【上下文】回答用户问题。 如果上下文不足以回答问题请直接说根据现有文档无法回答该问题。 【上下文】 {context} 【用户问题】 {question} 请给出准确、简洁的回答 prompt ChatPromptTemplate.from_template(template) LCEL方式构建RAG链 rag_chain ( { context: retriever | (lambda docs: \n\n.join(d.page_content for d in docs)), question: RunnablePassthrough() } | prompt | gpt ) 执行查询 question 公司的年假制度是怎样的 response rag_chain.invoke(question) print(response.content)这一小段代码就是RAG的核心不到20行实现了检索→拼接→生成的完整链路。第四章Chain —— 构建多步骤工作流4.1 简单的顺序链from langchain_core.runnables import RunnableLambda 步骤1提取关键词 extract_prompt ChatPromptTemplate.from_template(提取以下文本中的3个核心关键词用逗号分隔\n{text}) extract_chain extract_prompt | gpt 步骤2基于关键词生成摘要 summary_prompt ChatPromptTemplate.from_template(基于关键词[{keywords}]为以下文本写一段100字摘要\n{text}) summary_chain summary_prompt | gpt 组合成顺序链注意这里用RunnablePassthrough传递原始text full_chain ( RunnablePassthrough.assign(keywordsextract_chain) | summary_chain ) result full_chain.invoke({text: 你的长文本内容})4.2 分支路由链Router Chain# 根据问题类型路由到不同的处理链 router_template 根据问题内容分类为以下之一[技术, 财务, 人事]。 只输出分类名称。 问题{question} router_chain ChatPromptTemplate.from_template(router_template) | gpt def route_output(output): category output.content.strip() if category 技术: return tech_chain # 假设已定义 elif category 财务: return finance_chain else: return hr_chain 使用RunnableBranch实现条件路由略思路如上第五章Agent —— 让LLM自主使用工具Agent是LangChain最智能的部分。LLM不再是一次性回答而是循环思考→决定调用工具→观察结果→继续思考直至完成任务。5.1 定义工具Toolsfrom langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper from langchain_core.tools import tool 内置工具 search DuckDuckGoSearchRun() wikipedia WikipediaQueryRun(api_wrapperWikipediaAPIWrapper()) 自定义工具使用tool装饰器 tool def calculate(expression: str) - str: 安全地计算数学表达式如23*4 try: return str(eval(expression)) except: return 表达式无效 tool def get_current_time() - str: 获取当前日期和时间 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) tools [search, wikipedia, calculate, get_current_time]5.2 创建并运行Agentfrom langchain.agents import create_tool_calling_agent, AgentExecutor 使用OpenAI的工具调用能力推荐 agent create_tool_calling_agent(gpt, tools, ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以调用工具获取信息。), (human, {input}), (placeholder, {agent_scratchpad}) ])) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) 执行复杂任务 result agent_executor.invoke({ input: 现在几点了然后帮我查一下明天北京天气最后计算一下3.14*5的平方 }) print(result[output])Agent会依次调用get_current_time→ 搜索天气 →calculate并在每一步观察结果后决定下一步。第六章生产部署要点避坑指南6.1 回调与可观测性from langchain_core.callbacks import StdOutCallbackHandler 调试时打印详细日志 chain.invoke(input_data, config{callbacks: [StdOutCallbackHandler()]}) 生产环境建议接入LangSmithLangChain官方监控平台 os.environ[LANGCHAIN_API_KEY] your-key chain.invoke(input_data, config{run_name: my-rag-chain})6.2 异步与流式处理# 异步调用 response await chain.ainvoke(input_data) 流式输出用户体验更好 async for chunk in chain.astream(input_data): print(chunk.content, end, flushTrue) 批量处理高吞吐场景 results chain.batch([{question: q1}, {question: q2}])6.3 缓存与成本控制from langchain_core.caches import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache()) # 相同输入直接返回缓存结果 生产环境可用Redis缓存langchain_community.cache.RedisCache6.4 常见陷阱问题解决方案Token超长使用RecursiveCharacterTextSplitter合理切分设置max_tokens_limit检索返回无关片段增加k值或尝试MMR检索策略search_typemmrAgent陷入死循环设置max_iterations10或使用create_structured_chat_agent