系列导航：[第一篇：LangChain 1.0 是什么，解决了什么问题] → 第二篇（当前） → 第三篇：PromptTemplate 详解 前置要求：已完成第一篇的环境搭建，能调通至少一个模型 本篇目标：深入掌握 LangChain 的模型接口层，学会四种调用方式、结构化输出、多模型切换，以及在生产环境中的常用配置技巧
系列导航：[第一篇：LangChain 1.0 是什么，解决了什么问题] → 第二篇（当前） → 第三篇：PromptTemplate 详解
前置要求：已完成第一篇的环境搭建，能调通至少一个模型
本篇目标：深入掌握 LangChain 的模型接口层，学会四种调用方式、结构化输出、多模型切换，以及在生产环境中的常用配置技巧
LangChain 里有两种模型接口，很多初学者容易混淆：
OpenAI
/v1/completions
ChatOpenAI
/v1/chat/completions
结论：现在几乎所有主流模型都是 ChatModel，你 99% 的时候只需要用 ChatOpenAI 这类 Chat 接口。 LLM 接口是上一代的遗留产物（对应 GPT-3 时代的 text-davinci 系列），现在的 GPT-4、DeepSeek、Qwen、Claude 都走 Chat 接口。
本篇重点讲 ChatModel，LLM 接口只在对比时提及。
在调用 ChatModel 之前，需要理解 LangChain 的消息体系。它把对话中的每一条消息都封装成一个对象：
from langchain_core.messages import ( SystemMessage, # 系统提示词 HumanMessage, # 用户消息 AIMessage, # 模型回复 ToolMessage, # 工具调用结果 FunctionMessage, # 已废弃，用 ToolMessage 替代 )
这些消息类型和原生 API 的 role 字段是直接对应的：
role
# 原生 API 写法 messages = [ {"role": "system", "content": "你是一个专业的代码审查工程师"}, {"role": "user", "content": "请审查这段代码：..."}, {"role": "assistant", "content": "这段代码有以下几个问题..."}, {"role": "user", "content": "第一个问题怎么修复？"}, ] # LangChain 写法（等价） from langchain_core.messages import SystemMessage, HumanMessage, AIMessage messages = [ SystemMessage(content="你是一个专业的代码审查工程师"), HumanMessage(content="请审查这段代码：..."), AIMessage(content="这段代码有以下几个问题..."), HumanMessage(content="第一个问题怎么修复？"), ]
使用消息对象而非字典的优势在于：类型安全、IDE 自动补全、以及后续可以直接传入 Memory 和 Agent 等组件。
from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() # 基础初始化（从环境变量读取 OPENAI_API_KEY） llm = ChatOpenAI(model="gpt-4o") # 完整参数初始化 llm = ChatOpenAI( model="gpt-4o", temperature=0.7, # 创造性，0~2，默认 1 max_tokens=2048, # 最大输出 Token 数 timeout=30, # 请求超时（秒） max_retries=3, # 失败自动重试次数 )
from langchain_anthropic import ChatAnthropic llm = ChatAnthropic( model="claude-opus-4-5", temperature=0.7, max_tokens=2048, )
国内的 DeepSeek、通义千问、智谱 GLM、零一万物等主流模型都提供 OpenAI 兼容接口，直接用 langchain_openai 即可，无需安装额外的包：
langchain_openai
import os from langchain_openai import ChatOpenAI # ---- DeepSeek ---- llm_deepseek = ChatOpenAI( model="deepseek-chat", openai_api_key=os.getenv("DEEPSEEK_API_KEY"), openai_api_base="https://api.deepseek.com/v1", temperature=0.7, ) # ---- 通义千问 Qwen ---- llm_qwen = ChatOpenAI( model="qwen-plus", openai_api_key=os.getenv("DASHSCOPE_API_KEY"), openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", ) # ---- 智谱 GLM-4 ---- llm_glm = ChatOpenAI( model="glm-4", openai_api_key=os.getenv("ZHIPU_API_KEY"), openai_api_base="https://open.bigmodel.cn/api/paas/v4/", ) # ---- 零一万物 Yi ---- llm_yi = ChatOpenAI( model="yi-large", openai_api_key=os.getenv("YI_API_KEY"), openai_api_base="https://api.lingyiwanwu.com/v1", )
如果你在本地跑 Ollama，同样走兼容接口：
# 需要先启动 Ollama：ollama serve # 并拉取模型：ollama pull qwen2.5:7b llm_local = ChatOpenAI( model="qwen2.5:7b", openai_api_key="ollama", # Ollama 不校验 key，随意填 openai_api_base="http://localhost:11434/v1", temperature=0.7, )
LangChain 1.0 的所有模型和链都实现了统一的 Runnable 接口，提供四种调用方式，适用于不同场景：
Runnable
from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage llm = ChatOpenAI(model="gpt-4o") # 方式一：传入消息列表（标准方式） response = llm.invoke([HumanMessage(content="什么是向量数据库？")]) # 方式二：直接传字符串（快捷方式，内部会自动包装成 HumanMessage） response = llm.invoke("什么是向量数据库？") # response 是 AIMessage 对象 print(response.content) # 回复文本 print(response.usage_metadata) # Token 用量统计 print(response.response_metadata) # 模型元信息（finish_reason 等）
AIMessage 的主要属性：
AIMessage
print(response.content) # str: 模型回复的文本 print(response.usage_metadata) # dict: {'input_tokens': 20, 'output_tokens': 150, 'total_tokens': 170} print(response.id) # str: 本次调用的唯一 ID
流式输出对用户体验影响很大，LangChain 的 stream 开箱即用，不需要手动处理 SSE：
llm = ChatOpenAI(model="gpt-4o") # stream 返回一个生成器，每次 yield 一个 AIMessageChunk for chunk in llm.stream("用 500 字介绍一下 RAG 技术"): print(chunk.content, end="", flush=True) print() # 最后换行
如果你需要在流式结束后拿到完整的 Token 统计，可以这样做：
full_response = None for chunk in llm.stream("介绍 RAG 技术"): print(chunk.content, end="", flush=True) if full_response is None: full_response = chunk else: full_response = full_response + chunk # AIMessageChunk 支持 + 拼接 print() print("总 Token 数：", full_response.usage_metadata)
一次发送多个请求，内部并发执行，适合批量处理任务：
questions = [ "什么是 Transformer？", "什么是注意力机制？", "什么是位置编码？", ] # batch 接受列表，返回列表，内部并发执行 responses = llm.batch(questions) for q, r in zip(questions, responses): print(f"Q: {q}") print(f"A: {r.content[:100]}...\n")
控制并发数，避免触发 API 限流：
# max_concurrency 控制最大并发请求数 responses = llm.batch(questions, config={"max_concurrency": 3})
生产环境（如 FastAPI）建议使用异步接口，避免阻塞事件循环：
import asyncio from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o") async def async_demo(): # 异步单次调用 response = await llm.ainvoke("什么是 LangChain？") print(response.content) # 异步流式输出 async for chunk in llm.astream("介绍一下向量数据库"): print(chunk.content, end="", flush=True) print() # 异步批量调用 questions = ["问题一", "问题二", "问题三"] responses = await llm.abatch(questions) for r in responses: print(r.content[:50]) asyncio.run(async_demo())
在 FastAPI 中的标准用法：
from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI app = FastAPI() llm = ChatOpenAI(model="gpt-4o") @app.get("/chat") async def chat(question: str): # 非流式 response = await llm.ainvoke(question) return {"answer": response.content} @app.get("/chat/stream") async def chat_stream(question: str): # 流式，返回 SSE async def generate(): async for chunk in llm.astream(question): yield f"data: {chunk.content}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream")
这是 LangChain 1.0 中非常实用的功能。在原生调用中，让模型返回 JSON 通常需要：在提示词里说"请用 JSON 格式回复"、手动解析字符串、处理解析失败的异常。LangChain 提供了更优雅的方式。
from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI # 定义期望的输出结构 class CodeReview(BaseModel): """代码审查结果""" has_bug: bool = Field(description="是否存在 Bug") bug_description: str = Field(description="Bug 的描述，如果没有则为空字符串") severity: str = Field(description="严重程度：low / medium / high") suggestions: list[str] = Field(description="改进建议列表") llm = ChatOpenAI(model="gpt-4o") # with_structured_output 返回一个新的 Runnable，输出直接是 Pydantic 对象 # 如果你使用的是deepseek模型，此处可能会报错，需要改成llm.with_structured_output(CodeReview,method="function_calling") # 因为deepseek目前不支持DeepSeek API 目前不支持 response_format 参数（即 JSON mode）。 structured_llm = llm.with_structured_output(CodeReview) code = """ def divide(a, b): return a / b """ result = structured_llm.invoke(f"请审查以下代码：\n{code}") # result 直接是 CodeReview 实例，不需要任何解析 print(type(result)) # <class 'CodeReview'> print(result.has_bug) # True print(result.severity) # high print(result.bug_description) # 未处理除数为零的情况 for s in result.suggestions: print(f" - {s}")
不需要 Pydantic 的验证能力时，可以用 TypedDict：
from typing import TypedDict from langchain_openai import ChatOpenAI class SentimentResult(TypedDict): sentiment: str # positive / negative / neutral confidence: float # 0.0 ~ 1.0 keywords: list[str] # 情感关键词 llm = ChatOpenAI(model="gpt-4o") structured_llm = llm.with_structured_output(SentimentResult) result = structured_llm.invoke("这款产品真的太棒了，用了之后爱不释手！") print(result["sentiment"]) # positive print(result["confidence"]) # 0.95 print(result["keywords"]) # ['太棒了', '爱不释手']
with_structured_output()
JsonOutputParser
PydanticOutputParser
推荐策略： 使用支持 Function Call 的模型时，优先用 with_structured_output()，它是最可靠的方式。
# with_structured_output 内部机制： # 1. 把 Pydantic Schema 转成 Function Call 的 JSON Schema # 2. 调用模型时传入这个 Schema 作为工具定义 # 3. 模型输出符合 Schema 的 JSON，LangChain 自动解析成对象 # 所以它的成功率远高于"请用 JSON 格式回复"的提示词引导方式
一个完整的信息抽取示例，演示如何从非结构化文本中提取结构化数据：
from typing import Optional from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class PersonInfo(BaseModel): """从文本中提取的人物信息""" name: str = Field(description="人物姓名") age: Optional[int] = Field(description="年龄，无法确定时为 null", default=None) occupation: Optional[str] = Field(description="职业", default=None) location: Optional[str] = Field(description="所在地", default=None) skills: list[str] = Field(description="技能列表，没有则为空列表", default_factory=list) llm = ChatOpenAI(model="gpt-4o") extractor = llm.with_structured_output(PersonInfo) text = """ 张伟，28岁，目前在北京一家互联网公司担任后端工程师。 他精通 Python 和 Go 语言，对分布式系统有深入研究， 业余时间喜欢写技术博客，在 GitHub 上有不少开源项目。 """ info = extractor.invoke(f"从以下文本中提取人物信息：\n{text}") print(f"姓名：{info.name}") print(f"年龄：{info.age}") print(f"职业：{info.occupation}") print(f"地点：{info.location}") print(f"技能：{', '.join(info.skills)}")
初始化 ChatOpenAI 时有很多参数，下面列出生产环境中最常用的几个：
from langchain_openai import ChatOpenAI llm = ChatOpenAI( # ---- 必填 ---- model="gpt-4o", # 模型名称 # ---- 生成控制 ---- temperature=0.7, # 创造性程度，0=确定性，>1=更随机，默认 1 # 代码生成建议 0~0.3，创意写作可用 0.7~1.2 max_tokens=2048, # 最大输出 Token 数，不设则用模型默认值 top_p=1.0, # 核采样，一般不与 temperature 同时调整 # ---- 工程配置 ---- timeout=60, # 请求超时秒数，长输出任务建议适当调大 max_retries=3, # 网络错误自动重试次数 request_timeout=60, # 同 timeout，兼容旧版写法 # ---- 认证 ---- openai_api_key="sk-xxx", # 不填则读取环境变量 OPENAI_API_KEY openai_api_base="https://...", # 自定义 base url，用于接入兼容接口 )
有时需要在调用时临时覆盖初始化时的参数（比如某些请求需要更高的 temperature）：
llm = ChatOpenAI(model="gpt-4o", temperature=0.3) # 用 bind 创建一个参数被覆盖的新 Runnable，不影响原始 llm creative_llm = llm.bind(temperature=1.2) precise_llm = llm.bind(temperature=0.0) # 各自独立使用 poem = creative_llm.invoke("写一首关于秋天的诗") calc = precise_llm.invoke("1 + 1 等于几？")
这是 LangChain 统一接口最直观的应用场景：写一次业务逻辑，批量测试不同模型的效果。
import os from langchain_openai import ChatOpenAI from langchain_anthropic import ChatAnthropic from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 统一的业务链（prompt + parser） prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的技术文档写作专家。"), ("human", "请用不超过 200 字解释：{concept}"), ]) parser = StrOutputParser() # 准备多个模型 models = { "GPT-4o": ChatOpenAI(model="gpt-4o"), "DeepSeek": ChatOpenAI( model="deepseek-chat", openai_api_key=os.getenv("DEEPSEEK_API_KEY"), openai_api_base="https://api.deepseek.com/v1", ), "Claude-3.5": ChatAnthropic(model="claude-opus-4-5"), } concept = "Transformer 的注意力机制" print(f"问题：{concept}\n{'='*60}") for name, model in models.items(): # 每个模型只需替换 llm，prompt 和 parser 完全复用 chain = prompt | model | parser try: response = chain.invoke({"concept": concept}) print(f"\n【{name}】") print(response) except Exception as e: print(f"\n【{name}】调用失败：{e}")
在有成本控制需求的项目中，追踪 Token 用量很重要：
from langchain_core.callbacks import UsageMetadataCallbackHandler # 方式一：从 AIMessage 获取（单次调用） response = llm.invoke("什么是 RAG？") print(response.usage_metadata) # {'input_tokens': 12, 'output_tokens': 200, 'total_tokens': 212} # 方式二：用回调追踪（适合链式调用） callback = UsageMetadataCallbackHandler() chain = prompt | llm | parser response = chain.invoke( {"concept": "向量数据库"}, config={"callbacks": [callback]} ) print(callback.usage_metadata)
开发调试阶段，重复相同问题会浪费 Token。可以开启缓存：
from langchain_core.globals import set_llm_cache from langchain_community.cache import InMemoryCache, SQLiteCache # 内存缓存（进程重启后失效） set_llm_cache(InMemoryCache()) # SQLite 持久化缓存（适合开发调试） set_llm_cache(SQLiteCache(database_path=".langchain_cache.db")) # 开启缓存后，相同输入直接返回缓存结果，不消耗 Token llm = ChatOpenAI(model="gpt-4o") response1 = llm.invoke("什么是 LangChain？") # 实际调用 API response2 = llm.invoke("什么是 LangChain？") # 命中缓存，瞬间返回
当主模型不可用时自动切换到备用模型：
from langchain_openai import ChatOpenAI primary_llm = ChatOpenAI(model="gpt-4o") fallback_llm = ChatOpenAI( model="deepseek-chat", openai_api_key=os.getenv("DEEPSEEK_API_KEY"), openai_api_base="https://api.deepseek.com/v1", ) # with_fallbacks 返回一个带降级逻辑的 Runnable # 当 primary_llm 抛出异常时，自动切换到 fallback_llm llm_with_fallback = primary_llm.with_fallbacks([fallback_llm]) response = llm_with_fallback.invoke("你好") print(response.content)
调试时想看完整的请求和响应内容：
import langchain langchain.debug = True # 开启全局 Debug 日志，会打印详细的输入输出 # 或者针对单次调用开启 verbose from langchain_core.callbacks import StdOutCallbackHandler response = llm.invoke( "你好", config={"callbacks": [StdOutCallbackHandler()]} )
/chat/completions
SystemMessage / HumanMessage / AIMessage
openai_api_base
invoke
stream
batch
ainvoke
with_structured_output(PydanticModel)
llm.bind()
你现在应该能做到： - 接入任意 OpenAI 兼容接口的模型，只需修改两行参数 - 熟练使用四种调用方式，根据场景选择同步/异步/流式 - 用 with_structured_output() 替代手动 JSON 解析，大幅提升稳定性 - 用 with_fallbacks() 构建高可用的模型调用层
with_fallbacks()
《提示词工程的工程化——PromptTemplate 详解》
在拿到稳定的模型接口之后，下一步要解决的是提示词管理问题。你会学到： - 为什么散落在代码里的 f-string 提示词是一个工程隐患 - ChatPromptTemplate 的各种构建方式和最佳实践 - MessagesPlaceholder：多轮对话历史的正确注入方式 - Few-shot Prompt 模板的结构化构建 - 提示词的复用、组合与版本管理思路
ChatPromptTemplate
MessagesPlaceholder
代码仓库：本系列所有可运行代码示例统一维护在 GitHub，每篇文章对应一个独立目录，可直接克隆运行。 系列导航：[第一篇] → 第二篇（当前） → 第三篇 → ...
代码仓库：本系列所有可运行代码示例统一维护在 GitHub，每篇文章对应一个独立目录，可直接克隆运行。
系列导航：[第一篇] → 第二篇（当前） → 第三篇 → ...
还没有评论，来抢沙发吧！
博客管理员
40 篇文章
admin
还没有评论，来抢沙发吧！