Agent学习笔记（四）LangChain基础

2026-07-05

笔记

LangChain 不是一个更聪明的模型，而是 LLM 应用的工程骨架。

它把模型调用、messages、prompt、tool、RAG、结构化输出和 Agent 循环放到一套统一接口里。

前面几篇分别看了 Agent、Function Calling 和 RAG；这一篇补上“这些能力如何被组织成应用”。

一、LangChain 是什么

一句话：

LangChain = 用来构建 LLM 应用的组件框架。

它不是模型本身，也不是单纯的 API SDK。

它解决的是工程组织问题：

问题	LangChain 提供什么
模型厂商很多，调用方式不一致	统一模型接口
prompt 越写越散	PromptTemplate / ChatPromptTemplate
多轮对话需要结构化上下文	messages
模型要调用工具	tool / create_agent
输出要给程序继续处理	structured output / parser
链路要支持并发、流式、重试	Runnable 统一执行接口

所以 LangChain 的重点不是“让模型更会回答”，而是：

1	把模型能力变成可组合、可维护、可观测的应用链路

使用判断：

场景	选择
只调用一次模型，做翻译、总结、改写	直接用模型 SDK 更轻
需要 prompt、模型、解析器串起来	LangChain
需要工具调用、结构化结果、多轮上下文	LangChain
需要复杂状态机、循环、条件分支、持久化	LangGraph
需要 Trace、评估、线上调试	LangSmith

这一篇只看 LangChain 基础，不展开 LangGraph 工作流和 LangSmith 评估。

二、包拆分：先知道代码从哪里导入

LangChain v1 的一个重要变化是拆包。

不要把所有东西都想成 langchain 一个包。

包	作用	使用直觉
`langchain`	主入口，高阶应用能力	`create_agent`、统一模型初始化
`langchain-core`	核心抽象	messages、prompts、tools、runnables、output parsers
`langchain-openai` / `langchain-anthropic` 等	厂商集成	连接具体模型服务
`langchain-community`	社区集成	文档加载器、向量库、第三方工具等
`langchain-classic`	旧 API 兼容	迁移旧项目时再考虑

基本安装口径：

1 2	pip install -U langchain pip install -U langchain-openai

如果只用 OpenAI，就装 OpenAI 集成包。

如果用 DeepSeek、Ollama、Anthropic、Google，也优先找对应 provider 包。

langchain-community 很丰富，但生产里要额外确认维护状态、依赖版本和 provider 能力支持。

三、Model 和 messages

LangChain 里常用的是 ChatModel。

它的输入不是一整段字符串，而是一组带角色的 messages。

from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, SystemMessage

model = init_chat_model(
    "gpt-4o-mini",
    model_provider="openai",
    temperature=0,
)

messages = [
    SystemMessage("你是一个简洁的技术助手。"),
    HumanMessage("用一句话解释 LangChain。"),
]

response = model.invoke(messages)
print(response.content)

messages 的价值：

role	作用
`system`	设定身份、规则、风格
`user` / `human`	用户当前输入
`assistant` / `ai`	模型历史回复
`tool`	工具执行结果，回填给模型

要点：

单轮简单任务可以直接传字符串。
多轮对话、工具调用、多模态输入，优先用 messages。
不同 provider 对 role、metadata、content block 的支持不完全一致。
AIMessage 不只是文本，还可能包含 tool_calls、usage_metadata、content_blocks。

四、Prompt

PromptTemplate 的作用是把提示词变成可复用模板。

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    "请用{tone}的语气解释：{topic}"
)

text = prompt.format(tone="简洁", topic="Runnable")

ChatPromptTemplate 更适合对话模型：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 Python 代码审查助手。"),
    ("human", "请审查以下代码：\n{code}"),
])

messages = prompt.format_messages(code="def add(a,b): return a+b")

两类模板的区别：

类型	输出	适合
`PromptTemplate`	字符串	单段文本提示
`ChatPromptTemplate`	messages	对话、系统提示、多轮结构
`partial_variables`	预填固定变量	固定角色、风格、约束

实践口径：

用户输入用变量填。
系统规则用 system message 放。
RAG 的 context、Agent 的工具说明，不要和用户输入揉成一坨字符串。

五、Runnable 和 LCEL

