以观书法
108.85M · 2026-02-05
Large Language Models (LLMs) 是强大的 AI 工具,能够像人类一样理解和生成文本。它们足够versatile(多才多艺),可以编写内容、翻译语言、总结文本和回答问题,而无需针对每个任务进行专门训练。
现代 LLM 除了文本生成外,还支持:
| 能力 | 描述 | 应用场景 |
|---|---|---|
| 工具调用 | 调用外部工具(如数据库查询或API调用)并在响应中使用结果 | 数据查询、API集成 |
| 结构化输出 | 模型响应遵循定义的格式 | 数据提取、表单填充 |
| ️ 多模态 | 处理和返回文本以外的数据(图像、音频、视频) | 图像识别、语音处理 |
| 推理 | 执行多步推理得出结论 | 复杂问题解决 |
graph TB
A[用户输入] --> B[LangChain Models]
B --> C[文本生成]
B --> D[工具调用]
B --> E[结构化输出]
B --> F[多模态处理]
B --> G[推理分析]
C --> H[最终响应]
D --> H
E --> H
F --> H
G --> H
LangChain 提供了统一的 init_chat_model 接口,支持多个主流模型提供商:
import os
from langchain.chat_models import init_chat_model
# 设置 API Key
os.environ["OPENAI_API_KEY"] = "sk-..."
# 初始化模型
model = init_chat_model("gpt-4o")
# 简单调用
response = model.invoke("为什么鹦鹉会说话?")
print(response.content)
import os
from langchain.chat_models import init_chat_model
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
model = init_chat_model("claude-sonnet-4-5-20250929")
response = model.invoke("解释量子计算的基本原理")
print(response.content)
import os
from langchain.chat_models import init_chat_model
os.environ["GOOGLE_API_KEY"] = "..."
model = init_chat_model("google_genai:gemini-2.5-flash-lite")
response = model.invoke("总结人工智能的发展历史")
print(response.content)
model = init_chat_model(
"claude-sonnet-4-5-20250929",
# 模型参数配置
temperature=0.7, # 控制输出的随机性 (0-1)
timeout=30, # 超时时间(秒)
max_tokens=1000, # 最大输出token数
)
response = model.invoke("写一首关于春天的诗")
参数说明:
| 参数 | 类型 | 说明 |
|---|---|---|
model | string | 模型名称或标识符 |
api_key | string | API密钥(通常通过环境变量设置) |
temperature | number | 控制输出随机性,越高越有创意 |
max_tokens | number | 限制输出的最大token数 |
timeout | number | 请求超时时间(秒) |
max_retries | number | 请求失败时的最大重试次数 |
最直接的调用方式,等待完整响应后返回:
# 单条消息
response = model.invoke("鹦鹉为什么有彩色的羽毛?")
print(response.content)
# 对话历史
conversation = [
{"role": "system", "content": "你是一个翻译助手,将英文翻译成法语。"},
{"role": "user", "content": "翻译: I love programming."},
{"role": "assistant", "content": "J'adore la programmation."},
{"role": "user", "content": "翻译: I love building applications."}
]
response = model.invoke(conversation)
print(response.content)
使用消息对象格式:
from langchain.messages import HumanMessage, AIMessage, SystemMessage
conversation = [
SystemMessage("你是一个专业的Python编程助手。"),
HumanMessage("如何在Python中创建装饰器?"),
]
response = model.invoke(conversation)
print(response.content)
实时显示生成的内容,改善用户体验:
# 基础文本流式输出
print("AI: ", end="")
for chunk in model.stream("为什么鹦鹉有彩色的羽毛?"):
print(chunk.content, end="", flush=True)
print()
# 流式输出并构建完整消息
full_message = None
for chunk in model.stream("天空是什么颜色?"):
full_message = chunk if full_message is None else full_message + chunk
print(f"当前累积: {full_message.content}")
print(f"n完整消息: {full_message.content}")
流式处理工具调用和推理:
for chunk in model.stream("天空是什么颜色?"):
for block in chunk.content_blocks:
if block["type"] == "reasoning":
print(f"推理: {block.get('reasoning')}")
elif block["type"] == "tool_call_chunk":
print(f"工具调用: {block}")
elif block["type"] == "text":
print(block["text"], end="")
并行处理多个独立请求,提高性能和降低成本:
# 批量调用
questions = [
"为什么鹦鹉有彩色的羽毛?",
"飞机是如何飞行的?",
"什么是量子计算?"
]
responses = model.batch(questions)
for question, response in zip(questions, responses):
print(f"问题: {question}")
print(f"回答: {response.content}n")
批量流式处理(按完成顺序返回):
for response in model.batch_as_completed(questions):
index = response.index # 原始输入的索引
result = response.result
print(f"问题 {index}: {questions[index]}")
print(f"回答: {result.content}n")
控制并发数:
responses = model.batch(
questions,
config={
'max_concurrency': 5, # 限制为5个并行调用
}
)
工具调用是 LLM 最强大的功能之一,允许模型调用外部工具来执行特定任务。
sequenceDiagram
participant U as 用户
participant M as 模型
participant T as 工具
U->>M: "旧金山和纽约的天气如何?"
M->>M: 分析请求并决定需要的工具
par 并行工具调用
M->>T: get_weather("San Francisco")
M->>T: get_weather("New York")
end
par 工具执行
T-->>M: 旧金山天气数据
T-->>M: 纽约天气数据
end
M->>M: 处理结果并生成响应
M->>U: "旧金山: 72°F晴天, 纽约: 68°F多云"
from langchain.tools import tool
# 定义工具
@tool
def get_weather(location: str) -> str:
"""获取指定地点的天气信息。
Args:
location: 城市和州,例如 San Francisco, CA
"""
# 实际应用中这里会调用天气API
return f"{location}当前天气晴朗,温度22°C。"
@tool
def get_population(location: str) -> int:
"""获取指定地点的人口数量。
Args:
location: 城市和州,例如 San Francisco, CA
"""
# 实际应用中这里会查询数据库
populations = {
"San Francisco, CA": 873965,
"New York, NY": 8336817,
"Los Angeles, CA": 3979576
}
return populations.get(location, 0)
# 绑定工具到模型
model_with_tools = model.bind_tools([get_weather, get_population])
# 调用模型
response = model_with_tools.invoke("波士顿的天气如何?")
# 查看模型生成的工具调用
for tool_call in response.tool_calls:
print(f"工具: {tool_call['name']}")
print(f"参数: {tool_call['args']}")
print(f"ID: {tool_call['id']}")
from langchain.tools import tool
@tool
def calculator(expression: str) -> float:
"""计算数学表达式。
Args:
expression: 数学表达式,如 "2 + 2" 或 "10 * 5"
"""
try:
return eval(expression)
except:
return "计算错误"
@tool
def search_database(query: str) -> str:
"""在数据库中搜索信息。
Args:
query: 搜索查询
"""
# 模拟数据库查询
return f"找到关于'{query}'的相关信息"
# 绑定工具
model_with_tools = model.bind_tools([calculator, search_database])
# 步骤1: 模型生成工具调用
messages = [{"role": "user", "content": "计算 (15 + 25) * 2 的结果"}]
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)
print(f"模型请求调用工具: {ai_msg.tool_calls}")
# 步骤2: 执行工具并收集结果
for tool_call in ai_msg.tool_calls:
if tool_call['name'] == 'calculator':
result = calculator.invoke(tool_call)
messages.append(result)
print(f"工具执行结果: {result.content}")
# 步骤3: 将结果传回模型生成最终响应
final_response = model_with_tools.invoke(messages)
print(f"最终回答: {final_response.content}")
# 强制使用任意工具
model_with_tools = model.bind_tools(
[get_weather, get_population],
tool_choice="any"
)
# 强制使用特定工具
model_with_tools = model.bind_tools(
[get_weather, get_population],
tool_choice="get_weather"
)
response = model_with_tools.invoke("今天天气怎么样?")
# 模型可以同时调用多个工具
model_with_tools = model.bind_tools([get_weather, get_population])
response = model_with_tools.invoke(
"比较波士顿和东京的天气,并告诉我哪个城市人口更多"
)
print("工具调用列表:")
for tool_call in response.tool_calls:
print(f"- {tool_call['name']}({tool_call['args']})")
# 输出可能是:
# - get_weather({'location': 'Boston'})
# - get_weather({'location': 'Tokyo'})
# - get_population({'location': 'Boston'})
# - get_population({'location': 'Tokyo'})
model_with_tools = model.bind_tools([get_weather])
print("流式接收工具调用:")
for chunk in model_with_tools.stream("北京和上海的天气如何?"):
for tool_chunk in chunk.tool_call_chunks:
if name := tool_chunk.get("name"):
print(f"n工具: {name}")
if args := tool_chunk.get("args"):
print(f"参数片段: {args}", end="")
结构化输出确保模型的响应遵循预定义的格式,这对于数据提取和后续处理非常有用。
from pydantic import BaseModel, Field
from typing import List
class Actor(BaseModel):
"""演员信息"""
name: str = Field(..., description="演员姓名")
role: str = Field(..., description="角色名称")
class Movie(BaseModel):
"""电影详细信息"""
title: str = Field(..., description="电影标题")
year: int = Field(..., description="上映年份")
director: str = Field(..., description="导演姓名")
rating: float = Field(..., description="评分(满分10分)")
cast: List[Actor] = Field(..., description="主要演员列表")
genres: List[str] = Field(..., description="电影类型")
budget: float | None = Field(None, description="预算(百万美元)")
# 绑定结构化输出
model_with_structure = model.with_structured_output(Movie)
# 调用模型
response = model_with_structure.invoke(
"提供电影《盗梦空间》的详细信息"
)
# 输出是强类型的 Pydantic 对象
print(f"标题: {response.title}")
print(f"年份: {response.year}")
print(f"导演: {response.director}")
print(f"评分: {response.rating}")
print(f"主演:")
for actor in response.cast:
print(f" - {actor.name} 饰演 {actor.role}")
from typing_extensions import TypedDict, Annotated
class MovieDict(TypedDict):
"""电影信息字典"""
title: Annotated[str, ..., "电影标题"]
year: Annotated[int, ..., "上映年份"]
director: Annotated[str, ..., "导演姓名"]
rating: Annotated[float, ..., "评分(满分10分)"]
model_with_structure = model.with_structured_output(MovieDict)
response = model_with_structure.invoke("介绍电影《星际穿越》")
# 输出是字典
print(response)
# {'title': '星际穿越', 'year': 2014, 'director': '克里斯托弗·诺兰', 'rating': 8.6}
json_schema = {
"title": "Product",
"description": "产品信息",
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "产品名称"
},
"price": {
"type": "number",
"description": "价格"
},
"in_stock": {
"type": "boolean",
"description": "是否有货"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "产品标签"
}
},
"required": ["name", "price", "in_stock"]
}
model_with_structure = model.with_structured_output(
json_schema,
method="json_schema",
)
response = model_with_structure.invoke(
"提供iPhone 15 Pro的产品信息"
)
print(response)
from pydantic import BaseModel, Field
class BookInfo(BaseModel):
"""书籍信息"""
title: str = Field(..., description="书名")
author: str = Field(..., description="作者")
year: int = Field(..., description="出版年份")
# 设置 include_raw=True 以获取原始消息
model_with_structure = model.with_structured_output(
BookInfo,
include_raw=True
)
response = model_with_structure.invoke("介绍《三体》这本书")
# 响应包含三个字段
print(f"解析结果: {response['parsed']}")
print(f"原始消息: {response['raw']}")
print(f"解析错误: {response['parsing_error']}")
# 访问token使用情况
print(f"使用的tokens: {response['raw'].usage_metadata}")
某些模型可以处理图像、音频和视频等非文本数据。
from langchain.chat_models import init_chat_model
# 初始化支持多模态的模型
model = init_chat_model("gpt-4o")
# 发送图像
message = {
"role": "user",
"content": [
{"type": "text", "text": "这张图片中有什么?"},
{
"type": "image_url",
"image_url": {
"url": "https://example.com/image.jpg"
}
}
]
}
response = model.invoke([message])
print(response.content)
# 处理Base64编码的图像
import base64
with open("local_image.jpg", "rb") as image_file:
base64_image = base64.b64encode(image_file.read()).decode()
message = {
"role": "user",
"content": [
{"type": "text", "text": "描述这张图片"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
response = model.invoke([message])
# 流式显示推理过程
for chunk in model.stream("解释为什么鹦鹉有彩色的羽毛?"):
reasoning_steps = [
r for r in chunk.content_blocks
if r["type"] == "reasoning"
]
if reasoning_steps:
for step in reasoning_steps:
print(f"推理: {step['reasoning']}")
else:
print(chunk.content, end="")
# 获取完整推理过程
response = model.invoke("为什么鹦鹉有彩色的羽毛?")
reasoning_steps = [
b for b in response.content_blocks
if b["type"] == "reasoning"
]
full_reasoning = " ".join(
step["reasoning"] for step in reasoning_steps
)
print(f"完整推理过程: {full_reasoning}")
from langchain.chat_models import init_chat_model
from langchain_core.callbacks import get_usage_metadata_callback
model_1 = init_chat_model(model="gpt-4o-mini")
model_2 = init_chat_model(model="claude-haiku-4-5-20251001")
# 使用上下文管理器追踪token使用
with get_usage_metadata_callback() as cb:
model_1.invoke("你好")
model_2.invoke("你好")
print(cb.usage_metadata)
# 输出:
# {
# 'gpt-4o-mini-2024-07-18': {
# 'input_tokens': 8,
# 'output_tokens': 10,
# 'total_tokens': 18
# },
# 'claude-haiku-4-5-20251001': {
# 'input_tokens': 8,
# 'output_tokens': 21,
# 'total_tokens': 29
# }
# }
from langchain.chat_models import init_chat_model
# 创建运行时可配置的模型
configurable_model = init_chat_model(temperature=0)
# 使用不同的模型运行
configurable_model.invoke(
"你的名字是什么?",
config={"configurable": {"model": "gpt-4o-mini"}},
)
configurable_model.invoke(
"你的名字是什么?",
config={"configurable": {"model": "claude-sonnet-4-5-20250929"}},
)
# 配置默认值和可配置字段
model = init_chat_model(
model="gpt-4o-mini",
temperature=0,
configurable_fields=(
"model",
"model_provider",
"temperature",
"max_tokens"
),
config_prefix="first",
)
# 运行时覆盖配置
model.invoke(
"你好",
config={
"configurable": {
"first_model": "claude-sonnet-4-5-20250929",
"first_temperature": 0.5,
"first_max_tokens": 100,
}
},
)
from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain.chat_models import init_chat_model
# 创建速率限制器
rate_limiter = InMemoryRateLimiter(
requests_per_second=0.1, # 每10秒1个请求
check_every_n_seconds=0.1, # 每100ms检查一次
max_bucket_size=10, # 控制最大突发大小
)
# 应用速率限制
model = init_chat_model(
model="gpt-4o",
model_provider="openai",
rate_limiter=rate_limiter
)
# 批量请求会被限速
responses = model.batch([
"问题1",
"问题2",
"问题3",
])
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from pydantic import BaseModel, Field
from typing import List
# 定义工具
@tool
def query_order_status(order_id: str) -> str:
"""查询订单状态
Args:
order_id: 订单编号
"""
# 模拟查询订单
orders = {
"ORD001": "已发货,预计明天送达",
"ORD002": "正在处理中",
"ORD003": "已送达"
}
return orders.get(order_id, "未找到该订单")
@tool
def check_product_stock(product_name: str) -> dict:
"""查询产品库存
Args:
product_name: 产品名称
"""
stock = {
"iPhone 15": {"available": True, "quantity": 50},
"MacBook Pro": {"available": True, "quantity": 30},
"AirPods": {"available": False, "quantity": 0}
}
return stock.get(product_name, {"available": False, "quantity": 0})
# 初始化模型
model = init_chat_model("gpt-4o-mini", temperature=0.3)
model_with_tools = model.bind_tools([
query_order_status,
check_product_stock
])
# 客服对话
def customer_service_chat(user_message: str):
"""处理客户咨询"""
messages = [
{
"role": "system",
"content": "你是一个专业的客服助手,帮助用户查询订单和产品信息。"
},
{"role": "user", "content": user_message}
]
# 模型生成工具调用
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)
# 执行工具
if ai_msg.tool_calls:
for tool_call in ai_msg.tool_calls:
if tool_call['name'] == 'query_order_status':
result = query_order_status.invoke(tool_call)
elif tool_call['name'] == 'check_product_stock':
result = check_product_stock.invoke(tool_call)
messages.append(result)
# 获取最终响应
final_response = model_with_tools.invoke(messages)
return final_response.content
return ai_msg.content
# 测试
print(customer_service_chat("我的订单ORD001什么时候能送到?"))
print(customer_service_chat("iPhone 15还有货吗?"))
from pydantic import BaseModel, Field
from typing import List
from langchain.chat_models import init_chat_model
# 定义数据结构
class ContactInfo(BaseModel):
"""联系信息"""
email: str | None = Field(None, description="邮箱地址")
phone: str | None = Field(None, description="电话号码")
address: str | None = Field(None, description="地址")
class Company(BaseModel):
"""公司信息"""
name: str = Field(..., description="公司名称")
industry: str = Field(..., description="行业")
founded_year: int | None = Field(None, description="成立年份")
employees: int | None = Field(None, description="员工数量")
contact: ContactInfo = Field(..., description="联系方式")
# 初始化模型
model = init_chat_model("gpt-4o")
extractor = model.with_structured_output(Company)
# 待提取的文本
text = """
科技创新公司是一家成立于2015年的人工智能公司。
公司目前有超过500名员工,专注于开发企业级AI解决方案。
联系方式: info@techinnov.com, 电话: 400-888-8888
总部位于北京市海淀区中关村大街1号
"""
# 提取结构化数据
company_info = extractor.invoke(f"从以下文本中提取公司信息:n{text}")
print(f"公司名称: {company_info.name}")
print(f"行业: {company_info.industry}")
print(f"成立年份: {company_info.founded_year}")
print(f"员工数: {company_info.employees}")
print(f"邮箱: {company_info.contact.email}")
print(f"电话: {company_info.contact.phone}")
print(f"地址: {company_info.contact.address}")
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from typing import List
@tool
def search_web(query: str) -> str:
"""在网络上搜索信息
Args:
query: 搜索查询
"""
# 模拟网络搜索
return f"关于'{query}'的搜索结果: [相关信息...]"
@tool
def summarize_text(text: str) -> str:
"""总结文本内容
Args:
text: 要总结的文本
"""
# 模拟文本总结
return f"总结: {text[:100]}..."
@tool
def save_to_database(data: str) -> str:
"""保存数据到数据库
Args:
data: 要保存的数据
"""
# 模拟数据库保存
return f"数据已保存: {data[:50]}..."
# 初始化模型
model = init_chat_model("gpt-4o", temperature=0)
model_with_tools = model.bind_tools([
search_web,
summarize_text,
save_to_database
])
def execute_complex_task(task_description: str, max_iterations: int = 5):
"""执行复杂的多步骤任务"""
messages = [
{
"role": "system",
"content": "你是一个任务执行助手,可以搜索信息、总结内容和保存数据。"
},
{"role": "user", "content": task_description}
]
iteration = 0
while iteration < max_iterations:
print(f"n--- 迭代 {iteration + 1} ---")
# 模型决定下一步操作
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)
if not ai_msg.tool_calls:
# 没有工具调用,任务完成
print(f"任务完成: {ai_msg.content}")
return ai_msg.content
# 执行所有工具调用
for tool_call in ai_msg.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
print(f"调用工具: {tool_name}({tool_args})")
if tool_name == 'search_web':
result = search_web.invoke(tool_call)
elif tool_name == 'summarize_text':
result = summarize_text.invoke(tool_call)
elif tool_name == 'save_to_database':
result = save_to_database.invoke(tool_call)
print(f"工具结果: {result.content}")
messages.append(result)
iteration += 1
return "达到最大迭代次数"
# 执行复杂任务
task = """
请帮我完成以下任务:
1. 搜索"人工智能最新发展趋势"
2. 总结搜索结果
3. 将总结保存到数据库
"""
result = execute_complex_task(task)
from langchain.chat_models import init_chat_model
import base64
from pathlib import Path
def analyze_image(image_path: str, question: str):
"""分析图像并回答问题"""
# 读取图像并转换为base64
with open(image_path, "rb") as image_file:
base64_image = base64.b64encode(image_file.read()).decode()
# 初始化支持视觉的模型
model = init_chat_model("gpt-4o")
# 构建消息
message = {
"role": "user",
"content": [
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
# 调用模型
response = model.invoke([message])
return response.content
# 使用示例
# result = analyze_image("product.jpg", "这个产品有什么特点?")
# print(result)
# 快速任务使用较小的模型
quick_model = init_chat_model("gpt-4o-mini", temperature=0.3)
# 复杂任务使用更强大的模型
advanced_model = init_chat_model("gpt-4o", temperature=0.7)
# 根据任务类型选择
if task_complexity == "simple":
model = quick_model
else:
model = advanced_model
# 需要确定性输出(如数据提取)
deterministic_model = init_chat_model("gpt-4o", temperature=0)
# 需要创造性输出(如写作)
creative_model = init_chat_model("gpt-4o", temperature=0.9)
# 平衡方案
balanced_model = init_chat_model("gpt-4o", temperature=0.5)
from langchain.chat_models import init_chat_model
model = init_chat_model("gpt-4o", max_retries=3, timeout=30)
try:
response = model.invoke("你的问题")
print(response.content)
except TimeoutError:
print("请求超时,请稍后重试")
except Exception as e:
print(f"发生错误: {e}")
# 使用批处理减少API调用
questions = ["问题1", "问题2", "问题3"]
responses = model.batch(questions, config={'max_concurrency': 3})
# 使用缓存(如果提供商支持)
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
set_llm_cache(InMemoryCache())
# 重复查询会从缓存返回
response1 = model.invoke("什么是AI?") # API调用
response2 = model.invoke("什么是AI?") # 从缓存返回
LangChain Models 提供了强大而灵活的接口来使用各种大语言模型:
统一接口 - 支持多个模型提供商(OpenAI、Anthropic、Google等)
多种调用方式 - Invoke、Stream、Batch满足不同场景需求
工具调用 - 让模型能够调用外部工具和API
结构化输出 - 确保输出格式符合预期
高级特性 - 多模态、推理、速率限制等
易于集成 - 与LangChain生态系统无缝集成
通过本文的学习,你应该能够: