第02章：LangChain 模型（Models）

版本：LangChain v1.3.7 | 讲师：汤姆小白

---

1. Models 概述

Models 模块是 LangChain 与语言模型交互的核心组件，负责整个流程中的"调用模型"环节。

完整的 Model I/O 流程分为三步：输入提示(Format) → 调用模型(Predict) → 输出解析(Parse)。本章重点讲解前两步（Prompts + Models），输出解析在第6节单独讲解。

v1.3.7 中 Models 模块提供了以下核心能力：

能力	说明
统一模型初始化	init_chat_model() 一个函数接入所有提供商
结构化输出	Pydantic / TypedDict / JSON Schema 三种方案
多模态	图片、音频、视频输入输出
模型能力探查	model.profile 动态获取模型支持的功能
速率限制	InMemoryRateLimiter 控制请求频率
提示缓存	降低重复前缀的 token 消耗
流式处理	stream() / astream_events() 实时输出

LangChain 本身不提供大模型，而是作为"胶水层"统一接入各平台的大模型——OpenAI、Anthropic、阿里千问、智谱GLM、DeepSeek 等。学会一个接口，所有模型触类旁通。

---

2. 模型的初始化与调用

2.1 init_chat_model：统一初始化入口

v1.3.7 推荐使用 init_chat_model() 统一初始化模型，无论后端是 OpenAI、Anthropic 还是其他提供商：

from langchain.chat_models import init_chat_model

# 方式1：自动根据环境变量推断提供商
model = init_chat_model("gpt-4o-mini")

# 方式2：显式指定提供商
model = init_chat_model("openai:gpt-4o-mini")
model = init_chat_model("anthropic:claude-sonnet-4-6")

# 方式3：传入额外参数
model = init_chat_model(
    "openai:gpt-4o-mini",
    temperature=0.7,
    max_tokens=1024,
    timeout=30,
    max_retries=3,
)

response = model.invoke("你好，请介绍一下自己")
print(response.content)

支持的提供商（部分）：

提供商	格式	安装命令
OpenAI	openai:gpt-4o-mini	pip install langchain-openai
Anthropic	anthropic:claude-sonnet-4-6	pip install langchain-anthropic
Google Gemini	google_genai:gemini-2.5-flash	pip install langchain-google-genai
Azure OpenAI	azure_openai:gpt-4o-mini	pip install langchain-openai
AWS Bedrock	aws:anthropic.claude-3-5-sonnet	pip install langchain-aws
本地 Ollama	ollama:deepseek-r1:7b	pip install langchain-ollama

init_chat_model() 是最推荐的方式。它屏蔽了不同提供商的初始化差异，让你用一个统一接口管理所有模型。

2.2 模型的三种分类

类型1：LLMs（非对话模型）

输入字符串，返回字符串。适用于单次文本生成任务。

from langchain_openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo-instruct")
result = llm.invoke("写一首关于春天的诗")
print(result)  # 直接返回字符串

主要特点：

• 输入：文本字符串
• 输出：文本字符串
• 适用：单次问答、摘要、翻译、代码生成
• 局限：不支持多轮对话

类型2：Chat Models（对话模型）— 最常用

输入消息列表，返回带角色的消息对象。开发首选。

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-4o-mini")

messages = [
    SystemMessage(content="我是人工智能助手，我叫小智"),
    HumanMessage(content="你好，我是小明，很高兴认识你")
]

response = model.invoke(messages)
print(type(response))  # <class 'langchain_core.messages.ai.AIMessage'>
print(response.content)

主要特点：

• 输入：消息列表 List[BaseMessage]，每条消息需指定角色
• 输出：带角色的消息对象（AIMessage）
• 原生支持多轮对话，通过消息列表维护上下文

类型3：Embedding Model（嵌入模型）

将文本转换为浮点数向量（Embedding），用于相似度搜索。在第07章 Retrieval 中重点讲解。

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector = embeddings.embed_query("人工智能改变了世界")
print(f"向量维度: {len(vector)}")  # 1536

2.3 参数配置方式

常用参数

参数	说明	建议值
model	模型名称	如 gpt-4o-mini
temperature	随机性，0=确定，1=创意	精确：≤0.5，平衡：0.7-0.8，创意：~1
max_tokens	最大输出长度	短回复：128-256，常规：512-1024，长文：1024-4096
timeout	超时时间（秒）	默认无限制
max_retries	最大重试次数	默认 6 次（网络错误自动重试）

