# 使用LangChain+LangGraph开发AI应用（3） - 消息与结构化输出


![](https://static.xiongneng.me/langchain-langgraph-cover-20261017150727.png)

## 学习进度

消息是模型接口的通用语言。你传进 `invoke` 的一切，最后都会被整理成一个消息列表发给模型，字符串只是框架帮你包好的语法糖。多轮对话、多模态、工具调用，这几个场景里你必须亲手构造消息，所以先把这门语言讲透。

讲完消息讲结构化输出，让模型从陪你聊天变成给你干活。示例在 `source/hello_messages.py`，七个 demo。完整案例是一个结构化信息抽取台 `source/extract.py`，一条命令把自然语言抽成结构化数据。

## 前言用途

消息是模型接口的通用货币。展开说，你传进 `invoke` 的一切，最后都会被整理成一个消息列表发给模型。字符串只是语法糖，框架帮你包成了 `HumanMessage`，字典列表也能收，框架按 OpenAI chat completions 的风格自动转换。但糖吃多了会蛀牙。多轮对话、多模态、工具调用，这几个场景里你必须知道消息长什么样，因为要亲手构造它。

再说结构化输出。模型天生只会唠嗑，你问它一句，它还你一段自由文本。可程序要的是数据，字段、类型、能直接塞进数据库或者 `if` 判断里的东西。老办法是提示词里写「请输出 JSON」，拿到回复再 `json.loads`，失败了自己擦屁股。模型偶尔在 JSON 前后加句「好的，这是您要的信息」，解析就炸。更别提字段名时不时跑个偏，`year` 写成 `release_year`，校验又炸。

LangChain 的 `with_structured_output` 就是来收编这事的。你给它一份 schema，它还你一个解析好的对象，Pydantic 实例或者字典，解析失败有专门的错误字段兜底。这篇还会配一个结构化信息抽取台案例，一条命令把自然语言抽成结构化数据。

## 原理概念

一条消息由三样东西组成，角色、内容、元数据。角色决定权重和上下文，系统消息定人设，用户消息是输入。内容是载荷，纯文本或者内容块列表。元数据是边角料，消息 id、token 用量、提供商信息，不影响语义，但排查问题时救命。

四种消息类型对应四个角色。`SystemMessage` 装人设与规则，放在最前面。`HumanMessage` 装用户输入，可带多模态内容。`AIMessage` 装模型输出，可带 `tool_calls`。`ToolMessage` 装工具执行结果，靠 `tool_call_id` 与调用对齐。多轮对话就是这四种消息排成的列表，`invoke` 原样收下，历史你自己维护。

![](https://static.xiongneng.me/message-types-20261017150256.png)

消息的 `content` 是个松散类型，刚上手最容易在这儿犯迷糊。它可以是字符串，简单直接。也可以是原生内容块列表，注意是提供商原生格式，比如 Anthropic 的思考块长这样，`{"type": "thinking", "thinking": "...", "signature": "..."}`。同一个消息上还有个 `content_blocks` 属性，返回标准内容块，把各家的格式归一，上面那个 thinking 块会被归一成 `{"type": "reasoning", ...}`。`content_blocks` 是惰性解析的属性，访问它才干活，`content` 不受影响，原样保留。跨提供商处理消息，读 `content_blocks`；要存原始数据，读 `content`。

结构化输出的入口只有一行，`model.with_structured_output(schema, method=...)`。它返回一个可调用的 runnable，用法与 `model.invoke` 相同，返回值是解析校验后的结构化对象。schema 有三样写法，Pydantic 模型类、TypedDict、JSON Schema 字典，回报分别是带校验的模型实例、普通字典、普通字典。

`method` 参数决定 schema 怎么递到模型眼前，一共三条路。第一条 `json_schema`，提供商原生结构化输出，schema 通过 API 的 `response_format` 强约束，最可靠，OpenAI 和 Anthropic 都支持。第二条 `function_calling`，强制模型发起一次工具调用，schema 由工具参数承载，兼容广，需要模型支持工具调用。第三条 `json_mode`，只把 schema 写进提示词，不强制结构，最兼容也最松散，字段名容易跑偏。

![](https://static.xiongneng.me/structured-output-20261017150300.png)

能力边界由模型决定，不由你的代码决定。LangChain 1.1 起模型实例带 `profile` 属性，能查结构化输出能力，但数据来自 models.dev 开源项目，网关背后的模型经常查不到，我在 API易 网关上实测 `model.profile` 直接是 `None`，拿不到 degrade 线索。选 `method` 的靠谱顺序是这样，先试默认的 `json_schema`，被网关拒了换 `function_calling`，思考模型不支持强制 `tool_choice` 再退 `json_mode`。本篇测试环境是 API易 网关加 `deepseek-v4.1-flash`，前两条路都被拒，只有 `json_mode` 放行，所以案例里显式指定了它。

`include_raw=True` 也最好一开始就带上。默认情况下解析失败直接抛异常，线上你更想拿到原始输出自己兜底。开了它，返回值变成字典，带三个键，`raw` 是模型原始消息，`parsed` 是解析后的对象，`parsing_error` 是解析错误信息。三者永远同时在场，判个 `parsed is None` 就知道成没成。

## 核心用法

七个用法，逐个演示。示例在 `source/hello_messages.py`，不设 `AGENT_MODEL` 时用本地替身模型跑，不花一分钱。下面的输出本地替身和真实模型各给一版，对照着看。

### 1. 四种消息类型与三种入参

`invoke` 的入参有三种形式，字符串、消息对象列表、字典列表，后两种才承载完整对话。

```python
from langchain.messages import AIMessage, HumanMessage, SystemMessage

conversation = [
    SystemMessage("你是一个简洁的地理助手，回答不超过两句话。"),
    HumanMessage("北京是哪座城市？"),
    AIMessage("北京是中华人民共和国的首都。"),
    HumanMessage("它有多少年的建城史？"),
]
response = model.invoke(conversation)
```

对象列表是最常用的形式，四个构造函数的第一个参数都是内容，直接写字符串就行。字典列表是 OpenAI chat completions 风格，`role` 加 `content`，框架自动转换，适合从别处搬数据过来的场景。

```python
as_dicts = [
    {"role": "system", "content": "你是一个简洁的地理助手。"},
    {"role": "user", "content": "北京是哪座城市？"},
]
response = model.invoke(as_dicts)
```

真实模型跑一遍。

```text
当前模型: ChatOpenAI
模型名: deepseek-v4.1-flash

== 消息类型与三种入参 ==
对象列表入参: ['SystemMessage', 'HumanMessage', 'AIMessage', 'HumanMessage']
回答: 北京有3000多年建城史，通常认为始于西周时期的蓟城。
字典列表入参也能跑: 北京是中国的首都，也是中国的直辖市，不属于其他城市。...
```

### 2. content 与 content_blocks

不调模型，手工构造一条带提供商原生格式的消息，看两种视图的差别。这条 `AIMessage` 的 `content` 是 Anthropic 风格的块列表，带一个思考块和一个文本块。

```python
from langchain.messages import AIMessage

message = AIMessage(
    content=[
        {"type": "thinking", "thinking": "用户问的是消息的定义。", "signature": "WaUjzkyp"},
        {"type": "text", "text": "消息是对话的最小单元。"},
    ],
    response_metadata={"model_provider": "anthropic"},
)
print(message.content)
print(message.content_blocks)
```

```text
content 原样（提供商原生）: [{'type': 'thinking', 'thinking': '用户问的是消息的定义。', 'signature': 'WaUjzkyp'}, {'type': 'text', 'text': '消息是对话的最小单元。'}]
content_blocks（标准块）: [{'type': 'reasoning', 'reasoning': '用户问的是消息的定义。', 'extras': {'signature': 'WaUjzkyp'}}, {'type': 'text', 'text': '消息是对话的最小单元。'}]
```

看清楚了，`content` 原样保留提供商的 `thinking` 块，`content_blocks` 归一成标准的 `reasoning` 块，`signature` 挪进 `extras`。写跨提供商的代码，判断消息里有没有思考内容，读 `content_blocks` 判 `reasoning`，别去判各家的原生类型名。

### 3. 多模态输入

图片走 `HumanMessage` 的块列表，文本块和图片块混排。图片有三个来源，公网 URL、base64 加 `mime_type`、`file_id`，结构相同，换 `url` 键即可。

```python
from langchain.messages import HumanMessage

message = HumanMessage(content=[
    {"type": "text", "text": "用一句话描述这张图。"},
    {"type": "image", "url": "https://static.xiongneng.me/model-abstraction-20261017122518.png"},
])
response = model.invoke([message])
```

base64 来源适合图片存在本地或者数据库里的场景，`data` 前缀可选，`mime_type` 必须给。

```python
message = HumanMessage(content=[
    {"type": "text", "text": "用一句话描述这张图。"},
    {"type": "image", "base64": "<图片base64字符串>", "mime_type": "image/png"},
])
```

`file_id` 来源适合文件已经传给提供商的场景，比如 OpenAI 的 Files API，不用重复传字节。

```python
message = HumanMessage(content=[
    {"type": "text", "text": "用一句话描述这张图。"},
    {"type": "image", "file_id": "file-abc123"},
])
```

真实模型认图。

```text
== 多模态：图片输入 ==
图片输入也能跑: 这张图展示了一个统一的大模型调用架构：通过 `init_chat_model` ...
```

我的测试环境里图片 URL 走的是 OSS 公网地址，网关会自己抓取，模型确实看到了内容。但不同提供商对 URL 的可达性要求不一样，内网地址、需要鉴权的地址都会失败，遇到这种情况换 base64。

### 4. AIMessage 元数据

`AIMessage` 除内容外还带三样元数据，`id` 是消息标识，`usage_metadata` 是 token 用量，`response_metadata` 是提供商回的其他信息。注意用量要等响应完整才可信，流式场景里聚合之后再读。

```python
response = model.invoke("用一句话说明什么是消息")
print(response.id)
print(response.usage_metadata)
print(sorted(response.response_metadata)[:5])
```

```text
id: lc_run--01a148b3-aed1-7522-89cf-e91371d51a50-0
usage_metadata: {'input_tokens': 35, 'output_tokens': 97, 'total_tokens': 132, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 73}}
response_metadata 键: ['finish_reason', 'id', 'logprobs', 'model_name', 'model_provider']
```

这 97 个输出 token 里，73 个是 `reasoning`，推理模型的思考 token。做成本估算时把 `output_token_details.reasoning` 单独拉出来看，别只看 `total_tokens`，第二篇埋过这个伏笔，这里兑现。`input_token_details` 里的 `cache_read` 是提示缓存命中量，命中部分通常打折计费，长系统提示的场景能省不少。

### 5. ToolMessage 与 artifact

工具结果的回传是消息体系里最绕的一环，完整循环是下一篇的主题，这里只讲消息本身。模型先发起工具调用，返回的 `AIMessage` 里带 `tool_calls`，你把执行结果包成 `ToolMessage` 回传，`tool_call_id` 必须与调用的 `id` 对齐。

```python
from langchain.messages import HumanMessage, ToolMessage

def get_weather(location: str) -> str:
    """Get the weather at a location."""
    return "晴，26℃"

ai_message = model.bind_tools([get_weather]).invoke("北京天气怎么样？")
call = ai_message.tool_calls[0]

tool_message = ToolMessage(
    content="晴，26℃",
    tool_call_id=call["id"],
    name=call["name"],
    artifact={"source": "本地替身", "cached": False},
)
response = model.invoke([
    HumanMessage("北京天气怎么样？"),
    ai_message,
    tool_message,
])
```

```text
模型发起工具调用: get_weather({'location': '北京'})
工具结果回传后模型回答: 北京目前天气晴朗，气温约 26℃，体感比较舒适。
artifact 不下发给模型: {'source': '本地替身', 'cached': False}
```

`ToolMessage` 有两个内容字段，别搞混。`content` 是给模型看的工具执行结果，会进入上下文。`artifact` 是给下游程序看的附加数据，原样挂在消息对象上，不下发给模型。上游数据来源、缓存标记、原始 JSON，这些模型不需要知道但你的程序要用的东西，全挂 `artifact`。

### 6. 序列化

对话历史要落盘或者进数据库，用 `dumpd` 和 `load`。`dumpd` 把消息转成普通字典，可以存 JSON。`load` 把字典还原成消息对象，反过来。

```python
from langchain_core.load import dumpd, load

message = HumanMessage("北京是哪座城市？", id="msg_123")
serialized = dumpd(message)
restored = load(serialized)
print(serialized["type"])
print(type(restored).__name__)
```

```text
序列化成普通 dict: type=constructor（可存 JSON / 数据库）
反序列化回消息对象: HumanMessage，内容一致: True
⚠️ load() 会实例化 Python 对象，切勿对不可信数据调用
```

注意最后那行警告。`load` 会按字典里的类型信息实例化 Python 对象，对不可信数据调用等于开门揖盗。只存消息的话，把 `allowed_objects` 传成 `"messages"` 限定范围，这是 LangChain 1.1 起的推荐写法，还能消掉一串弃用警告。

### 7. 结构化输出三种 schema

同一份电影信息，换三种 schema 写法，回报各不相同。Pydantic 版带字段校验，返回模型实例。

```python
from pydantic import BaseModel, Field

class Movie(BaseModel):
    """A movie with details."""
    title: str = Field(description="The title of the movie")
    year: int = Field(description="The year the movie was released")
    director: str = Field(description="The director of the movie")

structured = model.with_structured_output(Movie, method="json_mode")
movie = structured.invoke(EXTRACT_PROMPT)
print(movie)
print(type(movie).__name__)
```

```text
解析结果: Movie(title='盗梦空间', year=2010, director='诺兰')
类型: Movie（带字段校验，返回模型实例）
```

TypedDict 版无运行时校验，返回普通字典，写起来更轻，适合字段简单的场景。

```python
from typing_extensions import Annotated, TypedDict

class MovieDict(TypedDict):
    title: Annotated[str, ..., "The title of the movie"]
    year: Annotated[int, ..., "The year the movie was released"]
    director: Annotated[str, ..., "The director of the movie"]

structured = model.with_structured_output(MovieDict, method="json_mode")
movie = structured.invoke(EXTRACT_PROMPT)
print(movie)
print(type(movie).__name__)
```

```text
解析结果: {'title': '盗梦空间', 'year': 2010, 'director': '诺兰'}
类型: dict（无运行时校验，返回 dict）
```

JSON Schema 字典的控制权最大，schema 可以来自文件或者接口，跨语言复用。用字典 schema 时必须显式指定 `method`，框架猜不出类型。

```python
MOVIE_JSON_SCHEMA = {
    "title": "Movie",
    "description": "A movie with details",
    "type": "object",
    "properties": {
        "title": {"type": "string", "description": "The title of the movie"},
        "year": {"type": "integer", "description": "The year the movie was released"},
        "director": {"type": "string", "description": "The director of the movie"},
    },
    "required": ["title", "year", "director"],
}

structured = model.with_structured_output(MOVIE_JSON_SCHEMA, method="json_mode")
movie = structured.invoke(EXTRACT_PROMPT)
print(movie)
```

```text
解析结果: {"title": "盗梦空间", "year": 2010, "director": "诺兰"}
```

三种写法怎么选。schema 就在代码里、要校验，用 Pydantic。schema 要跨语言或者来自外部，用 JSON Schema 字典。嫌弃前两者重，用 TypedDict。`method` 的选择看上一条原理，直连官方提供商用默认，走网关按探测结果降级。

## 完整案例

用法演示够了，写个完整的，结构化信息抽取台 `extract.py`，把一段自然语言抽成结构化电影信息，标题、年份、导演、演员表、类型。它不止能看电影。接一个新模型，先跑它十几条文本，看字段准不准，看烧多少 token，再决定用不用。

schema 比用法里的三字段版本多两个字段，`cast` 是演员列表，每项带 `name` 和 `role`，`genres` 是类型标签列表。两个字段都给了 `default_factory=list`，抽不到给空列表，不让必填校验把整次抽取毙掉。

```python
from pydantic import BaseModel, Field

class Actor(BaseModel):
    """一名演员及其饰演的角色。"""
    name: str = Field(description="演员姓名")
    role: str = Field(description="饰演的角色名")

class MovieInfo(BaseModel):
    """一部电影的结构化信息。"""
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")
    director: str = Field(description="导演")
    cast: list[Actor] = Field(default_factory=list, description="主要演员与角色")
    genres: list[str] = Field(default_factory=list, description="类型标签，如 科幻")
```

核心函数不到二十行，关键是 `include_raw=True`，解析失败不抛异常，把原始输出和错误一起返回给调用方。

```python
def extract(model, text):
    """抽取结构化电影信息，返回解析结果与统计数据。"""
    structured = model.with_structured_output(MovieInfo, method="json_mode", include_raw=True)
    started = time.perf_counter()
    out = structured.invoke(EXTRACT_PROMPT.format(text=text))
    elapsed = time.perf_counter() - started
    raw = out["raw"]
    return {
        "parsed": out["parsed"],
        "parsing_error": out["parsing_error"],
        "raw_text": raw.text if raw is not None else "",
        "usage": raw.usage_metadata if raw is not None else None,
        "elapsed_s": round(elapsed, 3),
    }
```

提示词里显式钉死了字段名，`json_mode` 只把 schema 写进提示词，模型有自由发挥的空间，钉死字段名才能把约束补回来。

```python
EXTRACT_PROMPT = (
    "从下面的文本里抽取电影信息，只输出 JSON，字段名必须严格使用 "
    "title、year、director、cast、genres，其中 cast 的每项用 name 和 role "
    "两个字段，不要增加任何其他字段，抽不到的元素给空列表。"
    "文本：{text}"
)
```

命令行入口用 argparse 收一个文本加三个参数，`--model` 换模型，`--temperature` 调随机性，`--max-tokens` 限长度。不传模型名时读环境变量，再没有就用本地替身。本地演示一遍。

```bash
uv run python extract.py "诺兰执导的《盗梦空间》2010年上映，莱昂纳多饰 Cobb"
```

```text
模型: ScriptedChatModel（本地替身）
参数: temperature=None max_tokens=None
文本: 诺兰执导的《盗梦空间》2010年上映，莱昂纳多饰 Cobb

---- 抽取结果 ----
标题: 盗梦空间
年份: 2010
导演: 克里斯托弗·诺兰
演员: 莱昂纳多·迪卡普里奥 饰 Cobb
类型: 科幻、悬疑

JSON: {
  "title": "盗梦空间",
  "year": 2010,
  "director": "克里斯托弗·诺兰",
  "cast": [
    {
      "name": "莱昂纳多·迪卡普里奥",
      "role": "Cobb"
    }
  ],
  "genres": [
    "科幻",
    "悬疑"
  ]
}

---- 统计 ----
耗时: 0.049s
token 用量: 无（fake 模型不产用量，真实模型才有）
```

替身模型按脚本供货，不产用量，程序明确告诉你原因，不装。接上真实模型再跑一遍，同一条文本。

```bash
# 先设好 AGENT_MODEL / YI_BASE_URL / YI_API_KEY
uv run python extract.py "诺兰执导的《盗梦空间》2010年上映，莱昂纳多饰 Cobb"
```

```text
模型: openai:deepseek-v4.1-flash
参数: temperature=None max_tokens=None
文本: 诺兰执导的《盗梦空间》2010年上映，莱昂纳多饰 Cobb

---- 抽取结果 ----
标题: 盗梦空间
年份: 2010
导演: 诺兰
演员: 莱昂纳多 饰 Cobb
类型: （无）

JSON: {
  "title": "盗梦空间",
  "year": 2010,
  "director": "诺兰",
  "cast": [
    {
      "name": "莱昂纳多",
      "role": "Cobb"
    }
  ],
  "genres": []
}

---- 统计 ----
耗时: 2.462s
token 用量: 输入 131 / 输出 176 / 合计 307（其中思考 136）
```

同一句话，两版输出的差别比想象中大。真实模型抽对了核心字段，但明显更保守，`director` 只给了「诺兰」没给全名，`genres` 直接空了，因为原句里确实没提类型。而 `genres` 给空列表不报错，正是 `default_factory=list` 的功劳。统计里 176 个输出 token，136 个是思考，占比接近八成。推理模型烧的 token，一大半在看不见的地方。

案例的测试在 `tests/test_extract.py`，五条用例，返回值结构、解析失败分支、提示词钉字段名、替身分支、真实模型分支的参数透传。全部用替身模型跑，不花一分钱。加上 `tests/test_hello_messages.py` 的十一条，本篇新增十六条测试，全绿。

## 源码

[https://github.com/yidao620c/xneng-langchain-langgraph](https://github.com/yidao620c/xneng-langchain-langgraph)

## 参考

- [LangChain 官方文档 - 消息](https://docs.langchain.com/oss/python/langchain/messages)
- [LangChain 官方文档 - 模型](https://docs.langchain.com/oss/python/langchain/models)
- [LangChain 官方文档 - 结构化输出](https://docs.langchain.com/oss/python/langchain/agents/structured-output)


---

> 作者: XiongNeng  
> URL: https://xiongneng.me/posts/ai/langchain/langchain-langgraph-03-messages-structured-output/