学习笔记 · Obsidian

短期状态、流式协议与结构化输出

LangChainPython

三个概念不要混在一起

概念解决的问题核心载体
短期记忆同一会话跨轮保存消息和自定义 statecheckpointer + thread_id
流式运行中持续输出 token、步骤、工具与状态stream/event projections
结构化输出最终结果满足业务 schema,可直接进入程序response_format + validation

它们可以组合:带 checkpointer 的 agent 在同一 thread 上运行,前端通过流式协议展示过程,完成时从 state 的 structured_response 读取已校验结果。

短期记忆:thread 级状态

短期记忆属于 agent state,最常见字段是会话消息。创建 agent 时传入 checkpointer,并在每次调用复用同一个 thread_id,即可在一个 thread 内恢复历史;不同 thread 彼此隔离。

状态在每次调用或工具步骤完成后写入,在下一步骤开始前读取。这意味着工具、副作用和中间件要按“步骤可能重放/恢复”设计,不能把未持久化的进程内变量当作事实来源。

  • InMemorySaver 适合示例和单元测试,不具备进程重启后的持久性。
  • 生产应使用数据库支持的 checkpointer;官方示例给出 Postgres saver,其他后端需按 LangGraph checkpointer 能力选择。
  • 自定义 state 通过扩展 AgentState 加字段;并行写入字段时必须定义合适 reducer。

长会话治理

完整历史既可能超过上下文窗口,也可能因旧信息干扰而降低质量、增加延迟和费用。常见策略:

  1. Trim:在单次模型调用前保留最近 N 条/限定 token,通常是瞬时上下文处理;
  2. Delete:用 RemoveMessage 永久删除 state 中的消息;
  3. Summarize:用 SummarizationMiddleware 把旧消息压成摘要并持久替换,同时保留最近消息;
  4. 自定义过滤:按业务相关度、附件状态、隐私等级或阶段选择消息。

删除或裁剪后必须保持 provider 合法的消息序列,特别是 assistant tool call 后必须有对应 tool result,且部分 provider 要求历史从 user 消息开始。摘要会损失原始细节;多模态旧消息被摘要后通常只保留文本语义,原媒体应放在对象存储/文件系统并用引用管理。

事件流:新项目优先 v3

LangChain agent 基于 LangGraph,因此同时支持旧的 stream modes 与新的 Event Streaming。

Event Streaming v3

新应用优先 stream_events(..., version="v3")。它返回一个 run 对象,不需要对混合 tuple 做大量分支,而是通过独立 typed projections 消费:

  • messages:模型文本、reasoning 与完整 message stream;
  • tool_calls:参数、输出 delta、最终输出或错误;
  • subagents:有名字的 create_agent 子代理;
  • subgraphs:所有嵌套 LangGraph 子图;
  • values:state snapshots;
  • output:最终 agent state;
  • custom/extensions:检索进度、artifact、业务事件等自定义投影。

异步场景可用 astream_events 并发消费多个 projection;同步场景可用 interleave 合并所需频道。需要底层诊断时仍可遍历 raw protocol envelope。

Middleware 可以注册 stream transformer,把事件投影为业务频道或在 wire 层做 PII redaction;文档当前标注 middleware-registered transformer 需要 langchain>=1.3.2,且每个 subgraph scope 应产生独立 transformer 实例。

传统 streaming

stream/astream 仍支持:

mode内容
updates每个 agent step 的 state update
messagesLLM token/chunk 与 metadata
customnode/tool 通过 stream writer 发出的业务进度

旧接口仍适合既有应用和简单调试,但复杂 UI 应优先 typed projections。无论哪种接口,都要处理:工具调用参数渐进到达、reasoning 与文本交错、多工具并行、子代理 namespace、HITL 暂停与恢复、客户端断线后的持久运行。

结构化输出:schema 是契约,不是提示词愿望

create_agent(response_format=...) 把最终结果写入 state 的 structured_response。可选策略:

  • 直接传 Pydantic/dataclass/TypedDict schema type:框架根据模型 profile 自动选策略;
  • ProviderStrategy:provider 原生 schema 约束,通常最可靠;
  • ToolStrategy:把 schema 当成工具调用,适用于支持 tool calling 但无原生结构化输出的模型;
  • None:不要求结构化结果。

当前文档的关键版本/能力边界:

  • 直接传 schema 时,LangChain 根据 model profile 选择 provider/native 或 tool strategy;profile 数据能力从 langchain>=1.1 起动态读取。
  • JSON Schema 字典不会像 Python schema type 一样被自动识别,必须显式包在 ProviderStrategy 或 ToolStrategy 中。
  • ProviderStrategy(strict=...) 的 strict 参数要求 langchain>=1.2,且只对支持它的 provider 生效。
  • agent 同时使用业务 tools 与 structured output 时,模型必须支持二者并用。

Schema 与失败处理

Pydantic 适合需要运行时校验、范围和字段说明的业务结果;dataclass/TypedDict 更轻;JSON Schema 用于跨语言契约;ToolStrategy 还支持 union,让模型选择多个候选 schema 之一。

工具策略可能出现两类典型错误:模型一次返回多个结构化工具、或字段不符合 schema。默认会把校验反馈作为 ToolMessage 让模型重试。handle_errors 可以选择全部重试、只重试特定异常、返回固定/动态错误说明,或关闭重试让异常上抛。

生产建议:

  • 限制结构化重试次数,避免模型在不可满足 schema 上无限消耗;
  • schema 表达真实业务约束,不能在模型返回后盲目信任;
  • 业务侧仍要做授权、数据库约束、幂等和语义校验;
  • 日志记录校验类型和 trace,不记录敏感原文;
  • UI 对部分 streaming args 做完整性检查,关键字段未到齐时不要执行副作用或渲染成“已完成”。

逐页覆盖索引

页面学习结论
Short-term memorycheckpointer/thread 边界、自定义 state、生产持久化,以及 trim/delete/summarize 和在 tool/prompt/hooks 中读写 state。
Streamingupdates/messages/custom、多模式、reasoning/tool/subagent/HITL 流和兼容格式。
Event streamingv3 typed projections、sync/async 多频道、subagents/subgraphs、state/output 与 transformer 扩展。
Structured outputProviderStrategy/ToolStrategy 自动选择、schema 类型、strict/profile 版本边界和可配置校验重试。

延伸