推荐方式：使用 .env 配置文件

pip install python-dotenv

创建 .env 文件：

OPENAI_API_KEY="sk-xxxxxxxxx"
OPENAI_BASE_URL="https://api.openai-proxy.org/v1"

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()

model = init_chat_model(
    "openai:gpt-4o-mini",
    temperature=0.7,
    max_tokens=1024,
)

.env 文件配置密钥，加入 .gitignore，安全且方便团队协作。

2.4 各平台 API 调用举例

OpenAI

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o-mini")
response = model.invoke("请解释什么是机器学习")
print(response.content)

阿里云百炼（兼容 OpenAI 接口）

# .env
DASHSCOPE_API_KEY="sk-xxx"

import os
from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="deepseek-r1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
response = model.invoke("9.9和9.11谁大")
print(response.content)

百度千帆

import os
from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="ernie-4.0-turbo-8k",
    api_key=os.getenv("BAIDU_API_KEY"),
    base_url="https://qianfan.baidubce.com/v2",
    default_headers={"appid": "app-xxx"},
)

智谱 GLM

import os
from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="glm-4-flash",
    api_key=os.getenv("ZHIPUAI_API_KEY"),
    base_url="https://open.bigmodel.cn/api/paas/v4",
)

硅基流动（SiliconFlow）

import os
from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="deepseek-ai/DeepSeek-R1",
    api_key=os.getenv("SILICON_API_KEY"),
    base_url="https://api.siliconflow.cn/v1",
)

2.5 如何选择大模型

没有最好的大模型，只有最适合的。基础模型选型，合规和安全是首要考量因素。不同任务用不同模型，关注性价比而非排行榜。

本课程以 OpenAI 为例的原因：

1. OpenAI 最具代表性，其它模型都在追赶和模仿 OpenAI
2. 学会 OpenAI 接口，其余模型触类旁通
3. 通过 init_chat_model() 切换模型只需改一个参数

---

3. 消息与调用方法

3.1 消息类型

LangChain 提供了完整的消息类型体系：

类型	角色	用途
SystemMessage	system	设定 AI 行为规则和背景
HumanMessage	user	用户输入
AIMessage	assistant	AI 回复
ToolMessage	tool	工具调用结果返回

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="你是一个数学家，只会回答数学问题"),
    HumanMessage(content="1 + 2 * 3 = ?"),
    AIMessage(content="根据运算优先级，1 + 2 × 3 = 7"),
]

3.2 多轮对话

多轮对话的关键：每次调用时将历史消息追加到 messages 列表中。

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage

model = init_chat_model("openai:gpt-4o-mini")

# 第一轮
messages = [HumanMessage(content="我叫小明")]
response = model.invoke(messages)
messages.append(response)  # 把 AI 回复加入历史

# 第二轮
messages.append(HumanMessage(content="我叫什么名字？"))
response = model.invoke(messages)
print(response.content)  # 你叫小明

LangChain 本身不记忆对话历史，需要在每次调用时手动追加历史消息。

3.3 三种调用方式

invoke：单次调用

messages = [HumanMessage(content="介绍人工智能")]
response = model.invoke(messages)
print(response.content)

stream：流式输出

messages = [HumanMessage(content="写一篇关于春天的短文")]
for chunk in model.stream(messages):
    print(chunk.content, end="", flush=True)  # 逐字实时输出

高级流式：astream_events

