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

XiongNeng 收录于 AI

2026-10-03 约 7000 字预计阅读 14 分钟

学习进度

消息是模型接口的通用语言。你传进 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 原样收下，历史你自己维护。

消息的 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 写进提示词，不强制结构，最兼容也最松散，字段名容易跑偏。

能力边界由模型决定，不由你的代码决定。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 的入参有三种形式，字符串、消息对象列表、字典列表，后两种才承载完整对话。

1
2
3
4
5
6
7
8
9
from langchain.messages import AIMessage, HumanMessage, SystemMessage

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

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

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

真实模型跑一遍。

1
2
3
4
5
6
7
当前模型: ChatOpenAI
模型名: deepseek-v4.1-flash

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

2. content 与 content_blocks

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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)

1
2
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 键即可。

1
2
3
4
5
6
7
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 必须给。

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

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

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

真实模型认图。

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

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

4. AIMessage 元数据

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

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

1
2
3
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 对齐。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
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,
])

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

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

6. 序列化

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

1
2
3
4
5
6
7
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__)

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

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

7. 结构化输出三种 schema

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
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__)

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

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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__)

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

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
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)

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

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

完整案例

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

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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，解析失败不抛异常，把原始输出和错误一起返回给调用方。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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 写进提示词，模型有自由发挥的空间，钉死字段名才能把约束补回来。

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

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

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
模型: ScriptedChatModel（本地替身）
参数: temperature=None max_tokens=None
文本: 诺兰执导的《盗梦空间》2010年上映，莱昂纳多饰 Cobb

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

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

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

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

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
模型: 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