LangChain 1.3实战教程:从零搭建大模型问答系统
2026/8/31 6:48:03 网站建设 项目流程

LangChain 1.3 这类框架,市面上教程很多,但多数都在讲概念:什么是 Chain、什么是 Agent、什么是记忆。真正能照着做、能把一个问答系统跑起来、能从单条测试走到批量任务的实战教程,反而很少。这篇文章我按自己实际跑过的顺序来写,先讲清楚它解决什么问题,再给环境、代码、参数和排查方法。适合三类人:刚接触大模型应用开发、想把 LangChain 从 Demo 升级成稳定服务、以及被各种概念绕晕但需要尽快写出可运行代码的人。我的核心建议是:不要一上来就研究所有组件,先把最小链路跑通,再往里面加记忆、工具和检索。

1. 先搞清楚:LangChain 1.3 到底帮你解决什么问题

1.1 不要把 LangChain 当成大模型本身

很多人第一次用 LangChain 会有一个误解,以为装上它就有了智能问答能力。其实 LangChain 本身不提供模型,它只负责把模型能力、工具、数据、记忆、业务流程串起来。你可以把它理解成一条生产线:模型是机器,工具是辅料,提示词是操作说明书,LangChain 提供的是传送带和控制逻辑。

在实际项目中,这个定位决定了你该怎么用它。如果你想解决“让模型知道最新资料”,LangChain 负责把资料切块、检索、拼进提示词;如果你想让模型调用外部 API,LangChain 负责定义工具、传给模型、执行工具并返回结果;如果你要做多轮对话,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 "langchain==1.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_urlapi_key按服务商文档填就行。还有一点,1.3 版本如果报ModuleNotFoundError,先别急着搜代码,多数情况是某个分包没有安装,例如langchain_text_splitterslangchain_core

2.2 模型接入:API Key、基础模型、温度参数

不管后面做什么,先要有一个能调通的模型对象。以 OpenAI 兼容接口为例,代码结构如下:

from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="your-model-name", # 按模型服务商实际的模型名填写 api_key="your-api-key", # 建议从环境变量读取,不要硬编码 base_url="https://your-provider-endpoint/v1", # OpenAI兼容接口地址 temperature=0.3, )

这里有几个参数值得说清楚:

  • model:不同服务商支持不同模型,不要照抄我的名字。
  • temperature:控制随机性,0 到 1 之间。做问答、抽取、分类我一般设 0 到 0.3;创意文案可以调到 0.7 以上。
  • base_url:只填到/v1级别,不要在后面多加路径,否则容易 404。
  • api_key:放在环境变量里更安全,ChatOpenAI(api_key=os.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)

这里用到了 LCEL(LangChain 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_key="question", history_messages_key="history", )

调用时通过configsession_id

resp = 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(agent=agent, tools=tools, verbose=True) result = agent_executor.invoke({"input": "北京今天多少度?然后帮我算 23 + 45"}) print(result)

这段代码里不要手动传agent_scratchpad,Agent 会自己填充。verbose=True可以让你看到整个执行过程,非常有用。

我不建议一上手就写多工具复杂 Agent。先放两个工具,跑通“模型识别意图、调用工具、组织回答”的完整流程。如果模型频繁调用错误工具,优先检查工具描述和参数约束,不一定改代码。比如get_weather的参数名最好就是city,不要传一个 JSON 对象,省得模型解析错。

另外,Agent 不是万能的。它会增加请求次数,也会增加失败概率。生产环境如果业务流程固定,直接用普通链更稳。

4. 再进一步:搭一个本地 RAG 问答系统

4.1 文档加载与拆分的正确姿势

RAG 是现在落地最多的 LangChain 场景:先加载你自己的文档,切成小块,向量化存入向量库,用户提问时先检索相关片段,再把片段拼进提示词让模型回答。

加载方式按文件类型来。纯文本最简单:

from langchain_community.document_loaders import TextLoader loader = TextLoader("docs/raw.txt", encoding="utf-8") docs = loader.load()

如果是 PDF、Word、Markdown,需要装对应加载器。注意编码问题,中文文档经常因为gbkutf-8不一致读不出来。报错时先看是UnicodeDecodeError还是文件路径不存在,不要急着换库。

拆分是 RAG 里容易被忽略的一环。我常用的参数:

from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, 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( model="your-embedding-model", api_key="your-api-key", base_url="https://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系列包之间版本不匹配会报ImportErrorTypeError。建议同一个虚拟环境里安装,不要混用 pip 和 conda 源。遇到Pydantic报错,多半是某个包版本被安装成了冲突版本。

第五,再看框架限制。比如模型不支持工具调用,但硬要跑 Agent;文档超出上下文长度;向量维度不匹配。这些属于能力边界,不是 bug。

5.2 资源占用和并发:别一上来就开最大并发

如果你用 API 模型,本地主要消耗内存和少量 CPU,显存几乎不占。如果你用本地模型,比如通过 Ollama 或 vLLM 起服务,显存占用就很重要。低显存机器不代表不能跑,但要把模型换成小尺寸,并且降低并发数。

批量任务也不要一上来就开 20 个并发。很多模型服务有 QPS 限制,并发太高会触发限流,反而拖慢整体速度。我一般按这个顺序测:

  1. 先单线程跑 10 条,确认稳定性。
  2. 再开 3 到 5 个并发,观察响应时间和失败率。
  3. 最后根据模型服务商限制,逐步提升到合理并发。

另外要注意输出目录。批量任务如果都写同一个文件,容易互相覆盖。给每次运行生成带时间戳的目录,把结果、日志单独放。这个习惯在调试时特别有用。

5.3 生产化落地:日志、缓存、失败重试与版本锁定

Demo 跑通不等于服务稳定。真要放到生产环境,至少要补齐四件事。

日志是第一位。print只能临时看,生产环境必须记录请求 ID、输入摘要、耗时、错误信息。LangChain 里开启 verbose 只能看流程,不能代替业务日志。

缓存是第二位。相同问题可以缓存结果,尤其是检索结果,能省很多 token。但要注意:缓存键不能只放问题,要把版本号和模型参数也带上,否则改提示词后缓存会串。

失败重试要区分错误类型。网络超时可以重试,模型报context length就不要再重试,先压缩输入。工具调用失败要看是参数错误还是服务不可用,不能盲目重试。这里做个简单的指数退避,比立即刷请求更稳妥。

最后是依赖版本锁定。LangChain 1.x 迭代很快,今天能跑的代码,两周后升级小版本可能就报错了。生产环境建议在requirements.txt里锁定具体版本,升级前先在测试环境跑一遍完整流程。

还有一点,敏感信息不要写进日志和缓存。如果做的是企业问答,文档里可能有内部数据,检索结果会进入模型请求,这一点要在设计阶段就和业务方确认数据边界。网络上还常见把 API Key 传到公网仓库的问题,这种事风险很大,建议直接放进环境变量或密钥管理服务。

现在低代码平台也逐渐接入大模型编排能力,比如 Mendix 这类产品也有 AI 相关组件。但对于需要精细控制、排查问题、对接私有数据的场景,LangChain 这类代码方案仍然更灵活。低代码适合快速搭界面和简单流程,复杂链路的调试还是代码更直观。

最后留一个个人经验:先跑通最小链路,再逐步加组件。不管是 LangChain 1.3 还是其他版本,真正影响落地的往往不是某个高级概念,而是环境、输入格式、参数边界和失败处理。把这几件事理顺了,剩下的就是耐心调提示词。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询