async for event in model.astream_events(messages, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

astream_events 提供了更细粒度的流式事件控制，可以精确捕获每个阶段的事件。

batch：批量处理

questions = [
    [HumanMessage(content="什么是机器学习？")],
    [HumanMessage(content="什么是深度学习？")],
    [HumanMessage(content="什么是强化学习？")],
]

responses = model.batch(questions)
for r in responses:
    print(r.content[:50], "...")

batch 是客户端并行，多个请求同时发出，显著提升吞吐量。

3.4 同步与异步

import asyncio

# 同步
response = model.invoke(messages)

# 异步
response = await model.ainvoke(messages)

# 异步并发
results = await asyncio.gather(
    model.ainvoke(msg1),
    model.ainvoke(msg2),
    model.ainvoke(msg3),
)

---

4. Prompt 模板

Prompt Template 将变量插入到模板中，灵活构建提示词。

4.1 PromptTemplate

用于 LLMs（非对话模型）的字符串模板。

from langchain_core.prompts import PromptTemplate

# 方式1：构造方法
template = PromptTemplate(
    template="请用{style}风格介绍{topic}。",
    input_variables=["style", "topic"],
)
prompt = template.format(style="幽默", topic="人工智能")

# 方式2：from_template()
template = PromptTemplate.from_template("请给我一个关于{topic}的{type}解释。")
prompt = template.format(type="详细", topic="量子力学")

部分变量（Partial Variables）：提前固定某些变量值。

template = PromptTemplate.from_template("你是一个{role}，{question}")
partial_template = template.partial(role="资深厨师")
print(partial_template.format(question="如何煎牛排？"))
# 你是一个资深厨师，如何煎牛排？

4.2 ChatPromptTemplate

用于 Chat Models 的消息模板，支持多角色。

from langchain_core.prompts import ChatPromptTemplate

# 方式1：构造方法
prompt = ChatPromptTemplate([
    ("system", "你是一个{role}，名字叫{name}。"),
    ("human", "{user_input}"),
])

messages = prompt.invoke({"role": "AI工程师", "name": "小智", "user_input": "你能做什么？"})

# 方式2：from_messages()
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI助手，名字是{name}。"),
    ("human", "你好，最近怎么样？"),
    ("ai", "我很好，谢谢！"),
    ("human", "{user_input}"),
])

推荐使用 format_messages()：

messages = prompt.format_messages(
    name="小智",
    user_input="你叫什么名字？"
)
# 返回 list[BaseMessage]，可直接传给 model.invoke()

4.3 插入消息列表：MessagesPlaceholder

当需要插入一组动态消息时使用，常用于多轮对话历史。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。"),
    MessagesPlaceholder("history"),          # 历史消息插入点
    ("human", "{question}"),
])

messages = prompt.format_messages(
    history=[
        HumanMessage(content="1+2*3等于几？"),
        AIMessage(content="等于7"),
    ],
    question="我刚才问了什么？",
)

4.4 FewShotPromptTemplate

通过少量示例教会模型如何回答。

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate

examples = [
    {"input": "北京天气", "output": "北京市"},
    {"input": "南京下雨吗", "output": "南京市"},
    {"input": "武汉热吗", "output": "武汉市"},
]

example_prompt = PromptTemplate.from_template("Input: {input}\nOutput: {output}")

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    suffix="Input: {input}\nOutput:",
    input_variables=["input"],
)

print(prompt.invoke({"input": "长沙多少度"}).to_string())

示例选择器：从大量示例中自动选取最相关的几个，减少 token 消耗。

from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    Chroma,
    k=1,
)
# 自动选最相似的那个示例

4.5 从文件加载 Prompt

便于版本管理和团队共享。

# prompt.yaml
_type: "prompt"
input_variables: ["name", "topic"]
template: "请给{name}讲一个关于{topic}的故事"

from langchain_core.prompts import load_prompt

prompt = load_prompt("prompt.yaml")
print(prompt.format(name="小朋友", topic="勇气"))

---

5. 结构化输出

这是 v1.3.7 的核心特性之一——让模型按照你定义的数据结构来输出，而不是自由文本。

5.1 Pydantic 方式（推荐）

最强大的方案，支持字段验证、嵌套结构。

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

# 1. 定义输出结构
class Person(BaseModel):
    """人物信息"""
    name: str = Field(description="人物的姓名")
    age: int = Field(description="人物的年龄")
    skills: list[str] = Field(description="人物的技能列表")

# 2. 初始化模型并绑定结构化输出
model = init_chat_model("openai:gpt-4o-mini")
structured_model = model.with_structured_output(Person)

# 3. 调用
result = structured_model.invoke("小明今年25岁，擅长Python、Java和Go语言")
print(f"姓名: {result.name}, 年龄: {result.age}, 技能: {result.skills}")
# 姓名: 小明, 年龄: 25, 技能: ['Python', 'Java', 'Go']
print(type(result))  # <class 'Person'>

with_structured_output() 让模型直接返回 Pydantic 对象，不再是字符串解析。

