LangChain 是大模型应用工程化里绕不开的一个框架。它解决的问题很直接把模型能力、提示词、知识库、外部工具和业务系统串成一条可维护的链路而不是每次从零拼接 Prompt 和模型调用。这次我们直接进入主题从示例代码分析开始把 LangChain 里最常用的几个能力全部过一遍RAG 知识库问答、ReAct Agent 推理、MapReduce 长文本处理、MCP 外部工具接入和工具调用。最后再从示例代码延伸到企业项目落地包括接口封装、批量任务设计和排查思路。这篇文章适合正在学 LangChain 的开发者也适合已经跑通 demo、准备把 RAG 或 Agent 放到生产流程里的同学。文章不会去讲太多概念史核心是把每个能力怎么用、怎么验证、落地时容易踩什么坑讲清楚。1. LangChain 核心能力速览能力项说明项目类型开源大模型应用编排框架由 LangChain 主导维护核心模块langchain-core、langchain、langchain-community、langchain-openai、langgraph 等主要功能Prompt 模板、模型调用、输出解析、RAG、工具调用、Agent、长文本 MapReduce、MCP 工具接入是否支持 API支持可以封装为 FastAPI 等 HTTP 服务供业务系统调用是否支持批量任务支持可基于脚本或队列做批量文档处理、批量问答运行平台跨平台需要 Python 环境模型可以走云端 API 也可以接本地模型启动方式非独立服务而是作为 Python 库集成进应用适合场景智能客服知识库、文档摘要、自动化 Agent、企业内部知识问答、数据分析助手硬件门槛取决于接入的模型用 OpenAI 等 API 不需要 GPU接本地模型则看模型参数量从表格里能看到LangChain 本身不是一个“一键启动的服务”它是一个开发框架。你要把它用到项目里通常需要自己写业务代码再选择底层的模型和向量库。2. 适用场景与使用边界LangChain 最适合的场景是那些需要把大模型和业务数据、外部工具组合起来完成的复杂任务。典型的例子包括企业知识库问答把内部文档切成片段向量化后检索让模型基于检索结果回答。长文档摘要合同、论文、财报这类超长文本直接塞给模型会超出上下文用 MapReduce 把长文档拆成多段分别摘要再合并。自动化 Agent模型根据用户问题决定调用哪些工具、按什么顺序调用最终完成任务。工具接入通过 Tool Calling 或 MCP 统一接入内部 API、数据库、第三方服务。这个框架不太适合的场景是只做一次“模型对话”的小工具。如果只是单轮调用 ChatGPT 接口直接用 OpenAI SDK 会更快不需要引入 LangChain。LangChain 的价值在链式编排、多步骤推理、长期记忆和复杂状态管理。使用边界方面要特别关注安全合规。RAG 如果接企业内部数据要注意权限隔离和敏感数据脱敏不能把全量文档一股脑发给外部模型 API。Agent 工具调用如果接内部系统必须做权限校验和操作审计。MCP 接入第三方服务时要防止越权访问和敏感信息泄露。任何涉及用户数据的内容都要先确认授权和合规要求。3. 环境准备与前置条件开始之前需要准备 Python 环境和模型访问凭证。这里的示例代码主要面向云端模型 API如果你要接本地模型可以把ChatOpenAI替换成兼容 OpenAI 协议的本地推理服务地址。推荐环境Python 3.9 及以上版本。建议使用虚拟环境避免依赖冲突。一个可用的模型 API Key例如 OpenAI 兼容接口。安装常用依赖langchain、langchain-community、langchain-openai、langchain-text-splitters、faiss-cpu、pypdf。如果要用 MCP需要额外安装 mcp 和 langchain-mcp-adapters。环境变量示例export OPENAI_API_KEYsk-xxxxxxxx export OPENAI_BASE_URLhttps://api.openai.com/v1如果你接的是本地模型比如基于 vLLM、Ollama 或 llama.cpp 启动的 OpenAI 兼容服务可以把OPENAI_BASE_URL改成http://127.0.0.1:8000/v1。这种部署方式尤其适合内网场景数据不需要离开本地环境。4. 安装部署与项目初始化安装命令建议一次性装齐常用依赖pip install langchain langchain-community langchain-openai langchain-text-splitters faiss-cpu pypdf如果你要写 API 服务再加安装 FastAPI 和 Uvicornpip install fastapi uvicorn创建项目目录时建议把模型配置、文档素材、输出结果分层管理project/ ├── config/ │ └── settings.py ├── docs/ │ └── knowledge/ ├── app/ │ ├── rag.py │ ├── agent.py │ └── api.py ├── data/ │ ├── vectorstore/ │ └── output/ └── requirements.txt初始化完成后先用一个最简单的模型调用验证环境是否可用。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(用一句话解释 LangChain) print(response.content)如果这一步能输出文字说明模型接入没有问题。如果报错优先检查 API Key、模型名称和网络连通性。5. 核心功能测试从示例代码到实战5.1 Prompt 模板与输出解析实际项目中不会直接裸调模型通常需要把 Prompt 模板和输出解析器组合起来。下面的例子把“解释概念”包装成一个可复用的链from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_messages([ (system, 你是一名资深技术顾问回答要准确、简洁。), (human, 请用中文解释什么是{concept}控制在200字以内。) ]) chain prompt | llm | StrOutputParser() result chain.invoke({concept: RAG}) print(result)这段代码里prompt | llm | StrOutputParser()就是 LangChain 的 LCEL 表达方式。管道符左边是输入右边是处理后输出逻辑清楚也方便后面替换或插入新节点。输出解析器的价值在于把模型返回的字符串转成结构化对象。实际业务中你可以用PydanticOutputParser直接拿到 JSON 格式的字段避免手写正则解析。5.2 工具调用工具调用是 Agent 的基础。模型本身不会执行计算但它可以在需要时返回一个“调用哪个工具、传入什么参数”的结构。from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个整数之和。 return a b tools [add] llm_with_tools llm.bind_tools(tools) response llm_with_tools.invoke(请计算 3 和 5 的和并使用工具) print(response.tool_calls)运行时response.tool_calls会输出类似[{name: add, args: {a: 3, b: 5}, id: ...}]的结构。这里的重点是LangChain 并不强制模型必须执行工具你依然需要自己写“拿到工具调用结果之后怎么继续”的逻辑。这也是 Agent 要解决的问题。5.3 ReAct Agent 推理ReAct 是一种 Agent 推理模式它的核心是让模型在“思考、行动、观察结果、再思考”之间循环直到得出结论。LangChain 里可以通过create_react_agent快速构建from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate react_prompt PromptTemplate.from_template( 你可以使用以下工具 {tools} 请严格按照下面的格式回答 Question: 用户输入的问题 Thought: 思考需要做什么 Action: 选择一个工具必须是 [{tool_names}] 中的一个 Action Input: 输入给工具的参数 Observation: 工具返回的结果 ... 重复若干次 Final Answer: 最终答案 Question: {input} Thought: ) agent create_react_agent(llm, tools, react_prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: 计算 5 7并把结果乘以 2}) print(result[output])这个示例里有两个工具才有意义。实际操作中Agent 会根据问题自行决定是否调用工具、调用哪个工具、调用几次。verboseTrue会让控制台打印完整的思考过程非常适合调试。新版本 LangChain 里create_react_agent是官方推荐的 Agent 构建方式。如果你看到很多老教程还在用initialize_agent建议新项目优先使用新 API必要时再按官方迁移文档调整。5.4 工具调用与 Agent 的边界很多人会把 Agent 和工具调用混淆。简单区分工具调用是模型输出“想用工具”的结构Agent 是拿到这个结构后真正去执行工具、观察结果、继续推理的完整循环。在项目里如果把这两层分开设计会更容易排查问题。先用llm.bind_tools(tools)验证模型是否能正确输出工具调用参数再引入 AgentExecutor 跑完整循环。这样可以快速定位是模型理解问题还是代码执行问题。5.5 RAG 知识库问答RAG 是当前企业项目里价值最高的 LangChain 应用。下面是一个基于 FAISS 和本地文档的完整示例。首先准备一个文本文件rag_example.txt内容随便写一段关于 LangChain 的介绍。然后执行from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough loader TextLoader(rag_example.txt) documents loader.load() splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) chunks splitter.split_documents(documents) embeddings OpenAIEmbeddings() vectorstore FAISS.from_documents(chunks, embeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) prompt ChatPromptTemplate.from_messages([ (system, 你是知识库问答助手请只根据上下文回答。如果上下文中没有答案请明确回答不知道。), (human, 上下文\n{context}\n\n问题{question}) ]) def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) answer chain.invoke(根据文档LangChain 的核心价值是什么) print(answer)这个链路里最值得关注的是文档切分。chunk_size500表示每块大概 500 个字符chunk_overlap50表示相邻块之间保留 50 个字符重叠。重叠能减少切断语义的问题但重叠太大会浪费 embedding 空间并影响检索精度。真正的企业项目里文档格式不会只有纯文本往往是 PDF、Word、Markdown 混在一起。建议根据文件类型选择对应的 Loader比如PyPDFLoader、Docx2txtLoader再统一走同一条切分和向量化流程。从实践角度看RAG 的效果不止取决于 LangChain 代码更取决于文档质量和切分策略。如果你检索出来的片段本身不相关模型再强也回答不好。所以第一步应该先检查检索结果而不是反复调 Prompt。5.6 MapReduce 长文本处理长文档摘要场景下MapReduce 是把文档切成多段、逐段摘要、再合并总结的经典思路。LangChain 老版本提供了现成的load_summarize_chainfrom langchain.chains.summarize import load_summarize_chain summary_chain load_summarize_chain(llm, chain_typemap_reduce) summary summary_chain.invoke(chunks) print(summary[output_text])这段代码适合快速验证效果尤其是数据量不大、不追求极致性能的场景。如果你更习惯 LCEL 风格也可以自己实现 map 和 reduce 两个阶段先用一个 Prompt 对每段文档单独摘要再把所有摘要拼起来用第二个 Prompt 做最终总结。MapReduce 的优点是能处理超长文本理论上只要拆成足够小的块都能放进模型上下文。缺点是分段可能导致上下文割裂最终摘要可能丢失全局信息。如果你要摘要的是几十页的合同建议在切分时尽量保持章节完整并让每段摘要保留关键数字和结论。5.7 MCP 工具接入MCP 是一套标准化的外部工具接入协议解决的是“每个工具都写一套自定义接入逻辑”的问题。LangChain 官方提供了langchain-mcp-adapters可以把 MCP 服务里的工具直接加载成 LangChain Tools 使用。下面是一个基于 stdio 方式连接本地 MCP 服务的示意代码from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters( commandpython, args[mcp_server.py], ) async def get_tools(): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools # 拿到 tools 后可以交给 AgentExecutor 使用 # tools await get_tools() # agent_executor AgentExecutor(agentagent, toolstools, verboseTrue)这里要注意MCP SDK 和 LangChain 适配器的版本迭代较快不同版本之间 API 可能不一样。建议以官方示例代码为准。如果出现“工具注册不上”的问题先检查 MCP Server 是否可以单独启动、stdio 通信是否正常再检查load_mcp_tools返回的工具数量是否大于 0。MCP 和 Agent Skill 的区别也需要简单说一句。MCP 侧重“工具如何被标准协议暴露和调用”Agent Skill 更偏向“一组可复用的能力模块包含提示词、工具和执行逻辑的组合”。MCP 解决的是接入层Agent Skill 解决的是抽象层两者可以共存。6. 接口 API 与批量任务6.1 用 FastAPI 封装 LangChain 服务企业项目里LangChain 代码通常不会直接暴露给业务方而是封装成 HTTP 接口。下面是一个最简封装from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Query(BaseModel): question: str app.post(/ask) def ask(query: Query): result agent_executor.invoke({input: query.question}) return {answer: result[output]}启动服务uvicorn app.api:app --host 0.0.0.0 --port 8000启动后用 curl 测试curl -X POST http://127.0.0.1:8000/ask \ -H Content-Type: application/json \ -d {question: 计算 5 7 并乘以 2}这里要注意agent_executor和chain这类对象应该在服务启动时初始化一次不要在每个请求里重新加载文档和向量库。否则接口延迟会非常高。6.2 批量任务处理批量问答和批量文档处理是 RAG 场景里的常见需求。批量任务设计时建议加错误隔离不能让一条失败数据拖垮整个任务。下面是一个简单的并发批量处理示例from concurrent.futures import ThreadPoolExecutor, as_completed def process_question(q: str): try: answer chain.invoke(q) return {question: q, answer: answer, status: ok} except Exception as e: return {question: q, error: str(e), status: failed} questions [ LangChain 是什么, RAG 的核心链路是什么, ReAct 和普通 Agent 有什么区别, ] with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(process_question, q) for q in questions] for future in as_completed(futures): result future.result() print(result)批量任务的关键点有三个控制并发数避免把模型 API 或向量库打挂。把失败记录单独写到日志文件里方便重跑。考虑幂等性同一个任务重复执行不要影响最终结果。如果批量量很大建议引入任务队列比如 Celery 或 Redis Queue。先入库再消费失败重试这样生产环境会更稳定。6.3 批量文档索引流程企业里第一次搭建 RAG 时通常会一次性导入大量历史文档。推荐流程是把文档按目录扫描。根据文件类型选择对应 Loader。切分文档。生成 embedding 并写入向量库。记录每个片段的来源文件名和页码。这个流程可以写成独立脚本避免在 API 服务里反复触发。索引完成后还可以定期增量更新避免每次都全量重建。7. 资源占用与性能观察LangChain 本身不是推理引擎资源占用很大程度取决于底层模型和向量库。但框架层面的性能问题也值得关注。先说云端 API 模式。这个时候 LangChain 进程主要消耗 CPU 和内存因为大量时间在等待外部接口返回。观察指标主要是请求延迟、token 消耗、并发下的队列堆积。如果接口变慢可以先看是不是并发过高导致 API 限流。本地模型模式则要看显存和 GPU 利用率。模型推理由 vLLM、Ollama 或 llama.cpp 负责LangChain 只是客户端。显存占用取决于模型参数量和推理框架配置。如果你在 8G 显存环境下跑 7B 模型不要开启过大的并发窗口否则很容易 OOM。性能优化的几个方向向量检索慢时优先考虑减少向量维度或增大k值合理性而不是盲目升级服务器。embedding 速度慢时可以用批量 embedding 替代逐条调用。Agent 推理慢时检查是否出现了多轮无效工具调用可以通过限制最大迭代次数解决。RAG 回答慢时重点看检索耗时和上下文长度上下文太长会拖慢模型推理。观察这些指标建议接入日志系统把每次请求的模型、token 数、耗时、检索结果都记录下来。企业排障时这批日志比任何参数调优都重要。8. 常见问题与排查方法问题现象可能原因排查方式解决方案模型调用报 401API Key 错误或未设置检查环境变量和模型服务日志重新配置 API Key模型名称不存在使用了不支持或已下线的模型名查看模型服务支持的模型列表替换为有效模型名安装依赖失败Python 版本或依赖冲突查看 pip 错误日志使用虚拟环境升级 pip 后重装向量库报 module not found缺少 faiss 或 chromadb检查 import 语句安装对应向量库依赖RAG 回答不准确切分不合理或检索结果不相关打印 retriever 返回片段调整 chunk_size、chunk_overlap 和 k 值Agent 死循环模型反复调用工具不收敛观察 verbose 输出增加最大迭代限制优化 PromptMCP 工具注册不上MCP Server 未启动或协议不匹配先单独运行 MCP Server检查 stdio 通信和适配器版本批量任务卡住并发过高或单条数据异常查看任务队列和日志降低并发数增加超时和重试API 服务重启后向量库丢失向量库未持久化检查保存路径使用FAISS.save_local()持久化长文本超 token 限制切分后仍超过模型上下文检查 Prompt 长度缩小 chunk_size或改用 MapReduce 摘要FAISS.save_local()和加载对应是常见用法。保存时vectorstore.save_local(data/vectorstore)加载时vectorstore FAISS.load_local(data/vectorstore, embeddings, allow_dangerous_deserializationTrue)注意新版 FAISS 加载本地文件为了安全需要显式开启allow_dangerous_deserialization参数。企业环境中向量库文件属于关键数据要注意访问权限和版本管理。9. 最佳实践与企业级落地建议从示例代码到企业项目差的不是代码而是工程习惯。第一优先使用 LCEL 组织链路。LCEL 把每个节点都抽象成 Runnable方便日志插入、并行执行和单元测试。建议所有项目默认使用 LCEL而不是过早封装自己的链式框架。第二RAG 项目先做检索效果验收。文档切分后单独跑一轮“问题-检索片段”对照确认检索结果是否相关。如果检索结果相关但模型回答不好再去调 Prompt如果检索结果本身就不相关先改切分和 embedding。第三Agent 项目要限制工具调用次数。很多生产事故来自 Agent 在循环里反复调用工具浪费 token 甚至触发副作用。务必设置max_iterations和工具超时时间。第四注重观测性。建议接入 LangSmith 或自定义日志。每次请求记录模型名、Prompt 摘要、工具调用链、token 消耗、耗时和最终输出。没有观测性的大模型应用出了问题很难定位。第五权限控制要前置。RAG 的文档访问权限、Agent 工具的内部系统权限、MCP 接入的外部服务权限都要在框架外层做。不要假设模型能在 Prompt 里隔离权限模型只会从上下文里找信息它不知道什么是“不该看的内容”。第六部署上建议把模型调用、向量库访问和业务 API 分开。向量库单独部署或使用云服务LangChain 应用只做编排这样资源分配更清晰异常影响面也更小。10. 总结与下一步LangChain 最值得尝试的是 RAG 链路和 Agent 编排。建议第一次实验时先跑通一个最小 RAG把文档切分、向量化、检索、生成这四步都看到日志再逐步加 Agent 和工具调用。最容易踩的坑是版本差异和依赖冲突建议项目一开始就锁定核心依赖版本。如果要把这个体系推向企业项目下一步可以重点看 LangGraph 做复杂状态管理以及 MCP 生态做统一工具接入。先把 RAG 跑稳再上 Agent最后接 MCP是风险最小的路径。建议把本文的文末示例保存成一套“可运行样板”作为团队内部快速起步的基础代码。后续再根据实际业务模型、文档类型和部署环境替换其中接口即可。
LangChain 实战指南：从 RAG 到 Agent 的企业级应用与落地实践
LangChain 是大模型应用工程化里绕不开的一个框架。它解决的问题很直接把模型能力、提示词、知识库、外部工具和业务系统串成一条可维护的链路而不是每次从零拼接 Prompt 和模型调用。这次我们直接进入主题从示例代码分析开始把 LangChain 里最常用的几个能力全部过一遍RAG 知识库问答、ReAct Agent 推理、MapReduce 长文本处理、MCP 外部工具接入和工具调用。最后再从示例代码延伸到企业项目落地包括接口封装、批量任务设计和排查思路。这篇文章适合正在学 LangChain 的开发者也适合已经跑通 demo、准备把 RAG 或 Agent 放到生产流程里的同学。文章不会去讲太多概念史核心是把每个能力怎么用、怎么验证、落地时容易踩什么坑讲清楚。1. LangChain 核心能力速览能力项说明项目类型开源大模型应用编排框架由 LangChain 主导维护核心模块langchain-core、langchain、langchain-community、langchain-openai、langgraph 等主要功能Prompt 模板、模型调用、输出解析、RAG、工具调用、Agent、长文本 MapReduce、MCP 工具接入是否支持 API支持可以封装为 FastAPI 等 HTTP 服务供业务系统调用是否支持批量任务支持可基于脚本或队列做批量文档处理、批量问答运行平台跨平台需要 Python 环境模型可以走云端 API 也可以接本地模型启动方式非独立服务而是作为 Python 库集成进应用适合场景智能客服知识库、文档摘要、自动化 Agent、企业内部知识问答、数据分析助手硬件门槛取决于接入的模型用 OpenAI 等 API 不需要 GPU接本地模型则看模型参数量从表格里能看到LangChain 本身不是一个“一键启动的服务”它是一个开发框架。你要把它用到项目里通常需要自己写业务代码再选择底层的模型和向量库。2. 适用场景与使用边界LangChain 最适合的场景是那些需要把大模型和业务数据、外部工具组合起来完成的复杂任务。典型的例子包括企业知识库问答把内部文档切成片段向量化后检索让模型基于检索结果回答。长文档摘要合同、论文、财报这类超长文本直接塞给模型会超出上下文用 MapReduce 把长文档拆成多段分别摘要再合并。自动化 Agent模型根据用户问题决定调用哪些工具、按什么顺序调用最终完成任务。工具接入通过 Tool Calling 或 MCP 统一接入内部 API、数据库、第三方服务。这个框架不太适合的场景是只做一次“模型对话”的小工具。如果只是单轮调用 ChatGPT 接口直接用 OpenAI SDK 会更快不需要引入 LangChain。LangChain 的价值在链式编排、多步骤推理、长期记忆和复杂状态管理。使用边界方面要特别关注安全合规。RAG 如果接企业内部数据要注意权限隔离和敏感数据脱敏不能把全量文档一股脑发给外部模型 API。Agent 工具调用如果接内部系统必须做权限校验和操作审计。MCP 接入第三方服务时要防止越权访问和敏感信息泄露。任何涉及用户数据的内容都要先确认授权和合规要求。3. 环境准备与前置条件开始之前需要准备 Python 环境和模型访问凭证。这里的示例代码主要面向云端模型 API如果你要接本地模型可以把ChatOpenAI替换成兼容 OpenAI 协议的本地推理服务地址。推荐环境Python 3.9 及以上版本。建议使用虚拟环境避免依赖冲突。一个可用的模型 API Key例如 OpenAI 兼容接口。安装常用依赖langchain、langchain-community、langchain-openai、langchain-text-splitters、faiss-cpu、pypdf。如果要用 MCP需要额外安装 mcp 和 langchain-mcp-adapters。环境变量示例export OPENAI_API_KEYsk-xxxxxxxx export OPENAI_BASE_URLhttps://api.openai.com/v1如果你接的是本地模型比如基于 vLLM、Ollama 或 llama.cpp 启动的 OpenAI 兼容服务可以把OPENAI_BASE_URL改成http://127.0.0.1:8000/v1。这种部署方式尤其适合内网场景数据不需要离开本地环境。4. 安装部署与项目初始化安装命令建议一次性装齐常用依赖pip install langchain langchain-community langchain-openai langchain-text-splitters faiss-cpu pypdf如果你要写 API 服务再加安装 FastAPI 和 Uvicornpip install fastapi uvicorn创建项目目录时建议把模型配置、文档素材、输出结果分层管理project/ ├── config/ │ └── settings.py ├── docs/ │ └── knowledge/ ├── app/ │ ├── rag.py │ ├── agent.py │ └── api.py ├── data/ │ ├── vectorstore/ │ └── output/ └── requirements.txt初始化完成后先用一个最简单的模型调用验证环境是否可用。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(用一句话解释 LangChain) print(response.content)如果这一步能输出文字说明模型接入没有问题。如果报错优先检查 API Key、模型名称和网络连通性。5. 核心功能测试从示例代码到实战5.1 Prompt 模板与输出解析实际项目中不会直接裸调模型通常需要把 Prompt 模板和输出解析器组合起来。下面的例子把“解释概念”包装成一个可复用的链from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_messages([ (system, 你是一名资深技术顾问回答要准确、简洁。), (human, 请用中文解释什么是{concept}控制在200字以内。) ]) chain prompt | llm | StrOutputParser() result chain.invoke({concept: RAG}) print(result)这段代码里prompt | llm | StrOutputParser()就是 LangChain 的 LCEL 表达方式。管道符左边是输入右边是处理后输出逻辑清楚也方便后面替换或插入新节点。输出解析器的价值在于把模型返回的字符串转成结构化对象。实际业务中你可以用PydanticOutputParser直接拿到 JSON 格式的字段避免手写正则解析。5.2 工具调用工具调用是 Agent 的基础。模型本身不会执行计算但它可以在需要时返回一个“调用哪个工具、传入什么参数”的结构。from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个整数之和。 return a b tools [add] llm_with_tools llm.bind_tools(tools) response llm_with_tools.invoke(请计算 3 和 5 的和并使用工具) print(response.tool_calls)运行时response.tool_calls会输出类似[{name: add, args: {a: 3, b: 5}, id: ...}]的结构。这里的重点是LangChain 并不强制模型必须执行工具你依然需要自己写“拿到工具调用结果之后怎么继续”的逻辑。这也是 Agent 要解决的问题。5.3 ReAct Agent 推理ReAct 是一种 Agent 推理模式它的核心是让模型在“思考、行动、观察结果、再思考”之间循环直到得出结论。LangChain 里可以通过create_react_agent快速构建from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate react_prompt PromptTemplate.from_template( 你可以使用以下工具 {tools} 请严格按照下面的格式回答 Question: 用户输入的问题 Thought: 思考需要做什么 Action: 选择一个工具必须是 [{tool_names}] 中的一个 Action Input: 输入给工具的参数 Observation: 工具返回的结果 ... 重复若干次 Final Answer: 最终答案 Question: {input} Thought: ) agent create_react_agent(llm, tools, react_prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: 计算 5 7并把结果乘以 2}) print(result[output])这个示例里有两个工具才有意义。实际操作中Agent 会根据问题自行决定是否调用工具、调用哪个工具、调用几次。verboseTrue会让控制台打印完整的思考过程非常适合调试。新版本 LangChain 里create_react_agent是官方推荐的 Agent 构建方式。如果你看到很多老教程还在用initialize_agent建议新项目优先使用新 API必要时再按官方迁移文档调整。5.4 工具调用与 Agent 的边界很多人会把 Agent 和工具调用混淆。简单区分工具调用是模型输出“想用工具”的结构Agent 是拿到这个结构后真正去执行工具、观察结果、继续推理的完整循环。在项目里如果把这两层分开设计会更容易排查问题。先用llm.bind_tools(tools)验证模型是否能正确输出工具调用参数再引入 AgentExecutor 跑完整循环。这样可以快速定位是模型理解问题还是代码执行问题。5.5 RAG 知识库问答RAG 是当前企业项目里价值最高的 LangChain 应用。下面是一个基于 FAISS 和本地文档的完整示例。首先准备一个文本文件rag_example.txt内容随便写一段关于 LangChain 的介绍。然后执行from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough loader TextLoader(rag_example.txt) documents loader.load() splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) chunks splitter.split_documents(documents) embeddings OpenAIEmbeddings() vectorstore FAISS.from_documents(chunks, embeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) prompt ChatPromptTemplate.from_messages([ (system, 你是知识库问答助手请只根据上下文回答。如果上下文中没有答案请明确回答不知道。), (human, 上下文\n{context}\n\n问题{question}) ]) def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) answer chain.invoke(根据文档LangChain 的核心价值是什么) print(answer)这个链路里最值得关注的是文档切分。chunk_size500表示每块大概 500 个字符chunk_overlap50表示相邻块之间保留 50 个字符重叠。重叠能减少切断语义的问题但重叠太大会浪费 embedding 空间并影响检索精度。真正的企业项目里文档格式不会只有纯文本往往是 PDF、Word、Markdown 混在一起。建议根据文件类型选择对应的 Loader比如PyPDFLoader、Docx2txtLoader再统一走同一条切分和向量化流程。从实践角度看RAG 的效果不止取决于 LangChain 代码更取决于文档质量和切分策略。如果你检索出来的片段本身不相关模型再强也回答不好。所以第一步应该先检查检索结果而不是反复调 Prompt。5.6 MapReduce 长文本处理长文档摘要场景下MapReduce 是把文档切成多段、逐段摘要、再合并总结的经典思路。LangChain 老版本提供了现成的load_summarize_chainfrom langchain.chains.summarize import load_summarize_chain summary_chain load_summarize_chain(llm, chain_typemap_reduce) summary summary_chain.invoke(chunks) print(summary[output_text])这段代码适合快速验证效果尤其是数据量不大、不追求极致性能的场景。如果你更习惯 LCEL 风格也可以自己实现 map 和 reduce 两个阶段先用一个 Prompt 对每段文档单独摘要再把所有摘要拼起来用第二个 Prompt 做最终总结。MapReduce 的优点是能处理超长文本理论上只要拆成足够小的块都能放进模型上下文。缺点是分段可能导致上下文割裂最终摘要可能丢失全局信息。如果你要摘要的是几十页的合同建议在切分时尽量保持章节完整并让每段摘要保留关键数字和结论。5.7 MCP 工具接入MCP 是一套标准化的外部工具接入协议解决的是“每个工具都写一套自定义接入逻辑”的问题。LangChain 官方提供了langchain-mcp-adapters可以把 MCP 服务里的工具直接加载成 LangChain Tools 使用。下面是一个基于 stdio 方式连接本地 MCP 服务的示意代码from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters( commandpython, args[mcp_server.py], ) async def get_tools(): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools # 拿到 tools 后可以交给 AgentExecutor 使用 # tools await get_tools() # agent_executor AgentExecutor(agentagent, toolstools, verboseTrue)这里要注意MCP SDK 和 LangChain 适配器的版本迭代较快不同版本之间 API 可能不一样。建议以官方示例代码为准。如果出现“工具注册不上”的问题先检查 MCP Server 是否可以单独启动、stdio 通信是否正常再检查load_mcp_tools返回的工具数量是否大于 0。MCP 和 Agent Skill 的区别也需要简单说一句。MCP 侧重“工具如何被标准协议暴露和调用”Agent Skill 更偏向“一组可复用的能力模块包含提示词、工具和执行逻辑的组合”。MCP 解决的是接入层Agent Skill 解决的是抽象层两者可以共存。6. 接口 API 与批量任务6.1 用 FastAPI 封装 LangChain 服务企业项目里LangChain 代码通常不会直接暴露给业务方而是封装成 HTTP 接口。下面是一个最简封装from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Query(BaseModel): question: str app.post(/ask) def ask(query: Query): result agent_executor.invoke({input: query.question}) return {answer: result[output]}启动服务uvicorn app.api:app --host 0.0.0.0 --port 8000启动后用 curl 测试curl -X POST http://127.0.0.1:8000/ask \ -H Content-Type: application/json \ -d {question: 计算 5 7 并乘以 2}这里要注意agent_executor和chain这类对象应该在服务启动时初始化一次不要在每个请求里重新加载文档和向量库。否则接口延迟会非常高。6.2 批量任务处理批量问答和批量文档处理是 RAG 场景里的常见需求。批量任务设计时建议加错误隔离不能让一条失败数据拖垮整个任务。下面是一个简单的并发批量处理示例from concurrent.futures import ThreadPoolExecutor, as_completed def process_question(q: str): try: answer chain.invoke(q) return {question: q, answer: answer, status: ok} except Exception as e: return {question: q, error: str(e), status: failed} questions [ LangChain 是什么, RAG 的核心链路是什么, ReAct 和普通 Agent 有什么区别, ] with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(process_question, q) for q in questions] for future in as_completed(futures): result future.result() print(result)批量任务的关键点有三个控制并发数避免把模型 API 或向量库打挂。把失败记录单独写到日志文件里方便重跑。考虑幂等性同一个任务重复执行不要影响最终结果。如果批量量很大建议引入任务队列比如 Celery 或 Redis Queue。先入库再消费失败重试这样生产环境会更稳定。6.3 批量文档索引流程企业里第一次搭建 RAG 时通常会一次性导入大量历史文档。推荐流程是把文档按目录扫描。根据文件类型选择对应 Loader。切分文档。生成 embedding 并写入向量库。记录每个片段的来源文件名和页码。这个流程可以写成独立脚本避免在 API 服务里反复触发。索引完成后还可以定期增量更新避免每次都全量重建。7. 资源占用与性能观察LangChain 本身不是推理引擎资源占用很大程度取决于底层模型和向量库。但框架层面的性能问题也值得关注。先说云端 API 模式。这个时候 LangChain 进程主要消耗 CPU 和内存因为大量时间在等待外部接口返回。观察指标主要是请求延迟、token 消耗、并发下的队列堆积。如果接口变慢可以先看是不是并发过高导致 API 限流。本地模型模式则要看显存和 GPU 利用率。模型推理由 vLLM、Ollama 或 llama.cpp 负责LangChain 只是客户端。显存占用取决于模型参数量和推理框架配置。如果你在 8G 显存环境下跑 7B 模型不要开启过大的并发窗口否则很容易 OOM。性能优化的几个方向向量检索慢时优先考虑减少向量维度或增大k值合理性而不是盲目升级服务器。embedding 速度慢时可以用批量 embedding 替代逐条调用。Agent 推理慢时检查是否出现了多轮无效工具调用可以通过限制最大迭代次数解决。RAG 回答慢时重点看检索耗时和上下文长度上下文太长会拖慢模型推理。观察这些指标建议接入日志系统把每次请求的模型、token 数、耗时、检索结果都记录下来。企业排障时这批日志比任何参数调优都重要。8. 常见问题与排查方法问题现象可能原因排查方式解决方案模型调用报 401API Key 错误或未设置检查环境变量和模型服务日志重新配置 API Key模型名称不存在使用了不支持或已下线的模型名查看模型服务支持的模型列表替换为有效模型名安装依赖失败Python 版本或依赖冲突查看 pip 错误日志使用虚拟环境升级 pip 后重装向量库报 module not found缺少 faiss 或 chromadb检查 import 语句安装对应向量库依赖RAG 回答不准确切分不合理或检索结果不相关打印 retriever 返回片段调整 chunk_size、chunk_overlap 和 k 值Agent 死循环模型反复调用工具不收敛观察 verbose 输出增加最大迭代限制优化 PromptMCP 工具注册不上MCP Server 未启动或协议不匹配先单独运行 MCP Server检查 stdio 通信和适配器版本批量任务卡住并发过高或单条数据异常查看任务队列和日志降低并发数增加超时和重试API 服务重启后向量库丢失向量库未持久化检查保存路径使用FAISS.save_local()持久化长文本超 token 限制切分后仍超过模型上下文检查 Prompt 长度缩小 chunk_size或改用 MapReduce 摘要FAISS.save_local()和加载对应是常见用法。保存时vectorstore.save_local(data/vectorstore)加载时vectorstore FAISS.load_local(data/vectorstore, embeddings, allow_dangerous_deserializationTrue)注意新版 FAISS 加载本地文件为了安全需要显式开启allow_dangerous_deserialization参数。企业环境中向量库文件属于关键数据要注意访问权限和版本管理。9. 最佳实践与企业级落地建议从示例代码到企业项目差的不是代码而是工程习惯。第一优先使用 LCEL 组织链路。LCEL 把每个节点都抽象成 Runnable方便日志插入、并行执行和单元测试。建议所有项目默认使用 LCEL而不是过早封装自己的链式框架。第二RAG 项目先做检索效果验收。文档切分后单独跑一轮“问题-检索片段”对照确认检索结果是否相关。如果检索结果相关但模型回答不好再去调 Prompt如果检索结果本身就不相关先改切分和 embedding。第三Agent 项目要限制工具调用次数。很多生产事故来自 Agent 在循环里反复调用工具浪费 token 甚至触发副作用。务必设置max_iterations和工具超时时间。第四注重观测性。建议接入 LangSmith 或自定义日志。每次请求记录模型名、Prompt 摘要、工具调用链、token 消耗、耗时和最终输出。没有观测性的大模型应用出了问题很难定位。第五权限控制要前置。RAG 的文档访问权限、Agent 工具的内部系统权限、MCP 接入的外部服务权限都要在框架外层做。不要假设模型能在 Prompt 里隔离权限模型只会从上下文里找信息它不知道什么是“不该看的内容”。第六部署上建议把模型调用、向量库访问和业务 API 分开。向量库单独部署或使用云服务LangChain 应用只做编排这样资源分配更清晰异常影响面也更小。10. 总结与下一步LangChain 最值得尝试的是 RAG 链路和 Agent 编排。建议第一次实验时先跑通一个最小 RAG把文档切分、向量化、检索、生成这四步都看到日志再逐步加 Agent 和工具调用。最容易踩的坑是版本差异和依赖冲突建议项目一开始就锁定核心依赖版本。如果要把这个体系推向企业项目下一步可以重点看 LangGraph 做复杂状态管理以及 MCP 生态做统一工具接入。先把 RAG 跑稳再上 Agent最后接 MCP是风险最小的路径。建议把本文的文末示例保存成一套“可运行样板”作为团队内部快速起步的基础代码。后续再根据实际业务模型、文档类型和部署环境替换其中接口即可。