力 力达云
教程聚合APItoken成本国产模型资讯工具关于内测申请
← 返回资讯

LangChain 接入大模型：ChatOpenAI 快速上手

2026-06-22

你大概率是这么撞上 LangChain 的：产品经理丢过来一句”客服机器人先接 GPT-4o-mini 试试水，等哪天便宜的国产模型能打了就切过去”。如果你直接用 openai 官方 SDK 手写调用，等到真要换模型、加检索、加记忆的那天，你会发现提示词拼接、上下文管理、工具调用全是散装代码，改一处牵一片。LangChain 解决的正是这个问题：把 LLM 调用、提示模板、记忆、工具调用、RAG 这些能力抽象成统一的 Runnable 接口，用管道符 | 串成可组合的 chain，换模型只改一行配置，链路结构完全不用动。本文只讲 ChatOpenAI 这一条路径——因为几乎所有 OpenAI 兼容平台（力达云、DeepSeek、Moonshot、通义千问兼容模式）都能直接复用它，不用为每家平台单独装一个 LangChain 集成包。

安装

pip install langchain langchain-openai

langchain-openai 是 LangChain 官方维护的 OpenAI 集成包，包含 ChatOpenAI（对话模型）和 OpenAIEmbeddings（向量嵌入）。这里有个新手常踩的坑：LangChain 从 0.1 版本起把核心框架（langchain、langchain-core）和各家厂商集成（langchain-openai、langchain-anthropic、langchain-community）拆成了独立发布的包，版本号各走各的。如果你只装了 langchain 没装 langchain-openai，导入 ChatOpenAI 会直接报 ModuleNotFoundError: No module named 'langchain_openai'；反过来如果 langchain-openai 版本比 langchain-core 新太多，又可能报 Runnable 相关的接口不兼容错误。稳妥做法是三个包一起锁版本装：

pip install "langchain==0.3.*" "langchain-core==0.3.*" "langchain-openai==0.2.*"

拆分是有道理的——核心框架不需要为了支持某家新模型就整体发版，而各家 SDK 又能按自己的节奏迭代，这样耦合度低很多，你在生产环境锁版本也更省心。

配置 ChatOpenAI

import os
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    openai_api_base=os.environ.get("OPENAI_BASE_URL",
                                   "https://api.openai.com/v1"),
    temperature=0.7,
    streaming=False,
)

切换至力达云或 DeepSeek 只需改 openai_api_base，其余链路代码不动——这也是选 ChatOpenAI 而不是各家厂商专属 SDK 的核心理由：只要平台兼容 OpenAI 的 /chat/completions 接口协议，ChatOpenAI 就能无缝对接，你的 chain、memory、tool 全部原样复用。

几个容易被忽略、但线上必须调的参数说明一下：

  • temperature：0 到 2 之间，越低越确定（适合抽取结构化数据、代码生成），越高越发散（适合创意文案）。生产环境的客服/问答场景一般给 0.2~0.5，别照抄示例里的 0.7 就直接上线。
  • max_tokens：不设的话默认让模型自己决定输出长度，某些平台会给一个偏小的默认值导致回答被截断一半。线上建议显式设置，比如 max_tokens=1024，配合业务场景估算。
  • request_timeout：默认值在不同版本里是 60 秒或不设超时，网络抖动或模型排队严重时容易挂起整个请求线程。生产环境建议显式设 request_timeout=30，配合下文的重试策略一起用。
  • max_retries：LangChain 内置的自动重试次数，默认是 2。遇到 429（限流）或 5xx 错误会自动退避重试，但重试策略比较简单粗暴，高并发场景我更建议在业务层自己控制重试（后面有示例）。

基础调用与 LCEL Chain

LangChain Expression Language（LCEL）用 | 管道把提示模板和 LLM 组合成 chain：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的技术文档写手。"),
    ("user",   "{question}"),
])

chain = prompt | llm | StrOutputParser()
answer = chain.invoke({"question": "什么是 RAG？"})
print(answer)

这个 | 不是语法糖魔法，它是 Python 的运算符重载：ChatPromptTemplate、ChatOpenAI、StrOutputParser 全都实现了 Runnable 协议，__or__ 方法把它们包成一个 RunnableSequence。这么设计的好处是，chain 组装完之后自动获得四个能力，不用你逐个实现：

  • invoke()：同步单次调用，上面示例用的就是这个。
  • stream()：流式输出，下一节会讲。
  • batch()：批量并发调用一组输入，内部自动做并发调度，不用你手写线程池。
  • ainvoke() / astream() / abatch()：对应的异步版本，配合 FastAPI 这类异步框架直接能用，不用额外包一层 run_in_executor。

