LangChain 1.3实战：从环境搭建到API服务封装全流程

这次我们直接进入 LangChain 1.3 的代码实战。如果你之前看过一些 LangChain 教程，但停留在“看懂了、没写过”的状态，这篇文章会把从环境搭建、核心模块、批量任务到 API 服务封装的完整路径走一遍。文章面向 CSDN 读者，默认你已经会 Python 基础语法，能创建虚拟环境。

LangChain 1.3 的核心价值不是“又一个 Python 包”，而是把 LLM 应用开发中的重复动作标准化：模型接入、提示词管理、多轮记忆、工具调用、检索增强、批量执行、服务化暴露。它把这些能力拆成模块，按需组合。无论你接 OpenAI 兼容接口，还是本地部署 Ollama、vLLM，都可以用同一套代码骨架切换模型。

本文不会写那些“知道就行”的概念，只会给你能直接复制到本机跑通的示例。代码会尽量保持版本宽松，标注容易变化的点。你照着做，能很快得到一个带接口、能批量处理、可扩展的 LLM 应用骨架。下面按章节展开。

1. LangChain 1.3 核心能力速览

先把关键信息列出来，方便你判断要不要继续读。

这里要强调一下：LangChain 1.3 本身不提供大模型能力，它负责的是“模型之上的那一层开发逻辑”。你可以把它理解为“给 LLM 写的后端框架”，编排输入、输出、工具、上下文。这一层能做很多手工拼 prompt 做不到的事。

2. 适用场景与使用边界

LangChain 1.3 适合这几类读者：

• 正在做 LLM 应用原型验证，需要快速把模型接入业务代码。
• 需要构建 RAG 知识库问答，把本地文档变成可检索上下文。
• 需要开发带多轮记忆的对话机器人，而不是每次请求都无状态调用。
• 需要让模型调用外部工具，比如查数据库、查天气、执行简单计算。
• 需要把 LLM 能力封装成接口，给前端或其他服务调用。

不适合什么场景？

• 如果你只是调用一次 API 查个文本，不需要 LangChain，直接 requests 就行。
• 如果你要做大规模分布式训练、微调模型，这是训练框架的范畴。
• 如果你对响应延迟极度敏感，必须压到 50ms 内，LangChain 的编排层会带来额外开销，建议评估后再用。

使用边界要说明清楚。接入大模型时，无论使用在线 API 还是本地模型，都要注意数据隐私和版权合规。不要把未脱敏的用户数据、商业秘密、版权材料直接发到第三方接口。本地部署能降低数据外泄风险，但不能解决授权问题。搭建 Agent 时，如果模型具备执行能力，比如调用数据库、发邮件、操作文件，必须加权限控制、操作审计和人工审核，避免错误执行造成事故。

3. 环境准备与前置条件

LangChain 1.3 是一个偏应用层的框架，前置条件并不高。核心环境建议如下：

• 操作系统：Windows、Linux、macOS 都行。
• Python：建议 3.10 到 3.12，过高或过低都可能遇到依赖冲突。
• 虚拟环境：推荐 conda 或 venv，隔离项目依赖。
• 模型服务：如果你没有在线 API，本地需要安装 Ollama 或 vLLM。
• 磁盘空间：框架本身只占几百 MB，但下载后的本地模型通常是几个 GB。
• 包管理工具：pip 或 poetry。

在安装 LangChain 前，先确认 Python 版本。

python --version

然后创建虚拟环境。这里用 conda 示例：

conda create -n langchain-lab python=3.11 -y conda activate langchain-lab

安装核心依赖包。LangChain 1.x 的包结构变化较快，建议安装时查看官方文档。下面是常见的安装方式：

pip install langchain pip install langchain-core pip install langchain-openai

如果要接本地 Ollama 模型，再加上：

pip install langchain-community

如果要把服务封装成 FastAPI：

pip install fastapi uvicorn

安装完成后，验证版本，避免因为版本不一致导致示例跑不通。

python -c "import langchain; print(langchain.__version__)"

这里的版本输出就是当前环境的 LangChain 版本。后续代码如果遇到类名找不到，大概率是版本差异，按错误提示调整即可。

4. 快速搭建第一个 LangChain 项目

先建立项目结构。建议从一开始就按工程化目录组织文件，不要把所有代码堆在一个脚本里。