5.2 TypedDict 方式

更轻量，无运行时验证，适合简单场景。

from typing import TypedDict, Annotated

class Movie(TypedDict):
    """电影信息"""
    title: Annotated[str, ..., "电影名称"]
    year: Annotated[int, ..., "上映年份"]
    director: Annotated[str, ..., "导演"]

structured_model = model.with_structured_output(Movie)
result = structured_model.invoke("《肖申克的救赎》是哪年上映的，导演是谁？")
print(result)  # {'title': '肖申克的救赎', 'year': 1994, 'director': '弗兰克·德拉邦特'}

5.3 JSON Schema 方式

最大灵活性和跨语言互操作性。

json_schema = {
    "title": "Weather",
    "description": "天气信息",
    "type": "object",
    "properties": {
        "city": {"type": "string", "description": "城市名"},
        "temperature": {"type": "number", "description": "温度（摄氏度）"},
        "condition": {"type": "string", "description": "天气状况"},
    },
    "required": ["city", "temperature", "condition"],
}

structured_model = model.with_structured_output(json_schema, method="json_schema")
result = structured_model.invoke("北京今天天气怎么样？")
print(result)  # {'city': '北京', 'temperature': 28.0, 'condition': '晴'}

5.4 三种方案对比

方案	验证	类型安全	嵌套结构	跨语言	推荐场景
Pydantic	运行时	强	支持	否	首选，复杂业务逻辑
TypedDict	无	中等	支持	否	简单结构
JSON Schema	无	弱	支持	是	跨语言、API 规范

5.5 include_raw：同时获取原始消息

structured_model = model.with_structured_output(Person, include_raw=True)

raw_result = structured_model.invoke("小明25岁，擅长Python")
print(raw_result["parsed"].name)         # 小明
print(raw_result["parsed"].age)          # 25
print(raw_result["raw"].response_metadata)  # 含 token 数量等元数据

---

6. Output Parsers

输出解析器将模型输出转换为特定格式。

6.1 StrOutputParser：提取纯文本

最简单的解析器，从 AIMessage 中提取 content 字段。

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
result = parser.invoke(response)  # response 是 AIMessage
print(type(result))  # <class 'str'>

6.2 JsonOutputParser：JSON 解析

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

parser = JsonOutputParser()

prompt = PromptTemplate(
    template="回答用户查询。\n{format_instructions}\n{query}",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# LCEL 组合
chain = prompt | model | parser
result = chain.invoke({"query": "给我讲一个笑话"})
print(result)  # {'joke': '...'}

v1 推荐方式是用第5节的 with_structured_output() 替代 JsonOutputParser，类型更安全。

6.3 XMLOutputParser

from langchain_core.output_parsers import XMLOutputParser

parser = XMLOutputParser()
chain = prompt | model | parser
result = chain.invoke({"query": "生成周星驰的代表作列表"})
# 返回 dict 格式

6.4 CommaSeparatedListOutputParser

from langchain_core.output_parsers import CommaSeparatedListOutputParser

parser = CommaSeparatedListOutputParser()
prompt = PromptTemplate.from_template(
    "生成5个关于{topic}的关键词。\n{format_instructions}",
    partial_variables={"format_instructions": parser.get_format_instructions()},
)
chain = prompt | model | parser
result = chain.invoke({"topic": "人工智能"})
# ['机器学习', '深度学习', '自然语言处理', '计算机视觉', '强化学习']

6.5 DatetimeOutputParser

from langchain_core.output_parsers import DatetimeOutputParser

parser = DatetimeOutputParser()
chain = prompt | model | parser
result = chain.invoke({"query": "中华人民共和国什么时候成立的？"})
print(result)  # 1949-10-01 00:00:00
print(type(result))  # <class 'datetime.datetime'>

---

7. 高级特性

7.1 Model Profiles：模型能力探查

通过 model.profile 属性动态获取模型的能力信息，无需手动查文档。

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o-mini")
profile = model.profile

print(f"最大输入 token: {profile.get('max_input_tokens')}")
print(f"支持工具调用: {profile.get('tool_calling')}")
print(f"支持图片输入: {profile.get('image_inputs')}")

应用场景：

• 根据上下文窗口大小自动触发摘要策略
• 自动判断模型是否支持图片输入
• Deep Agents 根据 profile 自动选择合适的模型

7.2 Multimodal：多模态处理

模型可以直接处理图片、音频、视频内容。

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
import base64

# 读取图片并编码为 base64
with open("photo.jpg", "rb") as f:
    image_data = base64.b64encode(f.read()).decode()

model = init_chat_model("openai:gpt-4o-mini")

message = HumanMessage(content=[
    {"type": "text", "text": "这张图片里有什么？"},
    {"type": "image", "base64": image_data, "mime_type": "image/jpeg"},
])

response = model.invoke([message])
print(response.content)

并非所有模型都支持多模态。使用 model.profile 前置检查模型能力。

7.3 Prompt Caching：提示缓存

对于包含重复前缀的多次调用，可以利用缓存降低延迟和成本。

from langchain.chat_models import init_chat_model

# Anthropic 显式缓存
model = init_chat_model("anthropic:claude-sonnet-4-6")

# 系统提示词会被自动缓存
messages = [
    {"role": "system", "content": "你是一个知识渊博的助手..."},  # 缓存的断点
    {"role": "user", "content": "解释黑洞的物理学原理"},
]

OpenAI、Gemini 等提供商自动启用了隐式缓存，无需手动配置。

7.4 Rate Limiter：速率限制

控制请求频率，避免触发 API 频率限制。

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=2,       # 每秒最多2次请求
    check_every_n_seconds=0.1,   # 每0.1秒检查一次
    max_bucket_size=10,          # 突发峰值缓冲区
)