调试 chain 时如果结果不对，先别急着怀疑 prompt 写错了，用 chain.get_prompts() 能拿到最终渲染出的提示模板，或者给 invoke 加个 config={"callbacks": [ConsoleCallbackHandler()]} 把每一步的输入输出都打到控制台，比一路加 print 排查快得多。

流式输出

llm_stream = ChatOpenAI(
    model="gpt-4o-mini",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    openai_api_base=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    streaming=True,
)
for chunk in llm_stream.stream("用三句话解释向量数据库"):
    print(chunk.content, end="", flush=True)

流式输出底层其实就是 HTTP 的 SSE（Server-Sent Events），模型每生成几个 token 就往连接里推一段 data: {...}，ChatOpenAI 帮你把这些分片解析成一个个 chunk 对象，你拿到的 chunk.content 就是这次分片里的文本增量。为什么要折腾流式？因为首字延迟（Time To First Token）直接决定用户体验——非流式模式下用户要等模型把几百字全生成完才看到结果，流式模式下几百毫秒就能看到第一个字往外蹦，哪怕总耗时一样，感知上快了一个量级。

如果你的服务是 FastAPI 这种异步框架，别用同步的 stream()，改用 astream() 配合 StreamingResponse：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat")
async def chat(question: str):
    async def event_generator():
        async for chunk in llm_stream.astream(question):
            yield chunk.content
    return StreamingResponse(event_generator(), media_type="text/plain")

这里有个真实踩过的坑：如果你在同步视图函数里调用 stream()，又跑在 uvicorn 的异步事件循环里，会直接把整个 worker 阻塞住，其他请求全部排队，表现就是”并发一高整个服务就卡死”。排查时看着像是模型慢，实际上是同步阻塞调用堵住了事件循环，换成 astream() 立刻恢复正常。

与 RAG 集成（简化版）

组件LangChain 类说明
文档加载TextLoader / WebBaseLoader载入本地或网页文档
切片RecursiveCharacterTextSplitter按字符递归分块
向量化OpenAIEmbeddings生成 embedding
检索FAISS / Chroma相似度检索
问答RetrievalQA / LCEL chain拼接上下文回答

完整 RAG 参考：RAG 应用构建指南。

表里没写但你实操时一定会撞到的两个坑：一是 RecursiveCharacterTextSplitter 的 chunk_size 和 chunk_overlap 不是随便填的数字——chunk_size 太小（比如 100）会把一段完整语义切成好几段，检索时召回的都是残句，回答质量明显下降；chunk_size 太大（比如 4000）又会让每个 chunk 里混进太多无关信息，稀释掉真正相关的那句话。中文文档从 500~800 字符起步试，配合 chunk_overlap 设成 chunk_size 的 10%~20% 防止关键信息被切断在边界上。二是换 OpenAIEmbeddings 的模型时要注意向量维度——text-embedding-3-small 是 1536 维，如果你之前用别的模型建的向量库是 1024 维，直接换模型会在检索阶段报维度不匹配的错误，或者更隐蔽地——不报错但检索结果全是乱的，因为向量库允许不同维度共存但相似度计算完全失真。换 embedding 模型必须重建整个向量库，这个坑没有近路可抄。

常见问题

Q：langchain 与 langchain-openai 需要分开装吗？ 是的，v0.1 起官方把各家 LLM 集成拆分到独立包（langchain-openai、langchain-anthropic 等），核心框架与集成解耦，避免依赖爆炸。

Q：LangChain 版本迭代很快，如何锁版本？ 建议在 requirements.txt 锁定小版本，如 langchain==0.3.x langchain-openai==0.2.x，并在 CI 中做回归测试。

Q：调用 DeepSeek/Moonshot 报 Invalid model 怎么办？ ChatOpenAI 的 model 参数需填目标平台实际的模型名称，如 deepseek-chat、moonshot-v1-8k，而非 OpenAI 的模型名。

Q：报 openai.AuthenticationError: Error code: 401 怎么排查？ 先确认这三处是不是对上了：openai_api_key 环境变量确实注入到进程里了（本地 .env 文件没加载、容器里没传环境变量都会导致这个）；openai_api_base 指向的平台和你手里的 key 是不是同一家（拿力达云的 key 去请求 OpenAI 官方地址必报 401）；key 前面有没有多余的空格或换行符（复制粘贴时很容易带上）。三处都确认过还报错，去平台后台看看 key 是不是过期或被禁用了。

