Part 2: Models（模型）

糖果店的幽灵

248人浏览 · 2026-05-22 04:13:41

糖果店的幽灵 · 2026-05-22 04:13:41 发布

Part 2: Models（模型）

文章目录

- Part 2: Models（模型）

2.1 模型类型概览

LangChain 支持多种类型的模型，它们之间有明确的继承和分类关系：

Chat Models vs LLMs 对比

特性	Chat Models（聊天模型）	LLMs（传统语言模型）
输入格式	消息列表（System/Human/AI）	纯文本字符串
输出格式	`AIMessage` 对象	纯文本字符串
适用场景	对话、多轮交互、指令遵循	文本补全、续写
代表模型	GPT-4o, Claude 3.5, Gemini	text-davinci-003（已停用）
推荐程度	推荐使用	不推荐（已过时）
上下文管理	内置消息角色区分	需要手动管理

重要建议：在新项目中，始终优先使用 Chat Models。LLMs 接口已基本被淘汰。

2.2 Chat Models（聊天模型）

概念解释

Chat Models 是 LangChain 中最常用的模型类型。它模拟"聊天"的形式，通过不同角色的消息来与 LLM 交互。

类比理解：

System Message（系统消息）= 老板给的工作指令：“你是一个翻译专家”

Human Message（用户消息）= 客户的需求：“请把这段话翻译成英文”

AI Message（AI 消息）= 你的回复：“Here is the translation…”

所有初始化参数

以 ChatOpenAI 为例：

参数名	类型	默认值	说明
`model`	`str`	`"gpt-4o-mini"`	模型名称
`temperature`	`float`	`0.7`	控制随机性。0 = 确定性，2 = 最随机
`max_tokens`	`int`	`None`	单次回复的最大 Token 数量
`timeout`	`float`	`None`	API 请求超时时间（秒）
`max_retries`	`int`	`2`	失败后的最大重试次数
`api_key`	`str`	`None`	OpenAI API 密钥
`base_url`	`str`	`None`	自定义 API 端点
`streaming`	`bool`	`False`	是否使用流式传输
`seed`	`int`	`None`	随机种子（可复现输出）
`top_p`	`float`	`1.0`	核采样参数
`frequency_penalty`	`float`	`0.0`	频率惩罚（-2.0 到 2.0）
`presence_penalty`	`float`	`0.0`	存在惩罚（-2.0 到 2.0）
`n`	`int`	`1`	生成回复数量
`stop`	`list[str]`	`None`	停止序列

所有调用方法

1. `invoke` – 同步单次调用

# demo_invoke.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,
    max_tokens=1024,
)

# 方式一：直接传入字符串
result = model.invoke("用一句话解释什么是量子计算")
print(f"结果: {result.content}")
print(f"消息类型: {type(result)}")  # AIMessage

# 方式二：传入消息列表
from langchain_core.messages import HumanMessage, SystemMessage

messages = [
    SystemMessage(content="你是一个物理学教授"),
    HumanMessage(content="用一句话解释什么是量子计算"),
]
result = model.invoke(messages)
print(f"结果: {result.content}")

# AIMessage 对象的常用属性
print(f"content: {result.content}")
print(f"response_metadata: {result.response_metadata}")
print(f"id: {result.id}")
print(f"usage_metadata: {result.usage_metadata}")

2. `batch` – 同步批量调用

# demo_batch.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

requests = [
    "用一句话解释什么是机器学习",
    "用一句话解释什么是深度学习",
    "用一句话解释什么是自然语言处理",
]

results = model.batch(requests)

for i, result in enumerate(results):
    print(f"问题 {i+1}: {requests[i]}")
    print(f"回答: {result.content}\n")

3. `stream` – 同步流式调用

# demo_stream.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

print("模型回答（流式输出）：")
for chunk in model.stream("请写一首关于春天的四行诗"):
    print(chunk.content, end="", flush=True)
print()

4. `ainvoke` – 异步单次调用

# demo_ainvoke.py
import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

async def main():
    # 并发调用多个请求
    tasks = [
        model.ainvoke("什么是 Python？"),
        model.ainvoke("什么是 JavaScript？"),
        model.ainvoke("什么是 Rust？"),
    ]
    results = await asyncio.gather(*tasks)
    for i, result in enumerate(results):
        print(f"结果 {i+1}: {result.content}\n")