Runnable 是 LangChain 的统一执行接口。

Prompt、Model、Parser、Tool、Retriever 都可以被看成“可运行组件”。

只要一个组件是 Runnable，它就可以用同一组方法调用，也可以被放进 chain 里。

chain 是多个 Runnable 按顺序组合出来的执行流水线。

1	前一个组件的输出 -> 后一个组件的输入

最常见的 chain：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个简洁的技术助手。"),
    ("human", "解释：{topic}"),
])

chain = prompt | model | StrOutputParser()

answer = chain.invoke({"topic": "LangChain Runnable"})

这里的 | 表示把三个 Runnable 串起来：

1	Prompt Runnable -> Model Runnable -> Parser Runnable

实际数据流是：

{"topic": "..."}
-> ChatPromptValue / messages
-> AIMessage
-> string

分段看：

组件	输入	输出	职责
`prompt`	`dict`	messages / prompt value	把变量渲染成模型输入
`model`	messages / prompt value	`AIMessage`	调用模型生成结果
`StrOutputParser()`	`AIMessage`	`str`	提取可直接使用的文本

所以 chain 不是一个新模型，也不是一段普通函数调用。

它是把多个可运行组件接成一条稳定的数据处理链。

统一执行方法：

方法	用途
`invoke()`	单次调用
`batch()`	多个独立输入批量处理
`stream()`	逐块输出，适合前端打字机效果
`ainvoke()` / `abatch()`	异步调用
`astream_events()`	观察链路中的语义事件

Runnable 的意义：

组件可以自由组合，执行方式保持统一。

六、Tool 和 Agent

Function Calling 解决“模型如何提出工具调用请求”。

LangChain 在此基础上提供了工具封装和 create_agent 入口。