model = init_chat_model(
    "openai:gpt-4o-mini",
    rate_limiter=rate_limiter,
)

7.5 Token Usage 追踪

追踪每次调用的 token 消耗。

from langchain.chat_models import init_chat_model
from langchain_core.callbacks import get_usage_metadata_callback

model = init_chat_model("openai:gpt-4o-mini")

with get_usage_metadata_callback() as cb:
    response = model.invoke("解释人工智能")
    for usage in cb.usage_metadata.values():
        print(f"输入 token: {usage['input_tokens']}")
        print(f"输出 token: {usage['output_tokens']}")
        print(f"总计 token: {usage['total_tokens']}")

---

8. 调用本地模型（Ollama）

Ollama 让你在本地运行开源大模型，无需联网、保护数据隐私。

8.1 安装与下载

访问 ollama.com 下载安装。

# 下载模型
ollama pull deepseek-r1:7b
ollama pull qwen2.5:7b

8.2 通过 LangChain 调用

from langchain.chat_models import init_chat_model

# 方式1：init_chat_model
model = init_chat_model("ollama:deepseek-r1:7b")
response = model.invoke("你好，请介绍一下自己")
print(response.content)

# 方式2：直接使用 ChatOllama
from langchain_ollama import ChatOllama

model = ChatOllama(model="deepseek-r1:7b")
response = model.invoke("请解释什么是机器学习")

自定义 Ollama 地址：

model = init_chat_model(
    "ollama:deepseek-r1:7b",
    base_url="http://192.168.1.100:11434",  # 远程服务器
)

结合 Prompt 使用：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译助手，将{source_lang}翻译成{target_lang}。"),
    ("human", "{text}"),
])

messages = prompt.format_messages(
    source_lang="中文", target_lang="英语", text="我爱编程"
)

model = init_chat_model("ollama:deepseek-r1:7b")
response = model.invoke(messages)
print(response.content)  # I love programming.

---

本章小结

本章完整介绍了 LangChain v1.3.7 Models 模块：

环节	核心内容
初始化模型	init_chat_model() 统一入口，支持所有主流提供商
消息类型	SystemMessage / HumanMessage / AIMessage / ToolMessage
调用方式	invoke / stream / batch / ainvoke / astream
Prompt 模板	PromptTemplate / ChatPromptTemplate / FewShotPromptTemplate / MessagesPlaceholder
结构化输出	Pydantic / TypedDict / JSON Schema — with_structured_output()
Output Parser	StrOutputParser / JsonOutputParser / XMLOutputParser / List / Datetime
高级特性	Model Profiles / Multimodal / Prompt Caching / Rate Limiter / Token Usage
本地模型	Ollama 本地部署与调用

核心流程：PromptTemplate → Model (invoke/stream/batch) → Output Parser（或用 with_structured_output 一步到位）