asyncio.run(main())

5. `abatch` – 异步批量调用

# demo_abatch.py
import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

async def main():
    requests = [
        "解释什么是 REST API",
        "解释什么是 GraphQL",
        "解释什么是 gRPC",
    ]
    results = await model.abatch(requests)
    for i, result in enumerate(results):
        print(f"问题 {i+1}: {requests[i]}")
        print(f"回答: {result.content}\n")

asyncio.run(main())

6. `astream` – 异步流式调用

# demo_astream.py
import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

async def main():
    print("异步流式输出：")
    async for chunk in model.astream("请写一首关于秋天的四行诗"):
        print(chunk.content, end="", flush=True)
    print()

asyncio.run(main())

消息类型详解

# demo_message_types.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    SystemMessage, HumanMessage, AIMessage,
    ToolMessage, AIMessageChunk,
)

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# SystemMessage：设定 AI 角色
system_msg = SystemMessage(content="你是一位资深 Python 开发工程师。")

# HumanMessage：用户输入
human_msg = HumanMessage(content="请解释装饰器是什么？")

# AIMessage：AI 回复（常用于对话历史）
ai_msg = AIMessage(content="装饰器是 Python 中的一种设计模式...")

# 组合使用：多轮对话
conversation = [
    SystemMessage(content="你是一个翻译助手，只做翻译。"),
    HumanMessage(content="翻译成英文：今天天气真好"),
    AIMessage(content="The weather is really nice today."),
    HumanMessage(content="翻译成英文：我想学习编程"),
]
result = model.invoke(conversation)
print(f"翻译结果: {result.content}")

# ToolMessage：工具调用结果（Agent 场景使用）
tool_msg = ToolMessage(
    content='{"temperature": 25, "weather": "sunny"}',
    tool_call_id="call_abc123",
    name="weather_tool",
)

# AIMessageChunk：流式消息块
chunk1 = AIMessageChunk(content="你好")
chunk2 = AIMessageChunk(content="，")
chunk3 = AIMessageChunk(content="世界！")
full_message = chunk1 + chunk2 + chunk3
print(f"合并后: {full_message.content}")  # "你好，世界！"

多轮对话 Demo

# demo_multi_turn.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

messages = [SystemMessage(content="你是一个友好的中文老师。")]

# 第一轮
messages.append(HumanMessage(content="你好，我想学中文。"))
response = model.invoke(messages)
print(f"AI: {response.content}")
messages.append(response)

# 第二轮
messages.append(HumanMessage(content="'苹果'用中文怎么说？"))
response = model.invoke(messages)
print(f"AI: {response.content}")
messages.append(response)

# 第三轮（AI 能记住上下文）
messages.append(HumanMessage(content="我刚才问了什么？"))
response = model.invoke(messages)
print(f"AI: {response.content}")

配置运行时参数 Demo

# demo_runtime_config.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# bind：临时绑定参数（不修改原始模型）
creative_model = model.bind(temperature=1.5)
precise_model = model.bind(temperature=0.0)

result1 = model.invoke("描述未来城市")
result2 = creative_model.invoke("描述未来城市")
result3 = precise_model.invoke("描述未来城市")

# bind_tools：绑定工具
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取天气信息",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string", "description": "城市名称"}},
            "required": ["city"],
        },
    },
}]
model_with_tools = model.bind_tools(tools)
result = model_with_tools.invoke("北京今天天气怎么样？")
print(f"工具调用: {result.tool_calls}")

2.3 LLMs（传统语言模型）

重要：LLMs 接口已不再推荐使用。所有新项目请使用 Chat Models。

# 在 LangChain 1.x 中，ChatOpenAI 也能接受字符串输入
from dotenv import load_dotenv
load_dotenv()
from langchain_openai import ChatOpenAI

chat_model = ChatOpenAI(model="gpt-4o-mini")
result = chat_model.invoke("用一句话解释什么是 Python")
print(result.content)

2.4 多模型支持

init_chat_model 统一接口

# demo_init_chat_model.py
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

# 通过字符串指定模型提供商和名称
model = init_chat_model("gpt-4o-mini", model_provider="openai", temperature=0.7)
result = model.invoke("你好")
print(f"OpenAI: {result.content}")

