接入模型
以 DeepSeek 为例：
1
2
3
4
5
6
7
8
9
10
11
|
from langchain_deepseek import ChatDeepSeek
model = ChatDeepSeek(
model="deepseek-chat",
api_key='xxx',
base_url='https://api.deepseek.com'
)
query = "你好，请你介绍一下你自己。"
response = model.invoke(query)
print(response.content)
|
Message 消息格式
在 LangChain 1.0 中，Message 是模型交互的最基本单元。它既代表模型接收到的输入，也代表模型生成的输出。
| 字段 |
说明 |
示例 |
| Role（角色） |
消息类型：system（系统提示）、user（用户输入）、assistant（模型回复） |
"role": "user" |
| Content（内容） |
消息的实际内容，可包含文本、图像、音频等多模态数据 |
"content": "你好，请帮我总结" |
| Metadata（元数据） |
可选字段，存储消息 ID、响应时间、token 消耗量等 |
"metadata": {"message_id": "abc123"} |
LangChain 1.0 提供了跨模型统一的 Message 标准，无论使用的是 OpenAI、Gemini 还是本地模型，都能保持一致的行为。
模型角色定义
大模型中通常由 system、user 和 assistant 三种角色组成消息列表：
1
2
3
4
5
6
7
8
|
from langchain.messages import HumanMessage, AIMessage, SystemMessage
conversation = [
SystemMessage("你是一个乐于助人的助手，负责将英文翻译成中文。"),
HumanMessage("请翻译，I love programming."),
AIMessage("我喜欢编程"),
HumanMessage("请翻译，I love swimming."),
]
|
stream 流式输出
1
2
3
4
5
6
7
|
query = '2 + 4 等于几'
full = None
for chunk in model.stream(query):
full = chunk if full is None else full + chunk
print(full.text)
print(full.content_blocks)
|
batch 批处理
1
2
3
4
5
6
7
8
9
10
|
query_list = [
"请介绍你自己",
"太阳从哪边升起",
"动物园里有什么"
]
responses = model.batch(query_list)
for response in responses:
print(response, end='\n\n')
|
控制并发数：
1
2
3
4
|
responses = model.batch(
query_list,
config={'max_concurrency': 2}, # 最多并行 2 个请求
)
|
流式批处理（每个任务完成后立即获取结果）：
1
2
|
for response in model.batch_as_completed(query_list):
print(response, end='\n\n')
|
模型结构化输出
LangChain 1.0 为所有模型都设置了结构化输出方法，可以让模型的回复严格匹配给定的 Schema，便于解析与后续处理。主要支持 Pydantic、TypedDict 和 JSON Schema 三种方法。
Pydantic
Pydantic 提供字段校验、默认值、描述信息、多层嵌套等高级能力，是生产场景首选：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
from pydantic import BaseModel, Field
class Movie(BaseModel):
"""A movie with details."""
title: str = Field(..., description="电影名称")
year: int = Field(..., description="电影发布年份")
director: str = Field(..., description="电影的导演")
model_with_structure = model.with_structured_output(Movie)
response = model_with_structure.invoke("请介绍电影《泰坦尼克号》")
print(response.title)
print(response.year)
print(response.director)
|
TypedDict
TypedDict 是更轻量的类型约束：
1
2
3
4
5
6
7
8
9
10
11
|
from typing_extensions import TypedDict, Annotated
class MovieDict(TypedDict):
"""A movie with details."""
title: Annotated[str, ..., "电影名称"]
year: Annotated[int, ..., "电影发布年份"]
director: Annotated[str, ..., "电影的导演"]
model_with_structure = model.with_structured_output(MovieDict)
response = model_with_structure.invoke("请介绍电影《泰坦尼克号》")
print(response)
|
JSON Schema
JSON Schema 在前后端/跨语言接口中最常用：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
json_schema = {
"title": "Movie",
"description": "A movie with details",
"type": "object",
"properties": {
"title": {"type": "string", "description": "电影名称"},
"year": {"type": "integer", "description": "电影发布年份"},
"director": {"type": "string", "description": "电影的导演"},
},
"required": ["title", "year", "director"]
}
model_with_structure = model.with_structured_output(
json_schema,
method="json_schema",
)
response = model_with_structure.invoke("请介绍电影《泰坦尼克号》")
print(response)
|