langchain-lab/ ├── .env ├── requirements.txt ├── main.py ├── app/ │ ├── __init__.py │ ├── chains.py │ ├── config.py │ ├── models.py │ └── prompts.py └── data/ └── documents/

看到这个结构，不要觉得复杂。第一版只需要 main.py、config.py、chains.py 三个文件，其他目录后续扩展。

创建 .env 文件，保存 API Key 和 Base URL。不要把它提交到 Git 仓库。

OPENAI_API_KEY=your-api-key OPENAI_BASE_URL=https://api.example.com/v1

config.py 负责读取配置。

import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL")

需要安装 python-dotenv：

pip install python-dotenv

接下来写第一个模型调用。下面的代码兼容 OpenAI 兼容接口，如果你使用本地 Ollama，只需要把ChatOpenAI换成ChatOllama。

from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage from app.config import OPENAI_API_KEY, OPENAI_BASE_URL llm = ChatOpenAI( model="gpt-4o-mini", api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL, temperature=0.7, ) response = llm.invoke([HumanMessage(content="用一句话介绍 LangChain")]) print(response.content)

运行：

python main.py

如果你看到控制台输出了模型回答，说明模型接入成功。如果报错，先检查 API Key、Base URL 是否正确，再检查网络是否能访问对应接口。

5. LangChain 1.3 核心模块代码实战

模型接入只是第一步。真正有价值的是后面这些模块。

5.1 提示词模板

直接拼字符串写 prompt，项目规模小还凑合，规模一大就失控。LangChain 的 PromptTemplate 帮你管理输入变量、固定指令和示例。

from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_messages( [ ( "system", "你是一个专业的{industry}咨询顾问，回答要简洁、准确、有依据。", ), ("human", "用户的问题是：{question}"), ] ) # 和模型组合 chain = prompt | llm result = chain.invoke( { "industry": "法律", "question": "合同里需要关注哪些风险条款？", } ) print(result.content)

这里的|是 LangChain 常见的组合操作符。它把提示词模板和模型串成一条调用链，代码很直观。实际开发时，可以把所有提示词集中到 prompts.py，方便多人维护和版本对比。

5.2 多轮记忆

无状态调用模型很容易，但对话应用必须带上历史上下文。LangChain 提供了多种记忆实现方式。下面是基础的示例。

from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_core.messages import HumanMessage history = InMemoryChatMessageHistory() history.add_user_message("我叫小明，是一名后端工程师。") # 在调用链中追加历史 messages = history.messages + [HumanMessage(content="我叫什么名字？")] response = llm.invoke(messages) print(response.content) history.add_ai_message(response.content)

生产环境不建议把所有历史放在内存里，最好用 Redis 或数据库保存。LangChain 的 RunnableWithMessageHistory 可以配合 session_id 管理多用户隔离，但具体实现需要按项目调整。基础逻辑是：每次请求前读取历史，拼接后调用模型，再将新对话写入历史。

5.3 工具调用与 Agent

让模型调用外部工具是 LangChain 1.3 的强项。先用@tool装饰器定义工具，再把工具交给 Agent。

from langchain_core.tools import tool from langchain.agents import create_tool_calling_agent, AgentExecutor @tool def add(a: int, b: int) -> int: """计算两个整数之和""" return a + b tools = [add] # 创建 Agent。这里的 llm 需要支持 tool calling agent = create_tool_calling_agent(llm, tools) agent_executor = AgentExecutor(agent=agent, tools=tools) result = agent_executor.invoke({"input": "请计算 12345 和 67890 的和"}) print(result["output"])

执行后，模型会生成工具调用参数，框架执行工具，把结果返回给模型，模型再汇总回答。这种方式比让模型直接心算更可靠。真实项目中，工具可能是查数据库、调用内部接口、发邮件。注意，凡是涉及写操作的工具，必须加确认环节。

5.4 RAG 检索增强生成

RAG 是 LangChain 最热门的应用方向。基本流程是：加载文档、切片、向量化、存入向量库、检索相关片段、注入 prompt、生成回答。

下面是最小可运行示例。先安装文档加载和向量库相关依赖：

pip install langchain-text-splitters langchain-community chromadb

常见的加载和切片代码：

from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter loader = TextLoader("data/documents/company_policy.txt") documents = loader.load() splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, ) chunks = splitter.split_documents(documents) print(f"文档切片数量：{len(chunks)}")

接着向量化并检索。如果是本地模型，Embedding 模型和 LLM 可以都走 Ollama。

from langchain_community.embeddings import OllamaEmbeddings from langchain_community.vectorstores import Chroma embeddings = OllamaEmbeddings(model="nomic-embed-text") vectorstore = Chroma.from_documents(documents=chunks, embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

最后把检索结果注入 prompt。这里直接写一个简单示例：

question = "公司年假政策是什么？" contexts = retriever.invoke(question) context_text = "\n".join([doc.page_content for doc in contexts]) prompt_text = f"""请根据下面的资料回答问题。如果资料里没有答案，请说明不知道。 资料： {context_text} 问题：{question} """ response = llm.invoke(prompt_text) print(response.content)

RAG 效果好不好，主要看三点：文档切分是否合理、Embedding 模型是否匹配领域、检索召回是否精准。LangChain 只是把流程串起来，领域的调优还是得自己做。

5.5 输出解析

模型返回的是字符串，但业务系统往往需要结构化数据。LangChain 的 StructuredOutputParser 可以把回答转成 JSON。

from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import PromptTemplate parser = JsonOutputParser() prompt = PromptTemplate( template="提取用户问题中的实体，输出 JSON 对象。\n问题：{question}\n{format_instructions}", input_variables=["question"], partial_variables={"format_instructions": parser.get_format_instructions()}, ) chain = prompt | llm | parser result = chain.invoke({"question": "帮我预订明天下午两点去上海的高铁票"}) print(result)

输出解析能帮你把模型结果直接喂给下一个系统，减少后续的字符串清洗工作。解析失败时可以增加重试机制，或者让模型只返回 JSON 代码块再单独提取。

6. 批量任务与接口 API 实战

聊完核心模块，来看工程落地的两个关键能力：批量处理和 API 封装。

6.1 批量任务处理

批量任务的核心是控制并发、记录进度、处理失败。最简单的实现是 for 循环，但效率低。推荐用 ThreadPoolExecutor 或 asyncio 并发调用模型接口。

from concurrent.futures import ThreadPoolExecutor, as_completed def process_text(text): response = llm.invoke(f"请总结以下内容：{text}") return response.content texts = [ "第一段需要总结的文本", "第二段需要总结的文本", # 更多数据 ] results = [] with ThreadPoolExecutor(max_workers=4) as executor: future_map = {executor.submit(process_text, t): t for t in texts} for future in as_completed(future_map): try: result = future.result() results.append(result) except Exception as e: print(f"处理失败：{e}")

并发数不要开太大。大多数在线 API 都有 QPS 限制，本地模型也会被显存和推理速度卡住。4 到 8 个并发通常比较稳妥。生产环境建议引入任务队列，比如 Redis 加 Celery，或者直接用消息队列，避免脚本崩溃后数据丢失。

6.2 封装 FastAPI 接口

把 LangChain 能力封装成 API，前端和业务系统就能直接调用。下面是带请求参数的 FastAPI 示例。

from fastapi import FastAPI from pydantic import BaseModel from app.chains import get_chat_chain app = FastAPI() class ChatRequest(BaseModel): question: str session_id: str = "default" class ChatResponse(BaseModel): answer: str chain = get_chat_chain() @app.post("/chat", response_model=ChatResponse) def chat(req: ChatRequest): result = chain.invoke({"question": req.question}) return ChatResponse(answer=result.content)

启动服务：

uvicorn main:app --host 0.0.0.0 --port 8000

用 curl 测试：

curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"question": "你好，请介绍一下你自己"}'

接口返回 JSON，前端可以直接解析。这里建议给接口加鉴权，至少用一个简单的 Token 校验，不要把没有保护的端口暴露到公网。

6.3 异步调用与超时控制

在线模型接口响应时间可能从几秒到几十秒。API 层一定要设置超时，避免请求卡死。比如：

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="gpt-4o-mini", timeout=60, max_retries=2, )

同时建议在上游幂等设计：如果服务超时，客户端可以重试，但要保证同一个请求不会产生重复副作用。对 Agent 来说，模型执行工具时如果重复调用“创建订单”这类接口会出问题，所以写操作工具必须幂等或加唯一请求 ID。

7. 性能、资源占用与稳定性观察

LangChain 是一个编排层，本身的资源占用很低，真正的消耗来自底层模型。