最小工具：

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询城市天气。"""
    return f"{city} 今天晴，温度 25 度。"

最小 Agent：

from langchain.agents import create_agent

agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="你是一个可靠的任务助手。",
)

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "北京今天适合出门吗？"}
    ]
})

print(result["messages"][-1].content)

需要分清：

层	做什么
Tool schema	告诉模型工具名、参数和用途
Model	判断要不要调用工具，生成工具请求
Program	校验参数，执行真实函数
ToolMessage	把工具结果放回上下文
`create_agent`	组织模型、工具和循环调用

所以：

1 2	模型不执行工具，程序执行工具。 Agent 不是一个模型，而是一套模型 + 工具 + 控制逻辑。

官方文档会把 create_agent 称为可配置的 harness。

基础阶段可以先把它理解成：

1	把模型调用、工具调用和结果回填组织成 Agent 循环的运行封装

七、结构化输出

结构化输出 = 把模型回答约束成 schema，让程序拿到可校验、可继续处理的数据。

它解决的是“模型结果怎么进入程序世界”的问题，不是“模型一定理解正确”的问题。

目标	普通回答	结构化输出
给人读	`"张三需要在下周一前完成项目报告。"`	不一定需要
给程序用	还要正则、切分、猜字段	`Todo(title=..., owner=..., due_date=...)`
写数据库 / 调 API	容易被自然语言格式影响	字段、类型、必填项先固定

心智模型：

用户文本
  -> schema 描述字段
  -> 模型按字段生成
  -> LangChain / Pydantic 校验
  -> 程序拿到对象或字典

常见场景：

场景	输出重点
信息抽取	从文本里抽 `title`、`owner`、`due_date`
意图识别	返回 `route` / `intent`，决定下一步走哪条链
生成配置	返回可直接保存或传给 API 的 JSON
Agent 最终结果	把多步工具调用后的结论固定成业务对象

比如希望从一句话里得到这样的对象：

1	Todo(title="完成项目报告", owner="张三", due_date="下周一")

先定义 schema。类型标注负责约束字段形状，Field(description=...) 负责告诉模型这个字段应该填什么：

from pydantic import BaseModel, Field

class Todo(BaseModel):
    title: str = Field(description="待办标题")
    owner: str | None = Field(default=None, description="负责人")
    due_date: str | None = Field(default=None, description="截止日期")

两种常用入口

如果只是单次模型调用，用 with_structured_output 把普通模型包装成结构化 runnable：

structured_model = model.with_structured_output(Todo)

todo = structured_model.invoke("张三需要在下周一前完成项目报告。")

print(todo)
# Todo(title="完成项目报告", owner="张三", due_date="下周一")

print(todo.title)
# 完成项目报告

关键点：structured_model.invoke(...) 返回的是 Todo 对象，不是普通字符串。后续代码可以直接读 todo.title，不需要再从自然语言里猜字段。

如果是 Agent 的最终结果需要结构化，用 create_agent(..., response_format=Todo)：

from langchain.agents import create_agent

agent = create_agent(
    model=model,
    tools=[],
    response_format=Todo,
)

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "张三需要在下周一前完成项目报告。"}
    ]
})

todo = result["structured_response"]

两种入口要分清：

入口	适合	返回位置
`model.with_structured_output(Todo)`	抽取、分类、生成配置这类单步任务	直接返回 `Todo` 或 dict
`create_agent(..., response_format=Todo)`	Agent 跑完工具循环后，最终答案要结构化	`result["structured_response"]`
`JsonOutputParser`	自己控制 prompt 和 JSON 解析格式	parser 解析后的结果

基础阶段优先记前两个。JsonOutputParser 更像“我自己写 prompt，最后手动解析 JSON”；with_structured_output 和 response_format 更像“把 schema 交给 LangChain 和模型能力层处理”。

底层怎么让模型按 schema 返回，取决于具体 provider 和模型能力。基础阶段不用展开，直接传 schema，再用实际调用结果和业务校验确认可靠性。

多任务对话怎么处理

不要把“整个聊天机器人”都绑定成 Todo 输出。

结构化输出更适合作为某个步骤的类型约束：这一步抽取待办，就返回 Todo；那一步抽取日程，就返回 CalendarEvent；主对话仍然可以返回自然语言。

场景	做法	直觉
只抽一种东西	一个 schema 配一个 structured model	简单、稳定
一句话里可能有多类结果	用更大的 schema 包住多种字段	输出形状稳定时好用
长期多功能助手	路由器 + 多个专用 structured model	每个节点只做一件事
Agent 多轮使用工具后再交付	只把最终结果设成 `response_format`	工具循环和最终交付分开

如果一次抽取里需要同时包含待办、日程和普通回复，可以用一个聚合 schema：

from pydantic import BaseModel, Field
from typing import Literal

class Todo(BaseModel):
    title: str
    due_date: str | None = None

class CalendarEvent(BaseModel):
    title: str
    start_time: str

class AssistantAction(BaseModel):
    kind: Literal["todo", "calendar", "reply"]
    todos: list[Todo] = Field(default_factory=list)
    events: list[CalendarEvent] = Field(default_factory=list)
    reply: str | None = None

structured_model = model.with_structured_output(AssistantAction)

如果对话要长期做很多事情，更推荐拆成多个专用步骤：

1
2
3

route_decider = model.with_structured_output(RouteDecision)
todo_extractor = model.with_structured_output(Todo)
event_extractor = model.with_structured_output(CalendarEvent)

这样主链路可以是：

用户消息
  -> 普通对话 / Agent 判断意图
  -> 需要结构化时调用专用 extractor
  -> 更新应用状态：todos、calendar、notes、memory
  -> 最后给用户自然语言回复

结论：with_structured_output(Todo) 这个 runnable 只负责产出 Todo，但系统可以有很多 runnable。多任务对话里，结构化输出通常是节点级约束，不是整个聊天机器人的唯一输出格式。

注意边界

schema 管的是输出形状，不是业务事实。

问题	例子	还要做什么
类型正确但语义不一定正确	`due_date="下周一"`	结合时区和当前日期归一化
字段存在但可能是猜的	文本没写负责人，模型补了人名	允许 `None`，缺失时回问用户
枚举合法但业务不合法	`priority="high"`	再跑业务规则校验
解析失败或校验失败	日期格式、数值范围不符合 schema	记录 raw response，重试或降级

工程上记四点：

Pydantic schema 会做运行时校验；TypedDict 和 JSON Schema 更偏结构描述，必要时自己补校验。
include_raw=True 可以同时拿到原始 AIMessage、解析结果和解析错误，排查时很有用。
provider 原生结构化输出通常更稳，但不是所有模型都支持同样能力。
关键链路不要只相信 schema，通过业务校验后再写数据库或调用外部 API。

八、Content Blocks：统一多模态内容

前面说 messages 不是一整段字符串，而是一组结构化消息。

Content Blocks 继续往下走一步：一条 message 里的 content 也不一定只是一段文本，它可以由多个内容块组成。

心智模型：

Message
  -> role：这是谁说的
  -> content：原始载荷，可能是字符串，也可能是 provider 原生 block 列表
  -> content_blocks：LangChain 标准化后的 block 视图

所以这里要分清两件事：

属性	更像什么	适合什么时候看
`content`	原始载荷	想看 provider 实际返回了什么
`content_blocks`	标准视图	想让程序按统一格式处理文本、图片、工具调用、reasoning 等
`text`	文本快捷视图	只关心最终可读文本

content_blocks 的价值，不是“让模型突然支持多模态”，而是：

1 2	把不同 provider 对文本、图片、文件、工具调用、reasoning 的不同表示，尽量整理成一套 LangChain 能统一读取和传递的内容块格式。

它常见的 block 类型包括：

block 类型	表示
`text`	普通文本
`tool_call`	工具调用
`reasoning`	推理或思考块，取决于 provider
`image` / `audio` / `video`	图片、音频、视频
`file`	PDF 等通用文件
`text-plain`	文档文本，例如 `.txt`、`.md`

两种使用方向

Content Blocks 可以出现在两个方向：

方向	解决什么问题	例子
输入给模型	用统一结构传图片、文件、音频等上下文	“看这张图，描述里面的问题”
读取模型输出	用统一结构拆出文本、工具调用、reasoning、引用等	“把文本块显示给用户，把工具调用交给程序执行”

最简单的输出读取：

response = model.invoke(messages)

for block in response.content_blocks:
    print(block["type"])

如果只要最终回答文本，直接用：

1	print(response.text)

如果要做应用层处理，就遍历 blocks：

for block in response.content_blocks:
    if block["type"] == "text":
        render_text(block["text"])
    elif block["type"] == "tool_call":
        run_tool(block["name"], block["args"])
    elif block["type"] == "reasoning":
        log_reasoning_summary(block.get("reasoning"))

这就是 content_blocks 的工程意义：程序不用先猜 content 里到底是 OpenAI 格式、Anthropic 格式，还是纯字符串。

输入多模态内容

如果用户消息里有图片，可以用标准 content block 描述：

from langchain.messages import HumanMessage

message = HumanMessage(content_blocks=[
    {"type": "text", "text": "请描述这张图里的系统结构。"},
    {
        "type": "image",
        "url": "https://example.com/architecture.png",
    },
])

response = model.invoke([message])

如果图片来自本地读取后转成 base64，通常还要给 mime_type：

message = HumanMessage(content_blocks=[
    {"type": "text", "text": "这张图里有哪些实体和关系？"},
    {
        "type": "image",
        "base64": image_base64,
        "mime_type": "image/png",
    },
])

PDF、音频、视频也是类似思路，只是 block 类型和 mime_type 不同：

message = HumanMessage(content_blocks=[
    {"type": "text", "text": "总结这个 PDF 的主要结论。"},
    {
        "type": "file",
        "base64": pdf_base64,
        "mime_type": "application/pdf",
    },
])

注意：有些 provider 还会要求额外字段，例如文件名、文件 ID 或特定 URL 形式。这些 provider 特有信息可以放在 block 顶层字段，或者放进 extras。

标准格式和原生格式

LangChain 允许 message 的内容有几种写法：

写法	示例	特点
字符串	`"你好"`	最简单，适合纯文本
provider 原生 blocks	OpenAI 的 `image_url` 等格式	贴近原厂 API，但可移植性差
LangChain 标准 blocks	`{"type": "image", "url": ...}`	更适合跨 provider 的应用代码

例如 OpenAI 原生格式可能长这样：

HumanMessage(content=[
    {"type": "text", "text": "描述这张图。"},
    {"type": "image_url", "image_url": {"url": image_url}},
])

标准 blocks 则写成：

HumanMessage(content_blocks=[
    {"type": "text", "text": "描述这张图。"},
    {"type": "image", "url": image_url},
])

实践里优先用标准 blocks。只有当某个 provider 的新能力还没有被 LangChain 标准格式覆盖，或者你明确依赖原厂字段时，再用原生格式。

和 Tool Calling 的关系

工具调用也可以被看成一种内容块。

但工程上通常还有一个更方便的入口：

1	response.tool_calls

所以排查工具调用时可以分两层看：

视角	看什么
`response.tool_calls`	模型提出了哪些工具调用，参数是什么
`response.content_blocks`	这条 AIMessage 里标准化后包含哪些 block
`response.content`	provider 原始返回长什么样

如果工具没有被调用，先看 tool_calls；如果 provider 返回结构很奇怪，再回到 content 和 content_blocks 对照。

和 reasoning 的关系

有些模型会返回 reasoning、thinking、summary 之类的信息。

content_blocks 会尽量把不同 provider 的这类内容统一成 reasoning block。

但不要把它当成稳定业务字段：

注意点	原因
不要依赖 reasoning 做业务判断	provider 可能不返回、只返回摘要，或格式变化
不要把 reasoning 原样展示给用户	它更适合调试、审计或内部观察
真正给程序用的结论，仍然用结构化输出	schema 比 reasoning 更适合做接口契约

可以这样理解：

1 2	reasoning block 适合观察模型怎么组织中间过程； structured output 适合约束模型最终交给程序的数据。

基础阶段怎么记

多模态场景里，优先确认三件事：

目标模型是否支持图片、音频或文件。
provider 需要的是标准 block 还是原生格式。
mime_type、URL、base64 是否符合 provider 要求。

工程上再加三条：

只读文本，用 response.text。
要分块处理，用 response.content_blocks。
要查 provider 原始行为，用 response.content。

不要默认“LangChain 支持 content blocks = 所有模型都支持所有多模态输入”。

九、Batch、Streaming、Retry

LangChain 基础链路里最常用的工程控制：

能力	方法	解决什么
批处理	`batch()` / `abatch()`	多条独立请求并发执行
流式输出	`stream()`	长回答更快反馈给用户
事件流	`astream_events()`	观察 prompt、model、chain 的执行过程
重试	`with_retry()`	临时网络错误、限流后的自动重试
并发限制	`RunnableConfig(max_concurrency=N)`	控制资源占用和接口压力

流式输出：

1 2	for chunk in model.stream("用一句话解释 LangChain。"): print(chunk.content, end="", flush=True)

批处理：

questions = [
    "什么是 LangChain？",
    "什么是 Runnable？",
    "什么是 Agent 循环？",
]

responses = model.batch(questions)

重试：

safe_model = model.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

判断口径：

云端 API 多数是 I/O-bound，适合并发。
本地单卡模型可能是 GPU-bound，并发不一定更快。
流式要求整条链路都能处理 chunk。
batch() 是客户端并发，不等同于模型厂商的离线 batch API。

十、排查顺序

LangChain 应用出错时，先别急着怀疑模型。

按链路拆：

症状	先看哪里
模型答非所问	最终 messages 里到底放了什么
工具没被调用	tool 名称、description、参数 schema 是否清楚
工具参数错	schema 是否太宽，程序是否做二次校验
JSON 解析失败	prompt 是否要求结构，是否保留 raw response
RAG 答案飘	retriever 取回了哪些 chunk
流式没效果	chain 中是否有组件必须等完整输出
并发没提速	API 限流、本地推理资源、同步客户端阻塞

核心排查问题：

模型实际看见了什么？
模型实际返回了什么？
程序实际执行了什么？
结果有没有回填到上下文？

这四个问题能定位大多数 LangChain 初学阶段的问题。

十一、这一篇先记住

核心结论：

LangChain 是 LLM 应用框架，不是模型。
v1 的主线是统一模型接口、messages、tools、structured output 和 Agent 循环封装。
PromptTemplate 解决复用，messages 解决上下文结构。
Runnable 让 prompt、model、parser、tool 可以用同一套方法执行和组合。
create_agent 把模型、工具和控制循环包起来，但工具仍由程序真实执行。
结构化输出让模型结果进入程序世界，但 schema 不能替代业务校验。
Batch、streaming、retry、trace 是从 demo 走向应用时必须补齐的工程能力。