LangChain 1.3 这类框架市面上教程很多但多数都在讲概念什么是 Chain、什么是 Agent、什么是记忆。真正能照着做、能把一个问答系统跑起来、能从单条测试走到批量任务的实战教程反而很少。这篇文章我按自己实际跑过的顺序来写先讲清楚它解决什么问题再给环境、代码、参数和排查方法。适合三类人刚接触大模型应用开发、想把 LangChain 从 Demo 升级成稳定服务、以及被各种概念绕晕但需要尽快写出可运行代码的人。我的核心建议是不要一上来就研究所有组件先把最小链路跑通再往里面加记忆、工具和检索。1. 先搞清楚LangChain 1.3 到底帮你解决什么问题1.1 不要把 LangChain 当成大模型本身很多人第一次用 LangChain 会有一个误解以为装上它就有了智能问答能力。其实 LangChain 本身不提供模型它只负责把模型能力、工具、数据、记忆、业务流程串起来。你可以把它理解成一条生产线模型是机器工具是辅料提示词是操作说明书LangChain 提供的是传送带和控制逻辑。在实际项目中这个定位决定了你该怎么用它。如果你想解决“让模型知道最新资料”LangChain 负责把资料切块、检索、拼进提示词如果你想让模型调用外部 APILangChain 负责定义工具、传给模型、执行工具并返回结果如果你要做多轮对话LangChain 负责保存历史、拼接上下文。模型的能力边界仍然在LangChain 只是让这些能力更容易组合。所以遇到问题不要先怀疑 LangChain先确认模型本身能不能完成这个任务。模型答不对你换了 LangChain 版本也没用。1.2 哪些场景值得用 LangChain 1.3从实际投入产出看下面几类场景最值得用 LangChain多步流程编排。比如先判断问题类型再决定走检索还是走工具调用最后统一生成答案。工具调用。模型需要查天气、查订单、算数学、读数据库时用 LangChain 比自己在代码里维护循环要省事。文档问答。对内部文档、说明手册、项目资料做 RAG这是当前最常见的落地场景。多轮对话带记忆。需要记住用户前面说了什么并且要区分不同会话。快速把 Demo 包装成接口。用 LangChain 组装好链路后外面套一层 FastAPI 就能暴露给其他系统。如果你只是简单调一次模型接口完全不需要 LangChain。直接请求模型 API 更轻量。凡是链路里超过两个步骤或者需要重试、缓存、日志、历史管理LangChain 的价值才会体现出来。1.3 哪些场景暂时不要硬上以下场景我建议先别用 LangChain或者至少不要把大量时间花在框架本身上只是简单翻译、改写、单轮问答直接调模型更简单。需要极低延迟的高并发接口框架本身不是瓶颈但层层封装会增加排查成本。业务流程非常固定用普通代码写 if 分支比 Agent 更可控。模型本身不支持工具调用又强行跑 Agent大概率会输出不稳定的 JSON。另外还要提醒一句LangChain 1.x 版本更新速度很快接口调整比 0.x 时代更频繁。你在网上看到的老教程很可能是旧版本写法。学习时先锁定一个稳定版本把核心链路跑通再考虑升级不要每天追新。2. 环境准备把最小可运行链路先跑通2.1 安装前先确认 Python 版本和依赖策略我习惯先把环境理顺再写代码。LangChain 1.3 基于 Python 3.10 以上环境会比较舒服如果你还在用 3.8建议先升级。用虚拟环境隔离项目不要直接装到全局 Python 里否则后面不同项目依赖冲突会非常难受。安装时按包拆分来装。只装一个langchain并不够还要按调用来源安装对应包。比如要用 OpenAI 兼容接口就装langchain-openai要用本地向量库就装langchain-community和对应的文档加载器依赖。以下是我测试时常用的安装命令python -m venv .venv source .venv/bin/activate # Windows 下是 .venv\Scripts\activate pip install langchain1.3.* pip install langchain-openai pip install langchain-community pip install langchain-text-splitters pip install faiss-cpu pip install fastapi uvicorn注意这里给的是通用示例。你安装时以官方 PyPI 上实际可用的版本为准。如果你用的是某个国内模型服务确认它提供了 OpenAI 兼容接口那么base_url和api_key按服务商文档填就行。还有一点1.3 版本如果报ModuleNotFoundError先别急着搜代码多数情况是某个分包没有安装例如langchain_text_splitters或langchain_core。2.2 模型接入API Key、基础模型、温度参数不管后面做什么先要有一个能调通的模型对象。以 OpenAI 兼容接口为例代码结构如下from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI llm ChatOpenAI( modelyour-model-name, # 按模型服务商实际的模型名填写 api_keyyour-api-key, # 建议从环境变量读取不要硬编码 base_urlhttps://your-provider-endpoint/v1, # OpenAI兼容接口地址 temperature0.3, )这里有几个参数值得说清楚model不同服务商支持不同模型不要照抄我的名字。temperature控制随机性0 到 1 之间。做问答、抽取、分类我一般设 0 到 0.3创意文案可以调到 0.7 以上。base_url只填到/v1级别不要在后面多加路径否则容易 404。api_key放在环境变量里更安全ChatOpenAI(api_keyos.getenv(API_KEY))是更推荐的方式。接入后立刻做一次最小调用response llm.invoke([HumanMessage(content用一句话介绍 LangChain)]) print(response.content)如果这一步能输出结果说明环境没问题。这一步失败了后面的链、Agent、RAG 都没必要继续。排查时先看三件事base_url是否填对api_key是否有效网络是否能访问到模型服务商接口。别一上来就怀疑 LangChain。2.3 第一次调用从一条 Prompt 到一个完整输出拿到模型对象后再往前走一步把 Prompt 模板和输出解析接上。这个组合非常常见from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template( 你是{role}请用简洁的语言回答{question} ) chain prompt | llm | StrOutputParser() result chain.invoke({ role: 技术顾问, question: 如何设计一个RAG系统 }) print(result)这里用到了 LCELLangChain Expression Language的管道写法prompt | llm | StrOutputParser()。含义是先把输入填充到模板然后交给模型再把模型返回的复杂对象转成字符串。这种写法在 LangChain 1.3 里非常核心后面所有链路都可以按同样方式组合。我建议第一次跑通这个最小示例后再往里面加组件。如果你连链的输出都拿不到说明前面的模型接入还有问题不要急着加记忆和工具。3. 代码实战从链路到工具调用3.1 用 LCEL 组装一条基础链基础链的作用是固定“输入格式、提示词、模型参数、输出格式”这四件事。很多项目里提示词会反复调整但链的骨架可以不变。举个例子我希望模型每次回答都先分类再给结论。这时可以不用复杂逻辑直接在提示词里约定输出结构from langchain_core.prompts import ChatPromptTemplate analysis_prompt ChatPromptTemplate.from_template( 请对以下问题做两件事 1. 判断它属于“知识问答”“操作建议”还是“闲聊”。 2. 给出不超过 80 字的回答。 问题{question} 输出格式 分类xxx 回答xxx ) analysis_chain analysis_prompt | llm | StrOutputParser() print(analysis_chain.invoke({question: Python列表和元组有什么区别}))这时候你可能发现纯 Prompt 引导也能达到效果。确实如此。先不要急着上 Agent能用 Prompt 解决的就用 Prompt能少一层封装就少一层。链的灵活之处在于后续可以把“分类”结果拿出来做条件判断那时候再考虑拆成不同的链。3.2 给链加上记忆处理多轮对话模型接口本身没有记忆LangChain 的RunnableWithMessageHistory可以帮我们把每次对话历史保存起来。1.3 版本里我一般这样写from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory store {} def get_session_history(session_id: str): if session_id not in store: store[session_id] InMemoryChatMessageHistory() return store[session_id] prompt ChatPromptTemplate.from_template( 以下是对话历史\n{history}\n\n用户问题{question} ) history_chain prompt | llm | StrOutputParser() chain_with_history RunnableWithMessageHistory( history_chain, get_session_history, input_messages_keyquestion, history_messages_keyhistory, )调用时通过config传session_idresp chain_with_history.invoke( {question: 我叫张三记住这个名字}, config{configurable: {session_id: user_001}} ) print(resp) resp2 chain_with_history.invoke( {question: 我刚才说我的名字是什么}, config{configurable: {session_id: user_001}} ) print(resp2)注意不同版本的参数名可能不同。RunnableWithMessageHistory要求你明确指定哪个字段是question哪个字段是history否则它不知道如何替换历史消息。我实际测试时发现这里最容易写错history_messages_key填错模型看不到历史或者直接报错。运行时如果输出正常但不带记忆优先检查这个字段。内存存储只适合单机演示。生产环境建议换成 Redis 或数据库存储按session_id读取历史。否则服务重启所有对话记录就丢了。3.3 工具调用让模型能查天气、算数学、查数据库工具调用是 LangChain 最有价值的能力之一。思路很简单先定义工具再把工具列表传给模型模型判断需要工具时会返回一个调用请求LangChain 帮你执行工具并把结果回传给模型。先看一个计算器工具from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个整数相加并返回结果。 return a b再定义一个模拟天气工具tool def get_weather(city: str) - str: 查询指定城市的当前天气。 weather_data { 北京: 晴天25度, 上海: 多云28度, } return weather_data.get(city, 暂无该城市天气数据)这里的函数注释很重要它是给模型看的。模型通过函数名、参数描述和注释来判断什么时候调用工具。如果你把注释写得很模糊模型可能不会调用或者乱传参数。绑定工具需要你的模型服务支持 function calling / tool calling。如果不支持后面的 Agent 流程就做不了tools [add, get_weather] llm_with_tools llm.bind_tools(tools) resp llm_with_tools.invoke([HumanMessage(content北京今天多少度)]) print(resp) # 通常是一个包含 tool_calls 的响应如果模型决定调用工具你会看到响应里有tool_calls。你还需要手动执行工具并把结果返回给模型这也是 Agent 框架替我们做的事。所以不理解原理时跑一次再观察响应结构比直接套 Agent 更好。3.4 用 Agent 把多个工具交给模型调度手动处理工具调用很繁琐尤其工具一多要写很多循环逻辑。LangChain 的 Agent 把“决定调用哪个工具、执行、回传、再生成”这个循环封装起来了。示意代码from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate agent_prompt ChatPromptTemplate.from_messages([ (system, 你是一个智能助手可以使用工具。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(llm, tools, agent_prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: 北京今天多少度然后帮我算 23 45}) print(result)这段代码里不要手动传agent_scratchpadAgent 会自己填充。verboseTrue可以让你看到整个执行过程非常有用。我不建议一上手就写多工具复杂 Agent。先放两个工具跑通“模型识别意图、调用工具、组织回答”的完整流程。如果模型频繁调用错误工具优先检查工具描述和参数约束不一定改代码。比如get_weather的参数名最好就是city不要传一个 JSON 对象省得模型解析错。另外Agent 不是万能的。它会增加请求次数也会增加失败概率。生产环境如果业务流程固定直接用普通链更稳。4. 再进一步搭一个本地 RAG 问答系统4.1 文档加载与拆分的正确姿势RAG 是现在落地最多的 LangChain 场景先加载你自己的文档切成小块向量化存入向量库用户提问时先检索相关片段再把片段拼进提示词让模型回答。加载方式按文件类型来。纯文本最简单from langchain_community.document_loaders import TextLoader loader TextLoader(docs/raw.txt, encodingutf-8) docs loader.load()如果是 PDF、Word、Markdown需要装对应加载器。注意编码问题中文文档经常因为gbk或utf-8不一致读不出来。报错时先看是UnicodeDecodeError还是文件路径不存在不要急着换库。拆分是 RAG 里容易被忽略的一环。我常用的参数from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , ], ) chunks splitter.split_documents(docs) print(len(chunks)) print(chunks[0].page_content)chunk_size是每个块的最大字符数chunk_overlap是相邻块的重叠字符数。重叠是为了避免句子被切断后丢失上下文。不要设太小比如 100 字检索时信息量不够也不要设太大比如 2000 字超出模型上下文窗口后要么截断要么浪费 token。我一般从 500 到 800 开始调。4.2 向量化与检索向量化需要选择 embedding 模型。你可以用 API 服务也可以用本地模型。API 方式省资源但每次调用有成本。本地方式更隐私但需要一定内存。from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS embeddings OpenAIEmbeddings( modelyour-embedding-model, api_keyyour-api-key, base_urlhttps://your-provider-endpoint/v1, ) vectorstore FAISS.from_documents(chunks, embeddings) retriever vectorstore.as_retriever(search_kwargs{k: 4})k表示每次检索返回几块文档。太小可能漏内容太大可能引入噪声。我通常先从 3 到 5 开始看回答质量再调整。向量库这里选了 FAISS适合单机场景如果想做千万级数据再换 Milvus 或其他数据库。第一次from_documents会有点慢因为要把所有 chunk 向量化。如果文档很多先跑一个小样确认切分和向量化没问题再全量跑。不要一上来就灌几千个文档出了错会很难定位。4.3 问答链组装与效果验证检索出来后要把检索到的多个文档拼成上下文再交给模型from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate rag_prompt ChatPromptTemplate.from_template( 请根据以下资料回答问题。如果资料中没有相关内容请直接回答“资料中未提及”。 资料 {context} 问题{question} ) def format_docs(doc_list): return \n\n.join([d.page_content for d in doc_list]) rag_chain ( {context: retriever | format_docs, question: lambda x: x[question]} | rag_prompt | llm | StrOutputParser() )调用一次result rag_chain.invoke({question: 这份文档里提到了哪些部署步骤}) print(result)如果回答不准确不要先调模型先看retriever检索到了什么。可以在调试时单独打印检索结果for i, doc in enumerate(retriever.invoke(部署步骤)): print(i, doc.page_content[:200])如果检索结果和问题不相关说明切块大小、k值、文档内容质量有问题。如果检索结果是相关的但模型回答不对再考虑改提示词或换更强模型。4.4 从单条问答到批量接口单条问答跑通后批量处理也很重要。常见场景是给一批问题生成答案或者对外提供服务。批量处理时不要只写一个 for 循环就完事要考虑失败重试和输出保存。questions [ 这份文档适合什么人, 安装依赖需要哪些步骤, 数据量大的时候怎么处理, ] results [] for i, q in enumerate(questions): try: ans rag_chain.invoke({question: q}) results.append({id: i, question: q, answer: ans}) print(i, ok) except Exception as e: results.append({id: i, question: q, answer: , error: str(e)}) print(i, failed, e)之后再统一输出到文件。如果你要暴露成 API用 FastAPI 包一层from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QARequest(BaseModel): question: str session_id: str default app.post(/qa) def qa_api(req: QARequest): answer rag_chain.invoke({question: req.question}) return {question: req.question, answer: answer}这里需要注意rag_chain在服务里是全局变量初始化一次即可不要每次请求都重新加载文档和向量库。对于更高并发可以再加异步接口或缓存。5. 常见报错、性能边界和生产化建议5.1 按现象排查先看输入、再看日志、最后看依赖我踩过最多的坑并不是 LangChain 本身有多复杂而是问题定位顺序不对。你按下面顺序排查基本能解决八成问题第一看现象。是报错、卡死、无输出还是输出质量差不同现象对应不同方向。报错看堆栈卡死看网络和资源无输出看提示词和输出解析质量差看输入内容。第二看输入。RAG 里最常见的问题是文档没加载成功、切分后空块、检索结果为空。先打印len(chunks)和retriever.invoke的结果确认输入链路是完整的。第三看配置。api_key有没有填对base_url有没有多写路径模型名是否有效温度参数是否设置过高。这些参数一旦有问题模型会返回 401、404 或者奇怪内容。第四看依赖。langchain系列包之间版本不匹配会报ImportError或TypeError。建议同一个虚拟环境里安装不要混用 pip 和 conda 源。遇到Pydantic报错多半是某个包版本被安装成了冲突版本。第五再看框架限制。比如模型不支持工具调用但硬要跑 Agent文档超出上下文长度向量维度不匹配。这些属于能力边界不是 bug。5.2 资源占用和并发别一上来就开最大并发如果你用 API 模型本地主要消耗内存和少量 CPU显存几乎不占。如果你用本地模型比如通过 Ollama 或 vLLM 起服务显存占用就很重要。低显存机器不代表不能跑但要把模型换成小尺寸并且降低并发数。批量任务也不要一上来就开 20 个并发。很多模型服务有 QPS 限制并发太高会触发限流反而拖慢整体速度。我一般按这个顺序测先单线程跑 10 条确认稳定性。再开 3 到 5 个并发观察响应时间和失败率。最后根据模型服务商限制逐步提升到合理并发。另外要注意输出目录。批量任务如果都写同一个文件容易互相覆盖。给每次运行生成带时间戳的目录把结果、日志单独放。这个习惯在调试时特别有用。5.3 生产化落地日志、缓存、失败重试与版本锁定Demo 跑通不等于服务稳定。真要放到生产环境至少要补齐四件事。日志是第一位。print只能临时看生产环境必须记录请求 ID、输入摘要、耗时、错误信息。LangChain 里开启 verbose 只能看流程不能代替业务日志。缓存是第二位。相同问题可以缓存结果尤其是检索结果能省很多 token。但要注意缓存键不能只放问题要把版本号和模型参数也带上否则改提示词后缓存会串。失败重试要区分错误类型。网络超时可以重试模型报context length就不要再重试先压缩输入。工具调用失败要看是参数错误还是服务不可用不能盲目重试。这里做个简单的指数退避比立即刷请求更稳妥。最后是依赖版本锁定。LangChain 1.x 迭代很快今天能跑的代码两周后升级小版本可能就报错了。生产环境建议在requirements.txt里锁定具体版本升级前先在测试环境跑一遍完整流程。还有一点敏感信息不要写进日志和缓存。如果做的是企业问答文档里可能有内部数据检索结果会进入模型请求这一点要在设计阶段就和业务方确认数据边界。网络上还常见把 API Key 传到公网仓库的问题这种事风险很大建议直接放进环境变量或密钥管理服务。现在低代码平台也逐渐接入大模型编排能力比如 Mendix 这类产品也有 AI 相关组件。但对于需要精细控制、排查问题、对接私有数据的场景LangChain 这类代码方案仍然更灵活。低代码适合快速搭界面和简单流程复杂链路的调试还是代码更直观。最后留一个个人经验先跑通最小链路再逐步加组件。不管是 LangChain 1.3 还是其他版本真正影响落地的往往不是某个高级概念而是环境、输入格式、参数边界和失败处理。把这几件事理顺了剩下的就是耐心调提示词。
LangChain 1.3 这类框架市面上教程很多但多数都在讲概念什么是 Chain、什么是 Agent、什么是记忆。真正能照着做、能把一个问答系统跑起来、能从单条测试走到批量任务的实战教程反而很少。这篇文章我按自己实际跑过的顺序来写先讲清楚它解决什么问题再给环境、代码、参数和排查方法。适合三类人刚接触大模型应用开发、想把 LangChain 从 Demo 升级成稳定服务、以及被各种概念绕晕但需要尽快写出可运行代码的人。我的核心建议是不要一上来就研究所有组件先把最小链路跑通再往里面加记忆、工具和检索。1. 先搞清楚LangChain 1.3 到底帮你解决什么问题1.1 不要把 LangChain 当成大模型本身很多人第一次用 LangChain 会有一个误解以为装上它就有了智能问答能力。其实 LangChain 本身不提供模型它只负责把模型能力、工具、数据、记忆、业务流程串起来。你可以把它理解成一条生产线模型是机器工具是辅料提示词是操作说明书LangChain 提供的是传送带和控制逻辑。在实际项目中这个定位决定了你该怎么用它。如果你想解决“让模型知道最新资料”LangChain 负责把资料切块、检索、拼进提示词如果你想让模型调用外部 APILangChain 负责定义工具、传给模型、执行工具并返回结果如果你要做多轮对话LangChain 负责保存历史、拼接上下文。模型的能力边界仍然在LangChain 只是让这些能力更容易组合。所以遇到问题不要先怀疑 LangChain先确认模型本身能不能完成这个任务。模型答不对你换了 LangChain 版本也没用。1.2 哪些场景值得用 LangChain 1.3从实际投入产出看下面几类场景最值得用 LangChain多步流程编排。比如先判断问题类型再决定走检索还是走工具调用最后统一生成答案。工具调用。模型需要查天气、查订单、算数学、读数据库时用 LangChain 比自己在代码里维护循环要省事。文档问答。对内部文档、说明手册、项目资料做 RAG这是当前最常见的落地场景。多轮对话带记忆。需要记住用户前面说了什么并且要区分不同会话。快速把 Demo 包装成接口。用 LangChain 组装好链路后外面套一层 FastAPI 就能暴露给其他系统。如果你只是简单调一次模型接口完全不需要 LangChain。直接请求模型 API 更轻量。凡是链路里超过两个步骤或者需要重试、缓存、日志、历史管理LangChain 的价值才会体现出来。1.3 哪些场景暂时不要硬上以下场景我建议先别用 LangChain或者至少不要把大量时间花在框架本身上只是简单翻译、改写、单轮问答直接调模型更简单。需要极低延迟的高并发接口框架本身不是瓶颈但层层封装会增加排查成本。业务流程非常固定用普通代码写 if 分支比 Agent 更可控。模型本身不支持工具调用又强行跑 Agent大概率会输出不稳定的 JSON。另外还要提醒一句LangChain 1.x 版本更新速度很快接口调整比 0.x 时代更频繁。你在网上看到的老教程很可能是旧版本写法。学习时先锁定一个稳定版本把核心链路跑通再考虑升级不要每天追新。2. 环境准备把最小可运行链路先跑通2.1 安装前先确认 Python 版本和依赖策略我习惯先把环境理顺再写代码。LangChain 1.3 基于 Python 3.10 以上环境会比较舒服如果你还在用 3.8建议先升级。用虚拟环境隔离项目不要直接装到全局 Python 里否则后面不同项目依赖冲突会非常难受。安装时按包拆分来装。只装一个langchain并不够还要按调用来源安装对应包。比如要用 OpenAI 兼容接口就装langchain-openai要用本地向量库就装langchain-community和对应的文档加载器依赖。以下是我测试时常用的安装命令python -m venv .venv source .venv/bin/activate # Windows 下是 .venv\Scripts\activate pip install langchain1.3.* pip install langchain-openai pip install langchain-community pip install langchain-text-splitters pip install faiss-cpu pip install fastapi uvicorn注意这里给的是通用示例。你安装时以官方 PyPI 上实际可用的版本为准。如果你用的是某个国内模型服务确认它提供了 OpenAI 兼容接口那么base_url和api_key按服务商文档填就行。还有一点1.3 版本如果报ModuleNotFoundError先别急着搜代码多数情况是某个分包没有安装例如langchain_text_splitters或langchain_core。2.2 模型接入API Key、基础模型、温度参数不管后面做什么先要有一个能调通的模型对象。以 OpenAI 兼容接口为例代码结构如下from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI llm ChatOpenAI( modelyour-model-name, # 按模型服务商实际的模型名填写 api_keyyour-api-key, # 建议从环境变量读取不要硬编码 base_urlhttps://your-provider-endpoint/v1, # OpenAI兼容接口地址 temperature0.3, )这里有几个参数值得说清楚model不同服务商支持不同模型不要照抄我的名字。temperature控制随机性0 到 1 之间。做问答、抽取、分类我一般设 0 到 0.3创意文案可以调到 0.7 以上。base_url只填到/v1级别不要在后面多加路径否则容易 404。api_key放在环境变量里更安全ChatOpenAI(api_keyos.getenv(API_KEY))是更推荐的方式。接入后立刻做一次最小调用response llm.invoke([HumanMessage(content用一句话介绍 LangChain)]) print(response.content)如果这一步能输出结果说明环境没问题。这一步失败了后面的链、Agent、RAG 都没必要继续。排查时先看三件事base_url是否填对api_key是否有效网络是否能访问到模型服务商接口。别一上来就怀疑 LangChain。2.3 第一次调用从一条 Prompt 到一个完整输出拿到模型对象后再往前走一步把 Prompt 模板和输出解析接上。这个组合非常常见from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template( 你是{role}请用简洁的语言回答{question} ) chain prompt | llm | StrOutputParser() result chain.invoke({ role: 技术顾问, question: 如何设计一个RAG系统 }) print(result)这里用到了 LCELLangChain Expression Language的管道写法prompt | llm | StrOutputParser()。含义是先把输入填充到模板然后交给模型再把模型返回的复杂对象转成字符串。这种写法在 LangChain 1.3 里非常核心后面所有链路都可以按同样方式组合。我建议第一次跑通这个最小示例后再往里面加组件。如果你连链的输出都拿不到说明前面的模型接入还有问题不要急着加记忆和工具。3. 代码实战从链路到工具调用3.1 用 LCEL 组装一条基础链基础链的作用是固定“输入格式、提示词、模型参数、输出格式”这四件事。很多项目里提示词会反复调整但链的骨架可以不变。举个例子我希望模型每次回答都先分类再给结论。这时可以不用复杂逻辑直接在提示词里约定输出结构from langchain_core.prompts import ChatPromptTemplate analysis_prompt ChatPromptTemplate.from_template( 请对以下问题做两件事 1. 判断它属于“知识问答”“操作建议”还是“闲聊”。 2. 给出不超过 80 字的回答。 问题{question} 输出格式 分类xxx 回答xxx ) analysis_chain analysis_prompt | llm | StrOutputParser() print(analysis_chain.invoke({question: Python列表和元组有什么区别}))这时候你可能发现纯 Prompt 引导也能达到效果。确实如此。先不要急着上 Agent能用 Prompt 解决的就用 Prompt能少一层封装就少一层。链的灵活之处在于后续可以把“分类”结果拿出来做条件判断那时候再考虑拆成不同的链。3.2 给链加上记忆处理多轮对话模型接口本身没有记忆LangChain 的RunnableWithMessageHistory可以帮我们把每次对话历史保存起来。1.3 版本里我一般这样写from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory store {} def get_session_history(session_id: str): if session_id not in store: store[session_id] InMemoryChatMessageHistory() return store[session_id] prompt ChatPromptTemplate.from_template( 以下是对话历史\n{history}\n\n用户问题{question} ) history_chain prompt | llm | StrOutputParser() chain_with_history RunnableWithMessageHistory( history_chain, get_session_history, input_messages_keyquestion, history_messages_keyhistory, )调用时通过config传session_idresp chain_with_history.invoke( {question: 我叫张三记住这个名字}, config{configurable: {session_id: user_001}} ) print(resp) resp2 chain_with_history.invoke( {question: 我刚才说我的名字是什么}, config{configurable: {session_id: user_001}} ) print(resp2)注意不同版本的参数名可能不同。RunnableWithMessageHistory要求你明确指定哪个字段是question哪个字段是history否则它不知道如何替换历史消息。我实际测试时发现这里最容易写错history_messages_key填错模型看不到历史或者直接报错。运行时如果输出正常但不带记忆优先检查这个字段。内存存储只适合单机演示。生产环境建议换成 Redis 或数据库存储按session_id读取历史。否则服务重启所有对话记录就丢了。3.3 工具调用让模型能查天气、算数学、查数据库工具调用是 LangChain 最有价值的能力之一。思路很简单先定义工具再把工具列表传给模型模型判断需要工具时会返回一个调用请求LangChain 帮你执行工具并把结果回传给模型。先看一个计算器工具from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个整数相加并返回结果。 return a b再定义一个模拟天气工具tool def get_weather(city: str) - str: 查询指定城市的当前天气。 weather_data { 北京: 晴天25度, 上海: 多云28度, } return weather_data.get(city, 暂无该城市天气数据)这里的函数注释很重要它是给模型看的。模型通过函数名、参数描述和注释来判断什么时候调用工具。如果你把注释写得很模糊模型可能不会调用或者乱传参数。绑定工具需要你的模型服务支持 function calling / tool calling。如果不支持后面的 Agent 流程就做不了tools [add, get_weather] llm_with_tools llm.bind_tools(tools) resp llm_with_tools.invoke([HumanMessage(content北京今天多少度)]) print(resp) # 通常是一个包含 tool_calls 的响应如果模型决定调用工具你会看到响应里有tool_calls。你还需要手动执行工具并把结果返回给模型这也是 Agent 框架替我们做的事。所以不理解原理时跑一次再观察响应结构比直接套 Agent 更好。3.4 用 Agent 把多个工具交给模型调度手动处理工具调用很繁琐尤其工具一多要写很多循环逻辑。LangChain 的 Agent 把“决定调用哪个工具、执行、回传、再生成”这个循环封装起来了。示意代码from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate agent_prompt ChatPromptTemplate.from_messages([ (system, 你是一个智能助手可以使用工具。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(llm, tools, agent_prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: 北京今天多少度然后帮我算 23 45}) print(result)这段代码里不要手动传agent_scratchpadAgent 会自己填充。verboseTrue可以让你看到整个执行过程非常有用。我不建议一上手就写多工具复杂 Agent。先放两个工具跑通“模型识别意图、调用工具、组织回答”的完整流程。如果模型频繁调用错误工具优先检查工具描述和参数约束不一定改代码。比如get_weather的参数名最好就是city不要传一个 JSON 对象省得模型解析错。另外Agent 不是万能的。它会增加请求次数也会增加失败概率。生产环境如果业务流程固定直接用普通链更稳。4. 再进一步搭一个本地 RAG 问答系统4.1 文档加载与拆分的正确姿势RAG 是现在落地最多的 LangChain 场景先加载你自己的文档切成小块向量化存入向量库用户提问时先检索相关片段再把片段拼进提示词让模型回答。加载方式按文件类型来。纯文本最简单from langchain_community.document_loaders import TextLoader loader TextLoader(docs/raw.txt, encodingutf-8) docs loader.load()如果是 PDF、Word、Markdown需要装对应加载器。注意编码问题中文文档经常因为gbk或utf-8不一致读不出来。报错时先看是UnicodeDecodeError还是文件路径不存在不要急着换库。拆分是 RAG 里容易被忽略的一环。我常用的参数from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , ], ) chunks splitter.split_documents(docs) print(len(chunks)) print(chunks[0].page_content)chunk_size是每个块的最大字符数chunk_overlap是相邻块的重叠字符数。重叠是为了避免句子被切断后丢失上下文。不要设太小比如 100 字检索时信息量不够也不要设太大比如 2000 字超出模型上下文窗口后要么截断要么浪费 token。我一般从 500 到 800 开始调。4.2 向量化与检索向量化需要选择 embedding 模型。你可以用 API 服务也可以用本地模型。API 方式省资源但每次调用有成本。本地方式更隐私但需要一定内存。from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS embeddings OpenAIEmbeddings( modelyour-embedding-model, api_keyyour-api-key, base_urlhttps://your-provider-endpoint/v1, ) vectorstore FAISS.from_documents(chunks, embeddings) retriever vectorstore.as_retriever(search_kwargs{k: 4})k表示每次检索返回几块文档。太小可能漏内容太大可能引入噪声。我通常先从 3 到 5 开始看回答质量再调整。向量库这里选了 FAISS适合单机场景如果想做千万级数据再换 Milvus 或其他数据库。第一次from_documents会有点慢因为要把所有 chunk 向量化。如果文档很多先跑一个小样确认切分和向量化没问题再全量跑。不要一上来就灌几千个文档出了错会很难定位。4.3 问答链组装与效果验证检索出来后要把检索到的多个文档拼成上下文再交给模型from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate rag_prompt ChatPromptTemplate.from_template( 请根据以下资料回答问题。如果资料中没有相关内容请直接回答“资料中未提及”。 资料 {context} 问题{question} ) def format_docs(doc_list): return \n\n.join([d.page_content for d in doc_list]) rag_chain ( {context: retriever | format_docs, question: lambda x: x[question]} | rag_prompt | llm | StrOutputParser() )调用一次result rag_chain.invoke({question: 这份文档里提到了哪些部署步骤}) print(result)如果回答不准确不要先调模型先看retriever检索到了什么。可以在调试时单独打印检索结果for i, doc in enumerate(retriever.invoke(部署步骤)): print(i, doc.page_content[:200])如果检索结果和问题不相关说明切块大小、k值、文档内容质量有问题。如果检索结果是相关的但模型回答不对再考虑改提示词或换更强模型。4.4 从单条问答到批量接口单条问答跑通后批量处理也很重要。常见场景是给一批问题生成答案或者对外提供服务。批量处理时不要只写一个 for 循环就完事要考虑失败重试和输出保存。questions [ 这份文档适合什么人, 安装依赖需要哪些步骤, 数据量大的时候怎么处理, ] results [] for i, q in enumerate(questions): try: ans rag_chain.invoke({question: q}) results.append({id: i, question: q, answer: ans}) print(i, ok) except Exception as e: results.append({id: i, question: q, answer: , error: str(e)}) print(i, failed, e)之后再统一输出到文件。如果你要暴露成 API用 FastAPI 包一层from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QARequest(BaseModel): question: str session_id: str default app.post(/qa) def qa_api(req: QARequest): answer rag_chain.invoke({question: req.question}) return {question: req.question, answer: answer}这里需要注意rag_chain在服务里是全局变量初始化一次即可不要每次请求都重新加载文档和向量库。对于更高并发可以再加异步接口或缓存。5. 常见报错、性能边界和生产化建议5.1 按现象排查先看输入、再看日志、最后看依赖我踩过最多的坑并不是 LangChain 本身有多复杂而是问题定位顺序不对。你按下面顺序排查基本能解决八成问题第一看现象。是报错、卡死、无输出还是输出质量差不同现象对应不同方向。报错看堆栈卡死看网络和资源无输出看提示词和输出解析质量差看输入内容。第二看输入。RAG 里最常见的问题是文档没加载成功、切分后空块、检索结果为空。先打印len(chunks)和retriever.invoke的结果确认输入链路是完整的。第三看配置。api_key有没有填对base_url有没有多写路径模型名是否有效温度参数是否设置过高。这些参数一旦有问题模型会返回 401、404 或者奇怪内容。第四看依赖。langchain系列包之间版本不匹配会报ImportError或TypeError。建议同一个虚拟环境里安装不要混用 pip 和 conda 源。遇到Pydantic报错多半是某个包版本被安装成了冲突版本。第五再看框架限制。比如模型不支持工具调用但硬要跑 Agent文档超出上下文长度向量维度不匹配。这些属于能力边界不是 bug。5.2 资源占用和并发别一上来就开最大并发如果你用 API 模型本地主要消耗内存和少量 CPU显存几乎不占。如果你用本地模型比如通过 Ollama 或 vLLM 起服务显存占用就很重要。低显存机器不代表不能跑但要把模型换成小尺寸并且降低并发数。批量任务也不要一上来就开 20 个并发。很多模型服务有 QPS 限制并发太高会触发限流反而拖慢整体速度。我一般按这个顺序测先单线程跑 10 条确认稳定性。再开 3 到 5 个并发观察响应时间和失败率。最后根据模型服务商限制逐步提升到合理并发。另外要注意输出目录。批量任务如果都写同一个文件容易互相覆盖。给每次运行生成带时间戳的目录把结果、日志单独放。这个习惯在调试时特别有用。5.3 生产化落地日志、缓存、失败重试与版本锁定Demo 跑通不等于服务稳定。真要放到生产环境至少要补齐四件事。日志是第一位。print只能临时看生产环境必须记录请求 ID、输入摘要、耗时、错误信息。LangChain 里开启 verbose 只能看流程不能代替业务日志。缓存是第二位。相同问题可以缓存结果尤其是检索结果能省很多 token。但要注意缓存键不能只放问题要把版本号和模型参数也带上否则改提示词后缓存会串。失败重试要区分错误类型。网络超时可以重试模型报context length就不要再重试先压缩输入。工具调用失败要看是参数错误还是服务不可用不能盲目重试。这里做个简单的指数退避比立即刷请求更稳妥。最后是依赖版本锁定。LangChain 1.x 迭代很快今天能跑的代码两周后升级小版本可能就报错了。生产环境建议在requirements.txt里锁定具体版本升级前先在测试环境跑一遍完整流程。还有一点敏感信息不要写进日志和缓存。如果做的是企业问答文档里可能有内部数据检索结果会进入模型请求这一点要在设计阶段就和业务方确认数据边界。网络上还常见把 API Key 传到公网仓库的问题这种事风险很大建议直接放进环境变量或密钥管理服务。现在低代码平台也逐渐接入大模型编排能力比如 Mendix 这类产品也有 AI 相关组件。但对于需要精细控制、排查问题、对接私有数据的场景LangChain 这类代码方案仍然更灵活。低代码适合快速搭界面和简单流程复杂链路的调试还是代码更直观。最后留一个个人经验先跑通最小链路再逐步加组件。不管是 LangChain 1.3 还是其他版本真正影响落地的往往不是某个高级概念而是环境、输入格式、参数边界和失败处理。把这几件事理顺了剩下的就是耐心调提示词。