# 动态切换模型
import os
provider = os.getenv("MODEL_PROVIDER", "openai")
model_name = os.getenv("MODEL_NAME", "gpt-4o-mini")
model = init_chat_model(model_name, model_provider=provider, temperature=0.7)

各提供商对比表格

提供商	模型名称	安装命令	特点
OpenAI	GPT-4o, GPT-4o-mini	`pip install langchain-openai`	最成熟，生态最好
Anthropic	Claude 3.5 Sonnet	`pip install langchain-anthropic`	安全性好，长文本能力强
Google	Gemini Pro	`pip install langchain-google-genai`	多模态，Google 生态
Ollama	Llama 3, Mistral	`pip install langchain-ollama`	本地运行，免费，隐私

2.5 嵌入模型（Embeddings）

概念解释

嵌入（Embedding）是将文本转换为高维向量的技术。语义相似的文本，其向量也相似。

类比：想象一个"语义地图"，"猫"和"狗"距离很近（都是宠物），"猫"和"汽车"距离很远。

# demo_embeddings.py
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import OpenAIEmbeddings
import numpy as np

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 嵌入单个文本
vector = embeddings.embed_query("LangChain 是一个强大的 AI 框架")
print(f"向量维度: {len(vector)}")  # 1536
print(f"前5个值: {vector[:5]}")

# 嵌入多个文本
vectors = embeddings.embed_documents(["我喜欢吃苹果", "今天天气很好"])
print(f"嵌入了 {len(vectors)} 个文本")

# 计算文本相似度
def cosine_similarity(vec1, vec2):
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

vec_a = embeddings.embed_query("猫是一种可爱的宠物")
vec_b = embeddings.embed_query("狗是人类的好朋友")
vec_c = embeddings.embed_query("量子力学是物理学分支")

print(f"猫 vs 狗: {cosine_similarity(vec_a, vec_b):.4f}")  # 较高
print(f"猫 vs 量子: {cosine_similarity(vec_a, vec_c):.4f}")  # 较低

# 自定义维度（text-embedding-3 系列支持）
small_emb = OpenAIEmbeddings(model="text-embedding-3-small", dimensions=512)
vec = small_emb.embed_query("测试")
print(f"自定义维度: {len(vec)}")  # 512

嵌入使用场景

场景	说明
语义搜索	根据含义搜索，非关键词匹配
文档问答（RAG）	找到最相关的文档片段
文本聚类	将语义相似的文本分组
推荐系统	推荐与用户兴趣相似的内容

2.6 模型最佳实践

模型选择建议

场景	推荐模型	原因
简单对话/分类	GPT-4o-mini	便宜、快速
复杂推理/分析	GPT-4o / Claude 3.5	推理能力强
长文本处理	Claude 3.5	200K 上下文
代码生成	GPT-4o	代码能力最强
本地部署/隐私	Ollama (Llama 3)	免费、不出本地

成本控制

# best_practices_cost_control.py
from dotenv import load_dotenv
load_dotenv()
from langchain_openai import ChatOpenAI
from langchain_core.caches import InMemoryCache
from langchain.globals import set_llm_cache

# 策略一：使用更便宜的模型
cheap_model = ChatOpenAI(model="gpt-4o-mini")

# 策略二：限制输出长度
model = ChatOpenAI(model="gpt-4o-mini", max_tokens=500)

# 策略三：缓存重复请求
set_llm_cache(InMemoryCache())
model = ChatOpenAI(model="gpt-4o-mini")
result1 = model.invoke("什么是 Python？")  # 调用 API
result2 = model.invoke("什么是 Python？")  # 从缓存读取

# 策略四：监控 Token 用量
result = model.invoke("写一首短诗")
if hasattr(result, "usage_metadata"):
    usage = result.usage_metadata
    print(f"输入 Tokens: {usage.get('input_tokens')}")
    print(f"输出 Tokens: {usage.get('output_tokens')}")

错误处理

# best_practices_error_handling.py
import time
from dotenv import load_dotenv
load_dotenv()
from langchain_openai import ChatOpenAI

def safe_invoke(model, prompt: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            result = model.invoke(prompt)
            return result.content
        except Exception as e:
            print(f"第 {attempt + 1} 次失败: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # 指数退避
            else:
                raise
    return ""

model = ChatOpenAI(model="gpt-4o-mini", max_retries=3, timeout=30)
result = safe_invoke(model, "你好")
print(result)


---





##