Q：报 RateLimitError: Error code: 429 怎么办？ 这是触发了平台的速率限制或余额不足，两种情况报的都是 429 但文案略有差别，看清楚 error.message 里到底是 rate_limit_exceeded 还是 insufficient_quota。前者是并发/QPS 超了目标，配合下文的重试退避处理；后者是账户余额或额度用完了，重试也没用，得先充值或换 key。

Q：报 context_length_exceeded 或类似的上下文超限错误怎么办？ 说明累计的 prompt token（历史消息 + 检索到的文档片段 + 当前问题）超过了模型的上下文窗口。别急着无脑加大 max_tokens（那是控制输出长度的，跟这个无关），而是要在业务层做截断：多轮对话保留最近 N 轮或用 ConversationSummaryMemory 把历史压缩成摘要；RAG 场景则减少检索返回的 k 值或把每个 chunk 切得更小。判断真实占用了多少 token，用 tiktoken 库离线数一遍最直接。

Q：中文输出偶尔出现乱码或半个字怎么回事？ 流式模式下模型是按 token 切分输出的，一个中文字符在某些编码方案下可能被切成两个 token 分片返回，如果你自己写代码按字节截断拼接就可能截出半个字。稳妥做法是别自己动手拼字节，直接用 chunk.content 拿到的字符串做字符串拼接，LangChain 内部已经处理好了分片对齐，不需要你自己按字节数组处理。

并发与成本控制

批量处理场景（比如给一批客户工单打标签）别写 for 循环挨个 invoke，用内置的 batch()，LangChain 会自动做并发调度，比手写线程池省心：

questions = [{"question": q} for q in ["什么是向量数据库？", "什么是 Embedding？", "什么是 RAG？"]]
answers = chain.batch(questions, config={"max_concurrency": 5})

max_concurrency 一定要显式设置，不设的话默认值在某些版本里是不限并发，直接把目标平台的 QPS 限制打爆，收获一片 429。异步场景同理用 abatch()，配合 asyncio.gather 也能自己实现，但没有内置的省心。

估算成本方面，如果你走的是 OpenAI 官方接口，可以用内置的 token 统计上下文管理器：

from langchain_community.callbacks import get_openai_callback

with get_openai_callback() as cb:
    chain.invoke({"question": "什么是 RAG？"})
    print(f"总 token: {cb.total_tokens}，预估花费: ${cb.total_cost}")

注意 get_openai_callback 的价格表是按 OpenAI 官方定价内置的，接第三方兼容平台（力达云、DeepSeek 这类）时 total_cost 这个字段是不准的——因为定价表根本不认识对方的计费标准，它只统计 total_tokens 是准的。真要算第三方平台的成本，拿到 total_tokens 之后按对方官方公布的单价（以官方为准，价格经常调整）自己乘一下，比信这个字段靠谱。

重试退避方面，ChatOpenAI 内置的 max_retries 策略比较简单，高并发场景我更倾向于在业务层用 tenacity 库自己控制指数退避：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=30))
def call_with_retry(question: str) -> str:
    return chain.invoke({"question": question})

这样能自己决定重试次数、退避的等待曲线，还能加日志记录每次重试的原因，比内置策略透明得多，线上排查问题时你会感谢自己当初这么写。

LangChain 该不该用？

不是所有场景都值得引入 LangChain 这层抽象，简单对比一下：

场景推荐方案理由
单次调用、无需切模型、不做 RAG官方 openai SDK 直连LangChain 的抽象反而增加理解成本，一个 client.chat.completions.create() 就够了
需要频繁切换模型/平台做对比测试LangChain ChatOpenAI只改 model 和 openai_api_base 两个参数，chain 结构完全不动
需要 RAG、多轮记忆、工具调用组合LangChain LCELRunnable 协议天然支持组合，不用自己攒胶水代码
强调文档级索引结构、多种检索策略对比LlamaIndex检索侧的抽象比 LangChain 更细，见LlamaIndex 接入大模型

我自己的判断标准很简单：项目一开始只是单纯调用模型出结果，别为了”显得专业”硬套框架；一旦你发现自己在手写 chain（提示词拼接完手动传给下一步、还要自己处理流式和重试），这就是该上 LangChain 的信号了。

延伸阅读：大模型 API 接入完全指南 · 接入教程 Hub · LlamaIndex 接入大模型 · Python 调用大模型 API 完整示例