如果你调用在线 API，需要重点观察的是延迟、Token 消耗、失败率和并发限制。可以在每次请求前后埋点，打印耗时和 token 用量：

response = llm.invoke(prompt) print(response.usage_metadata)

如果你使用本地模型，需要观察显存占用、推理速度和并发时的显存峰值。以 7B 量化模型为例，常见部署方式下显存占用约 6G 到 8G，但不同量化等级、上下文长度、并发数都会导致明显波动。观察显存可以用 nvidia-smi：

nvidia-smi -l 2

如果显存不足，要做这几件事：

• 降低上下文长度，减少 prompt 中的历史消息数量。
• 使用量化模型，比如 GGUF、GPTQ。
• 降低并发数，排队执行。
• 改用更小的 Embedding 模型。
• 必要时换更大的显卡或多卡部署。

RAG 场景中，向量检索通常比 LLM 推理快得多，瓶颈在最后一步生成。批量处理大量文档时，建议先把文档切片和向量化跑完，再做检索和问答，不要反复读取原始文件。

稳定性方面，在线 API 经常会因为限流返回 429，本地模型也会因为显存不足直接崩溃。建议在工具函数里统一捕获异常，记录日志，并支持失败重试。状态码、耗时、Token 消耗都要进入日志，方便后续排查。

8. 常见问题与排查方法

下面是 LangChain 1.3 本地开发中最高频的问题和解决思路。

第一次跑通不要指望顺风顺水。建议先把“最小链路”跑通：一个模型调用、一个 prompt 模板、一个 API 接口。确认链路稳定后再加记忆、工具、RAG。

9. LangChain 1.3 最佳实践与工程建议

到这里，你应该已经能跑通一个基础项目。以下是进一步工程化时需要注意的经验。

第一，锁定版本。LangChain 迭代非常快，不同版本之间 API 可能有破坏性变更。在 requirements.txt 中锁定小版本号，升级时单独验证。

langchain==1.3.* langchain-core==1.3.* langchain-openai==1.3.*

第二，提示词也要做版本管理。把提示词模板放进独立目录或文件，修改时记录变更原因。提示词不是不可改的配置，它会直接影响效果，建议像管理代码一样管理。

第三，建立测试集。准备一组典型输入，比如 RAG 的 50 个问答对、Agent 的 20 个工具调用场景。每次修改后回归测试，对比输出质量。没有测试集，就很难判断“改 prompt 是变好了还是变坏了”。

第四，日志和链路追踪。给每个请求分配 request_id，记录 prompt 内容、模型响应、耗时和 Token 用量。问题排查时，没有日志寸步难行。

第五，安全合规。调用在线大模型时，敏感数据要脱敏或选择私有化部署。Agent 涉及外部操作时要加权限控制和人工审核。RAG 使用的文档要确认版权和授权，不能把未授权的商业文档直接做成知识库对外提供服务。

第六，如果需要交付给业务团队使用，可以考虑用低代码平台做一层可视化编排。低代码开发平台可以承载表单、审批、流程逻辑，LangChain 服务作为后端能力中心，通过 API 对接。这种方式适合企业内部快速搭建知识库问答、客服辅助等应用。两者的边界是：低代码负责流程和界面，LangChain 负责模型编排和自然语言能力。实际开发时，可以让 LangChain 项目先输出稳定的 API，再接入低代码平台，避免把模型调用逻辑散落在前端流程里。

10. 总结与下一步

LangChain 1.3 本身不复杂，门槛在于你能否快速建立“模型、提示词、记忆、工具、检索、接口”这条完整链路。这篇文章给出的最小项目可以作为一个起点。建议你先按第 4 章跑通模型调用，再逐步加入提示词模板、记忆、工具调用和 RAG。每加一个模块，就做一次验证，不要一次性堆太多功能。

最容易踩的坑主要有三个：版本变更导致的 API 差异、在线接口的限流超时、RAG 检索效果不理想。这三个问题都可以通过打印日志、锁定版本、建立测试集来缓解。本地显存不够时，不要反复调参硬扛，先换更小的量化模型或减少并发。

如果你要真正用于生产，下一步建议按两个方向扩展：一是完善 API 层，加入鉴权、限流、日志、超时重试；二是完善数据层，用 Redis 存记忆、用 PostgreSQL 存日志、用对象存储管理文档。等这两块补齐，你的 LangChain 项目就不再是一个脚本，而是一个可以持续迭代的 LLM 应用服务。