Pydantic AI ProcessHistory 深度解析:每次模型请求前的消息历史拦截与重写【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai本篇围绕 Pydantic AI 的ProcessHistory能力展开:它如何在每次模型请求发出前拦截、改写甚至整体替换消息历史,适用于裁剪旧轮次、脱敏敏感内容、按 token 用量压缩上下文等场景。读完后,你将能够编写同步/异步、可感知RunContext的历史处理器,理解处理结果如何就地替换运行状态中的历史、如何影响new_messages()的边界判定,并掌握用FunctionModel离线验证处理器行为的完整测试方法。ProcessHistory 是什么:一个挂在 before_model_request 上的薄包装[ProcessHistory][pydantic_ai.capabilities.ProcessHistory] 是一个 capability,它包装一个历史处理器函数:在每次向模型发起请求之前,该函数接收当前消息历史,并返回(可能被修改过的)待发送消息列表。典型的动机包括:出于隐私考虑过滤敏感信息、为节省 token 成本裁剪旧上下文、或执行任意自定义处理逻辑(详见 消息历史文档)。最简用法如下:from pydantic_ai import Agent from pydantic_ai.capabilities import ProcessHistory from pydantic_ai.messages import ModelMessage def keep_recent(messages: list[ModelMessage]) - list[ModelMessage]: return messages[-5:] # (1)! agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(keep_recent)])只保留最近 5 条消息。实践中通常还需要保留第一条请求,让系统提示词(system prompt)存活下来——完整的裁剪模式参见处理消息历史。有三个关键性质需要牢记:处理器可以是同步的,也可以是异步的;可选地以 [RunContext][pydantic_ai.tools.RunContext] 作为第一个参数,从而访问依赖项(ctx.deps)、模型信息与用量统计等运行状态;多个ProcessHistory能力按注册顺序依次应用(第 N 个处理器看到的是第 N-1 个处理器的输出);处理后的消息会替换本次运行的消息历史,而不是只影响这一次请求的载荷——因此若你还想保留原始历史,必须先自己拷贝一份。从源码结构看,ProcessHistory的实现非常轻:它是一个只持有processor字段的数据类,唯一的逻辑就是实现before_model_request能力钩子,把request_context.messages替换为处理器返回值——见 capabilities/process_history.py:dataclass class ProcessHistory(AbstractCapability[AgentDepsT]): A capability that processes message history before model requests. processor: HistoryProcessorFunc[AgentDepsT] async def before_model_request( self, ctx: RunContext[AgentDepsT], request_context: ModelRequestContext, ) - ModelRequestContext: request_context.messages await _run_history_processor(self.processor, ctx, request_context.messages) return request_context也就是说,ProcessHistory本质上是一个面向迁移的薄封装(official note:ProcessHistoryis a migration-friendly wrapper aroundbefore_model_request)。如果你需要更丰富的控制——访问完整的 [ModelRequestContext][pydantic_ai.models.ModelRequestContext]、带超时的钩子、甚至抛出SkipModelRequest(response)来短路跳过本次模型调用——可以直接挂钩该生命周期事件,参见生命周期钩子文档。处理器函数契约:HistoryProcessor 的四种签名处理器类型定义在 _history_processor.py 中,是一个四元联合类型:_HistoryProcessorSync Callable[[list[_messages.ModelMessage]], list[_messages.ModelMessage]] _HistoryProcessorAsync Callable[[list[_messages.ModelMessage]], Awaitable[list[_messages.ModelMessage]]] _HistoryProcessorSyncWithCtx Callable[[RunContext[DepsT], list[_messages.ModelMessage]], list[_messages.ModelMessage]] _HistoryProcessorAsyncWithCtx Callable[ [RunContext[DepsT], list[_messages.ModelMessage]], Awaitable[list[_messages.ModelMessage]] ] HistoryProcessor ( _HistoryProcessorSync | _HistoryProcessorAsync | _HistoryProcessorSyncWithCtx[DepsT] | _HistoryProcessorAsyncWithCtx[DepsT] ) A function that processes a list of model messages and returns a list of model messages. Can optionally accept a RunContext as a parameter. 这个类型是公开可导入的(from pydantic_ai.capabilities import HistoryProcessor),便于你在自己的代码里给处理器做类型标注——tests/test_history_processor.py 中的test_history_processor_public_type专门验证了这一点:把一个标注为HistoryProcessor[None]的剥离系统提示词处理器传给ProcessHistory后运行正常。框架如何分派四种签名运行时分派逻辑集中在 _run_history_processor:async def _run_history_processor( processor: HistoryProcessorFunc[AgentDepsT], ctx: RunContext[AgentDepsT], messages: list[_messages.ModelMessage], ) - list[_messages.ModelMessage]: Run a history processor, handling sync/async and with/without context variants. takes_ctx takes_run_context(processor) if is_async_callable(processor): if takes_ctx: return await cast(_AsyncWithCtx, processor)(ctx, messages) else: return await cast(_AsyncNoCtx, processor)(messages) else: # A plain def may still return an awaitable, which run_in_executor would leave un-awaited. if takes_ctx: return await await_maybe(await run_in_executor(cast(_SyncWithCtx, processor), ctx, messages)) else: return await await_maybe(await run_in_executor(cast(_SyncNoCtx, processor), messages))这里有两处容易被忽略的工程细节:RunContext的识别靠运行时类型标注解析。takes_run_context 通过get_first_param_type检查第一个参数的类型注解是否为RunContext或其泛型实例。因此处理器签名里每一个类型标注都必须在运行时可导入,而不能只放在if TYPE_CHECKING:块里——否则任何无法解析的标注都会抛出 [UserError][pydantic_ai.exceptions.UserError],而不是悄悄退化成无上下文调用。普通def返回协程的坑会被兜住。若你写了一个非async def但返回协程对象的处理器(这在HistoryProcessor类型中是合法成员),run_in_executor本身不会 await 那个协程,于是代码额外套了一层 await_maybe:检查返回值是否 awaitable,是则 await,否则原样返回。测试用例test_sync_history_processor_returning_coroutine验证了这种形态下过滤逻辑确实生效——如果没兜住,转换就永远不会应用。同步(非异步)处理器会在线程执行器中运行,避免阻塞事件循环;同步处理器里做耗时 IO 时建议直接写成async def。核心语义:处理结果替换运行历史,并连带影响 new_messages()ProcessHistory最不像纯过滤器的地方在于副作用范围:处理器返回的消息列表会写回运行状态,成为后续逻辑眼中的历史。这一点在官方文档中有明确警告(message-history.md):History processors replace the message history in the state with the processed messages, including the new user prompt part. This means that if you want to keep the original message history, you need to make a copy of it.测试对这一语义有系统覆盖,例如 test_history_processor_run_replaces_message_history:处理器把历史收敛为最后一条请求 一条新的SystemPromptPart请求后,result.all_messages()反映的就是处理后的历史,且result.new_messages() result.all_messages()(因为原历史已被消费,整个可见历史都算本次运行的消息)。测试同时揭示了一个渲染细节:非首条请求中的SystemPromptPart属于会话中途系统提示,发给模型时会被渲染为同一位置的system.../system用户内容——测试快照中的systemProcessed answer/system正对应此行为。与 new_messages() 的配合规则[new_messages()][pydantic_ai.agent.AgentRunResult.new_messages] 只返回当前运行产生的消息,message_history提供的消息被排除在外(包括不带用户提示词恢复运行时那个可能被框架打上当前run_id的尾部ModelRequest)。为了让你的处理器在改动消息后仍不破坏这一边界,官方给出两条守则(message-history.md):若你重建了尾部ModelRequest,请保留它的parts、timestamp、instructions和metadata,以便它仍能被识别为先前上下文;若你插入了应当计入new_messages()的新消息,请使用感知上下文的处理器,并在新消息上设置run_idctx.run_id。no-op 处理器测试 给出了基线:new_messages() result.all_messages()[-2:],即新的 ModelRequest 新的 ModelResponse。不要破坏工具调用配对;延迟工具需要保留 ToolAvailabilityDeltaPart两条与工具相关的警告值得单独强调:裁剪/摘要时必须保证 tool call 与 tool return 成对出现,否则部分模型提供方会直接拒绝请求。官方文档在保留最近消息和摘要旧消息两个示例处都挂了同一警告:切片历史时要检查工具调用与返回的配对完整性(message-history.md)。使用延迟工具(deferred tools)时,处理器必须保留ToolAvailabilityDeltaPart条目,或保留可供 Pydantic AI 重建它们的完整load_capability调用/返回对。因为工具的已揭示(revealed)状态是从发给模型的处理后历史推导的——如果处理器或摘要器把两种表示都丢了,对应工具会重新变成隐藏状态。实战配方一:保留最近消息与上下文感知裁剪同步:过滤所有 ModelResponsefrom pydantic_ai import ( Agent, ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) from pydantic_ai.capabilities import ProcessHistory def filter_responses(messages: list[ModelMessage]) - list[ModelMessage]: Remove all ModelResponse messages, keeping only ModelRequest messages. return [msg for msg in messages if isinstance(msg, ModelRequest)] # Create agent with history processor agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(filter_responses)]) # Example: Create some conversation history message_history [ ModelRequest(parts[UserPromptPart(contentWhat is 22?)]), ModelResponse(parts[TextPart(content22 equals 4)]), # This will be filtered out ] # When you run the agent, the history processor will filter out ModelResponse messages # result agent.run_sync(What about 33?, message_historymessage_history)异步:只保留最近 5 条以控制 token 用量from pydantic_ai import Agent, ModelMessage from pydantic_ai.capabilities import ProcessHistory async def keep_recent_messages(messages: list[ModelMessage]) - list[ModelMessage]: Keep only the last 5 messages to manage token usage. return messages[-5:] if len(messages) 5 else messages agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(keep_recent_messages)])感知上下文:按 token 用量动态收缩处理器签名中带上RunContext后,就可以读取依赖、模型名、用量统计等:from pydantic_ai import Agent, ModelMessage, RunContext from pydantic_ai.capabilities import ProcessHistory def context_aware_processor( ctx: RunContext, messages: list[ModelMessage], ) - list[ModelMessage]: # Access current usage current_tokens ctx.usage.total_tokens # Filter messages based on context if current_tokens 1000: return messages[-3:] # Keep only recent messages when token usage is high return messages agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(context_aware_processor)])测试 test_history_processor_with_context 演示了经ctx.deps读取依赖并给每条用户消息加前缀的完整链路;test_history_processor_mixed_signatures 则确认了带上下文与不带上下文的处理器可以混用在同一批 capabilities 中。实战配方二:用 LLM 摘要旧消息用一个更便宜的模型把最旧的若干消息压缩成摘要,是保留语义又压低 token 的经典手法:from pydantic_ai import Agent, ModelMessage from pydantic_ai.capabilities import ProcessHistory # Use a cheaper model to summarize old messages. summarize_agent Agent( openai:gpt-5-mini, instructions Summarize this conversation, omitting small talk and unrelated topics. Focus on the technical discussion and next steps. , ) async def summarize_old_messages(messages: list[ModelMessage]) - list[ModelMessage]: # Summarize the oldest 10 messages if len(messages) 10: oldest_messages messages[:10] summary await summarize_agent.run(message_historyoldest_messages) # Return the last message and the summary return summary.new_messages() messages[-1:] return messages agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(summarize_old_messages)])再次提醒摘要场景的工具配对风险:摘要器丢弃的若是不成对的 tool call/return 片段,模型请求可能被提供方拒收。摘要属于上下文窗口管理的诸多手段之一,提供方原生的压缩能力与现成策略见压缩(Compaction)文档。实战配方三:上下文窗口接近打满时压缩前两个配方每轮都会重写历史;如果只想在会话逼近模型的context_window时才动手,可以用ctx.context_window_used——它返回最近一次响应之后被占用掉的窗口比例,无法可靠计算时返回None:from pydantic_ai import ( Agent, ModelMessage, ModelRequest, RetryPromptPart, RunContext, ToolReturnPart, UserPromptPart, ) from pydantic_ai.capabilities import ProcessHistory, ReinjectSystemPrompt def compact_when_window_fills( ctx: RunContext, messages: list[ModelMessage], ) - list[ModelMessage]: used ctx.context_window_used if used is None or used 0.8: return messages # Keep the most recent complete user turn, including any later tool calls and returns. for index in range(len(messages) - 1, -1, -1): message messages[index] if not isinstance(message, ModelRequest): continue has_user_prompt any(isinstance(part, UserPromptPart) for part in message.parts) has_tool_result any(isinstance(part, (ToolReturnPart, RetryPromptPart)) for part in message.parts) if has_user_prompt and not has_tool_result: return messages[index:] return messages agent Agent( openai:gpt-5.2, system_promptYou are a helpful assistant., capabilities[ProcessHistory(compact_when_window_fills), ReinjectSystemPrompt()], )围绕这个模式,文档给出了几条边界处理准则(message-history.md):把None当作未知而不是空窗口:它会在首次模型响应之前、以及模型窗口或响应用量未知时返回,示例在这些情况下保持历史不变;FallbackModel按候选中最小的窗口度量,因此压缩会提前到无论哪个候选应答都来得及的时机发生;ReinjectSystemPrompt必须排在压缩处理器之后(如示例所示),这样被连同旧历史一起丢掉的系统提示词才能被重新注入;示例从最近一个纯用户轮次(只含用户提示词、不含工具结果的请求)起保留全部后续内容;如果某轮把工具结果和新提示词绑在一起,该轮会被整体保留,可能比必要的留得更多;对自定义/本地模型,若数据源未收录其窗口大小,可显式设置profile{context_window: 128_000}。多个处理器的顺序、约束与失败模式多个ProcessHistory按注册顺序应用。官方示例(message-history.md):from pydantic_ai import Agent, ModelMessage, ModelRequest from pydantic_ai.capabilities import ProcessHistory def filter_responses(messages: list[ModelMessage]) - list[ModelMessage]: return [msg for msg in messages if isinstance(msg, ModelRequest)] def summarize_old_messages(messages: list[ModelMessage]) - list[ModelMessage]: return messages[-5:] agent Agent( openai:gpt-5.2, capabilities[ProcessHistory(filter_responses), ProcessHistory(summarize_old_messages)], )此时filter_responses先执行、summarize_old_messages后执行。test_multiple_history_processors 用两个加前缀处理器验证了顺序:最终用户提示词呈现为[SECOND] [FIRST] Question,证明第二个处理器作用在第一个的输出之上。处理结果还有两条硬性校验(由 tests/test_history_processor.py 固化):处理结果行为空列表抛出UserError,信息为Processed history cannot be empty.以ModelResponse结尾抛出UserError,信息为Processed history must end with a \ModelRequest.即:处理器必须返回一个以ModelRequest收尾的非空列表——这符合agent 总要有一个可响应的请求的运行不变量。另外注意,处理器拿到的输入已经包含本轮新的用户提示词(尾部ModelRequest),所以messages[-1:]这类保留最后一条的写法天然保住了当前轮输入。此外,从源码看 get_serialization_name 返回None:ProcessHistory因持有可调用的处理器而不参与 spec 序列化。用 FunctionModel 测试你的处理器验证模型实际收到了什么消息的推荐方式是FunctionModel——用一个纯函数模型捕获入参,完全离线、无需 API key:import pytest from pydantic_ai import ( Agent, ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) from pydantic_ai.capabilities import ProcessHistory from pydantic_ai.models.function import AgentInfo, FunctionModel pytest.fixture def received_messages() - list[ModelMessage]: return [] pytest.fixture def function_model(received_messages: list[ModelMessage]) - FunctionModel: def capture_model_function(messages: list[ModelMessage], info: AgentInfo) - ModelResponse: # Capture the messages that the provider actually receives received_messages.clear() received_messages.extend(messages) return ModelResponse(parts[TextPart(contentProvider response)]) return FunctionModel(capture_model_function) def test_history_processor(function_model: FunctionModel, received_messages: list[ModelMessage]): def filter_responses(messages: list[ModelMessage]) - list[ModelMessage]: return [msg for msg in messages if isinstance(msg, ModelRequest)] agent Agent(function_model, capabilities[ProcessHistory(filter_responses)]) message_history [ ModelRequest(parts[UserPromptPart(contentQuestion 1)]), ModelResponse(parts[TextPart(contentAnswer 1)]), ] agent.run_sync(Question 2, message_historymessage_history) assert received_messages [ ModelRequest(parts[UserPromptPart(contentQuestion 1)]), ModelRequest(parts[UserPromptPart(contentQuestion 2)]), ]该断言的成立依赖一条实现细节:被过滤掉的ModelResponse移除后,两条相邻的ModelRequest会被合并为一条(两个用户部分聚合),这正是消息历史修复管线中相邻兼容消息合并的体现。仓库内 tests/test_history_processor.py 的完整测试矩阵还包括:流式运行(run_stream/agent.iter驱动)下的替换语义、异步与同步返回协程处理器的等价行为、deps传递、混合签名、整体替换历史、空历史与以ModelResponse结尾的报错,以及可调用类(callable class)作为处理器——可据此直接复制到自己项目的测试中。何时越过 ProcessHistory,直接挂 before_model_request 钩子ProcessHistory只解决改写消息列表这一件事。若你的需求是:读取完整的ModelRequestContext(工具、模型设置等)、基于请求状态短路跳过模型调用(抛出SkipModelRequest),或为钩子配置超时,直接走 Hooks 能力更合适:from pydantic_ai.capabilities import Hooks hooks.on.before_model_request def log_request(ctx, request_context): ... # 可抛 SkipModelRequest(response) 直接短路本次模型请求 agent Agent(test, capabilities[Hooks(before_model_requestlog_request)])注意before_model_request每个模型轮次触发一次(提供方中途暂停或后台模式续跑时同样适用),且它看到的是即将发送的请求的完整状态(包括当前函数工具与模型设置)。ProcessHistory作为其薄包装,继承同样的触发时机,只是把返回修改后的request_context简化成了传入消息列表、返回消息列表。小结与延伸阅读ProcessHistory的契约一句话:一个list[ModelMessage] → list[ModelMessage]的同步/异步函数,可选首参RunContext;处理结果替换运行历史,会连带影响all_messages()/new_messages()的边界,改写尾部请求要保留元数据、新增消息要盖run_idctx.run_id;裁剪与摘要务必维护 tool call/return 配对,延迟工具场景保留ToolAvailabilityDeltaPart;处理器输出必须非空且以ModelRequest结尾,否则UserError;需要短路模型调用或访问请求级状态时,直接用Hooks(before_model_request...)。相关文档与源码入口:Process History 能力文档(本文主体)处理消息历史(完整指南)生命周期钩子、压缩能力、ReinjectSystemPrompt实现:capabilities/process_history.py、类型定义、分派与 RunContext 探测测试:tests/test_history_processor.py完整对话式应用示例:chat app【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Pydantic AI ProcessHistory 深度解析:每次模型请求前的消息历史拦截与重写
Pydantic AI ProcessHistory 深度解析:每次模型请求前的消息历史拦截与重写【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai本篇围绕 Pydantic AI 的ProcessHistory能力展开:它如何在每次模型请求发出前拦截、改写甚至整体替换消息历史,适用于裁剪旧轮次、脱敏敏感内容、按 token 用量压缩上下文等场景。读完后,你将能够编写同步/异步、可感知RunContext的历史处理器,理解处理结果如何就地替换运行状态中的历史、如何影响new_messages()的边界判定,并掌握用FunctionModel离线验证处理器行为的完整测试方法。ProcessHistory 是什么:一个挂在 before_model_request 上的薄包装[ProcessHistory][pydantic_ai.capabilities.ProcessHistory] 是一个 capability,它包装一个历史处理器函数:在每次向模型发起请求之前,该函数接收当前消息历史,并返回(可能被修改过的)待发送消息列表。典型的动机包括:出于隐私考虑过滤敏感信息、为节省 token 成本裁剪旧上下文、或执行任意自定义处理逻辑(详见 消息历史文档)。最简用法如下:from pydantic_ai import Agent from pydantic_ai.capabilities import ProcessHistory from pydantic_ai.messages import ModelMessage def keep_recent(messages: list[ModelMessage]) - list[ModelMessage]: return messages[-5:] # (1)! agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(keep_recent)])只保留最近 5 条消息。实践中通常还需要保留第一条请求,让系统提示词(system prompt)存活下来——完整的裁剪模式参见处理消息历史。有三个关键性质需要牢记:处理器可以是同步的,也可以是异步的;可选地以 [RunContext][pydantic_ai.tools.RunContext] 作为第一个参数,从而访问依赖项(ctx.deps)、模型信息与用量统计等运行状态;多个ProcessHistory能力按注册顺序依次应用(第 N 个处理器看到的是第 N-1 个处理器的输出);处理后的消息会替换本次运行的消息历史,而不是只影响这一次请求的载荷——因此若你还想保留原始历史,必须先自己拷贝一份。从源码结构看,ProcessHistory的实现非常轻:它是一个只持有processor字段的数据类,唯一的逻辑就是实现before_model_request能力钩子,把request_context.messages替换为处理器返回值——见 capabilities/process_history.py:dataclass class ProcessHistory(AbstractCapability[AgentDepsT]): A capability that processes message history before model requests. processor: HistoryProcessorFunc[AgentDepsT] async def before_model_request( self, ctx: RunContext[AgentDepsT], request_context: ModelRequestContext, ) - ModelRequestContext: request_context.messages await _run_history_processor(self.processor, ctx, request_context.messages) return request_context也就是说,ProcessHistory本质上是一个面向迁移的薄封装(official note:ProcessHistoryis a migration-friendly wrapper aroundbefore_model_request)。如果你需要更丰富的控制——访问完整的 [ModelRequestContext][pydantic_ai.models.ModelRequestContext]、带超时的钩子、甚至抛出SkipModelRequest(response)来短路跳过本次模型调用——可以直接挂钩该生命周期事件,参见生命周期钩子文档。处理器函数契约:HistoryProcessor 的四种签名处理器类型定义在 _history_processor.py 中,是一个四元联合类型:_HistoryProcessorSync Callable[[list[_messages.ModelMessage]], list[_messages.ModelMessage]] _HistoryProcessorAsync Callable[[list[_messages.ModelMessage]], Awaitable[list[_messages.ModelMessage]]] _HistoryProcessorSyncWithCtx Callable[[RunContext[DepsT], list[_messages.ModelMessage]], list[_messages.ModelMessage]] _HistoryProcessorAsyncWithCtx Callable[ [RunContext[DepsT], list[_messages.ModelMessage]], Awaitable[list[_messages.ModelMessage]] ] HistoryProcessor ( _HistoryProcessorSync | _HistoryProcessorAsync | _HistoryProcessorSyncWithCtx[DepsT] | _HistoryProcessorAsyncWithCtx[DepsT] ) A function that processes a list of model messages and returns a list of model messages. Can optionally accept a RunContext as a parameter. 这个类型是公开可导入的(from pydantic_ai.capabilities import HistoryProcessor),便于你在自己的代码里给处理器做类型标注——tests/test_history_processor.py 中的test_history_processor_public_type专门验证了这一点:把一个标注为HistoryProcessor[None]的剥离系统提示词处理器传给ProcessHistory后运行正常。框架如何分派四种签名运行时分派逻辑集中在 _run_history_processor:async def _run_history_processor( processor: HistoryProcessorFunc[AgentDepsT], ctx: RunContext[AgentDepsT], messages: list[_messages.ModelMessage], ) - list[_messages.ModelMessage]: Run a history processor, handling sync/async and with/without context variants. takes_ctx takes_run_context(processor) if is_async_callable(processor): if takes_ctx: return await cast(_AsyncWithCtx, processor)(ctx, messages) else: return await cast(_AsyncNoCtx, processor)(messages) else: # A plain def may still return an awaitable, which run_in_executor would leave un-awaited. if takes_ctx: return await await_maybe(await run_in_executor(cast(_SyncWithCtx, processor), ctx, messages)) else: return await await_maybe(await run_in_executor(cast(_SyncNoCtx, processor), messages))这里有两处容易被忽略的工程细节:RunContext的识别靠运行时类型标注解析。takes_run_context 通过get_first_param_type检查第一个参数的类型注解是否为RunContext或其泛型实例。因此处理器签名里每一个类型标注都必须在运行时可导入,而不能只放在if TYPE_CHECKING:块里——否则任何无法解析的标注都会抛出 [UserError][pydantic_ai.exceptions.UserError],而不是悄悄退化成无上下文调用。普通def返回协程的坑会被兜住。若你写了一个非async def但返回协程对象的处理器(这在HistoryProcessor类型中是合法成员),run_in_executor本身不会 await 那个协程,于是代码额外套了一层 await_maybe:检查返回值是否 awaitable,是则 await,否则原样返回。测试用例test_sync_history_processor_returning_coroutine验证了这种形态下过滤逻辑确实生效——如果没兜住,转换就永远不会应用。同步(非异步)处理器会在线程执行器中运行,避免阻塞事件循环;同步处理器里做耗时 IO 时建议直接写成async def。核心语义:处理结果替换运行历史,并连带影响 new_messages()ProcessHistory最不像纯过滤器的地方在于副作用范围:处理器返回的消息列表会写回运行状态,成为后续逻辑眼中的历史。这一点在官方文档中有明确警告(message-history.md):History processors replace the message history in the state with the processed messages, including the new user prompt part. This means that if you want to keep the original message history, you need to make a copy of it.测试对这一语义有系统覆盖,例如 test_history_processor_run_replaces_message_history:处理器把历史收敛为最后一条请求 一条新的SystemPromptPart请求后,result.all_messages()反映的就是处理后的历史,且result.new_messages() result.all_messages()(因为原历史已被消费,整个可见历史都算本次运行的消息)。测试同时揭示了一个渲染细节:非首条请求中的SystemPromptPart属于会话中途系统提示,发给模型时会被渲染为同一位置的system.../system用户内容——测试快照中的systemProcessed answer/system正对应此行为。与 new_messages() 的配合规则[new_messages()][pydantic_ai.agent.AgentRunResult.new_messages] 只返回当前运行产生的消息,message_history提供的消息被排除在外(包括不带用户提示词恢复运行时那个可能被框架打上当前run_id的尾部ModelRequest)。为了让你的处理器在改动消息后仍不破坏这一边界,官方给出两条守则(message-history.md):若你重建了尾部ModelRequest,请保留它的parts、timestamp、instructions和metadata,以便它仍能被识别为先前上下文;若你插入了应当计入new_messages()的新消息,请使用感知上下文的处理器,并在新消息上设置run_idctx.run_id。no-op 处理器测试 给出了基线:new_messages() result.all_messages()[-2:],即新的 ModelRequest 新的 ModelResponse。不要破坏工具调用配对;延迟工具需要保留 ToolAvailabilityDeltaPart两条与工具相关的警告值得单独强调:裁剪/摘要时必须保证 tool call 与 tool return 成对出现,否则部分模型提供方会直接拒绝请求。官方文档在保留最近消息和摘要旧消息两个示例处都挂了同一警告:切片历史时要检查工具调用与返回的配对完整性(message-history.md)。使用延迟工具(deferred tools)时,处理器必须保留ToolAvailabilityDeltaPart条目,或保留可供 Pydantic AI 重建它们的完整load_capability调用/返回对。因为工具的已揭示(revealed)状态是从发给模型的处理后历史推导的——如果处理器或摘要器把两种表示都丢了,对应工具会重新变成隐藏状态。实战配方一:保留最近消息与上下文感知裁剪同步:过滤所有 ModelResponsefrom pydantic_ai import ( Agent, ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) from pydantic_ai.capabilities import ProcessHistory def filter_responses(messages: list[ModelMessage]) - list[ModelMessage]: Remove all ModelResponse messages, keeping only ModelRequest messages. return [msg for msg in messages if isinstance(msg, ModelRequest)] # Create agent with history processor agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(filter_responses)]) # Example: Create some conversation history message_history [ ModelRequest(parts[UserPromptPart(contentWhat is 22?)]), ModelResponse(parts[TextPart(content22 equals 4)]), # This will be filtered out ] # When you run the agent, the history processor will filter out ModelResponse messages # result agent.run_sync(What about 33?, message_historymessage_history)异步:只保留最近 5 条以控制 token 用量from pydantic_ai import Agent, ModelMessage from pydantic_ai.capabilities import ProcessHistory async def keep_recent_messages(messages: list[ModelMessage]) - list[ModelMessage]: Keep only the last 5 messages to manage token usage. return messages[-5:] if len(messages) 5 else messages agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(keep_recent_messages)])感知上下文:按 token 用量动态收缩处理器签名中带上RunContext后,就可以读取依赖、模型名、用量统计等:from pydantic_ai import Agent, ModelMessage, RunContext from pydantic_ai.capabilities import ProcessHistory def context_aware_processor( ctx: RunContext, messages: list[ModelMessage], ) - list[ModelMessage]: # Access current usage current_tokens ctx.usage.total_tokens # Filter messages based on context if current_tokens 1000: return messages[-3:] # Keep only recent messages when token usage is high return messages agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(context_aware_processor)])测试 test_history_processor_with_context 演示了经ctx.deps读取依赖并给每条用户消息加前缀的完整链路;test_history_processor_mixed_signatures 则确认了带上下文与不带上下文的处理器可以混用在同一批 capabilities 中。实战配方二:用 LLM 摘要旧消息用一个更便宜的模型把最旧的若干消息压缩成摘要,是保留语义又压低 token 的经典手法:from pydantic_ai import Agent, ModelMessage from pydantic_ai.capabilities import ProcessHistory # Use a cheaper model to summarize old messages. summarize_agent Agent( openai:gpt-5-mini, instructions Summarize this conversation, omitting small talk and unrelated topics. Focus on the technical discussion and next steps. , ) async def summarize_old_messages(messages: list[ModelMessage]) - list[ModelMessage]: # Summarize the oldest 10 messages if len(messages) 10: oldest_messages messages[:10] summary await summarize_agent.run(message_historyoldest_messages) # Return the last message and the summary return summary.new_messages() messages[-1:] return messages agent Agent(openai:gpt-5.2, capabilities[ProcessHistory(summarize_old_messages)])再次提醒摘要场景的工具配对风险:摘要器丢弃的若是不成对的 tool call/return 片段,模型请求可能被提供方拒收。摘要属于上下文窗口管理的诸多手段之一,提供方原生的压缩能力与现成策略见压缩(Compaction)文档。实战配方三:上下文窗口接近打满时压缩前两个配方每轮都会重写历史;如果只想在会话逼近模型的context_window时才动手,可以用ctx.context_window_used——它返回最近一次响应之后被占用掉的窗口比例,无法可靠计算时返回None:from pydantic_ai import ( Agent, ModelMessage, ModelRequest, RetryPromptPart, RunContext, ToolReturnPart, UserPromptPart, ) from pydantic_ai.capabilities import ProcessHistory, ReinjectSystemPrompt def compact_when_window_fills( ctx: RunContext, messages: list[ModelMessage], ) - list[ModelMessage]: used ctx.context_window_used if used is None or used 0.8: return messages # Keep the most recent complete user turn, including any later tool calls and returns. for index in range(len(messages) - 1, -1, -1): message messages[index] if not isinstance(message, ModelRequest): continue has_user_prompt any(isinstance(part, UserPromptPart) for part in message.parts) has_tool_result any(isinstance(part, (ToolReturnPart, RetryPromptPart)) for part in message.parts) if has_user_prompt and not has_tool_result: return messages[index:] return messages agent Agent( openai:gpt-5.2, system_promptYou are a helpful assistant., capabilities[ProcessHistory(compact_when_window_fills), ReinjectSystemPrompt()], )围绕这个模式,文档给出了几条边界处理准则(message-history.md):把None当作未知而不是空窗口:它会在首次模型响应之前、以及模型窗口或响应用量未知时返回,示例在这些情况下保持历史不变;FallbackModel按候选中最小的窗口度量,因此压缩会提前到无论哪个候选应答都来得及的时机发生;ReinjectSystemPrompt必须排在压缩处理器之后(如示例所示),这样被连同旧历史一起丢掉的系统提示词才能被重新注入;示例从最近一个纯用户轮次(只含用户提示词、不含工具结果的请求)起保留全部后续内容;如果某轮把工具结果和新提示词绑在一起,该轮会被整体保留,可能比必要的留得更多;对自定义/本地模型,若数据源未收录其窗口大小,可显式设置profile{context_window: 128_000}。多个处理器的顺序、约束与失败模式多个ProcessHistory按注册顺序应用。官方示例(message-history.md):from pydantic_ai import Agent, ModelMessage, ModelRequest from pydantic_ai.capabilities import ProcessHistory def filter_responses(messages: list[ModelMessage]) - list[ModelMessage]: return [msg for msg in messages if isinstance(msg, ModelRequest)] def summarize_old_messages(messages: list[ModelMessage]) - list[ModelMessage]: return messages[-5:] agent Agent( openai:gpt-5.2, capabilities[ProcessHistory(filter_responses), ProcessHistory(summarize_old_messages)], )此时filter_responses先执行、summarize_old_messages后执行。test_multiple_history_processors 用两个加前缀处理器验证了顺序:最终用户提示词呈现为[SECOND] [FIRST] Question,证明第二个处理器作用在第一个的输出之上。处理结果还有两条硬性校验(由 tests/test_history_processor.py 固化):处理结果行为空列表抛出UserError,信息为Processed history cannot be empty.以ModelResponse结尾抛出UserError,信息为Processed history must end with a \ModelRequest.即:处理器必须返回一个以ModelRequest收尾的非空列表——这符合agent 总要有一个可响应的请求的运行不变量。另外注意,处理器拿到的输入已经包含本轮新的用户提示词(尾部ModelRequest),所以messages[-1:]这类保留最后一条的写法天然保住了当前轮输入。此外,从源码看 get_serialization_name 返回None:ProcessHistory因持有可调用的处理器而不参与 spec 序列化。用 FunctionModel 测试你的处理器验证模型实际收到了什么消息的推荐方式是FunctionModel——用一个纯函数模型捕获入参,完全离线、无需 API key:import pytest from pydantic_ai import ( Agent, ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) from pydantic_ai.capabilities import ProcessHistory from pydantic_ai.models.function import AgentInfo, FunctionModel pytest.fixture def received_messages() - list[ModelMessage]: return [] pytest.fixture def function_model(received_messages: list[ModelMessage]) - FunctionModel: def capture_model_function(messages: list[ModelMessage], info: AgentInfo) - ModelResponse: # Capture the messages that the provider actually receives received_messages.clear() received_messages.extend(messages) return ModelResponse(parts[TextPart(contentProvider response)]) return FunctionModel(capture_model_function) def test_history_processor(function_model: FunctionModel, received_messages: list[ModelMessage]): def filter_responses(messages: list[ModelMessage]) - list[ModelMessage]: return [msg for msg in messages if isinstance(msg, ModelRequest)] agent Agent(function_model, capabilities[ProcessHistory(filter_responses)]) message_history [ ModelRequest(parts[UserPromptPart(contentQuestion 1)]), ModelResponse(parts[TextPart(contentAnswer 1)]), ] agent.run_sync(Question 2, message_historymessage_history) assert received_messages [ ModelRequest(parts[UserPromptPart(contentQuestion 1)]), ModelRequest(parts[UserPromptPart(contentQuestion 2)]), ]该断言的成立依赖一条实现细节:被过滤掉的ModelResponse移除后,两条相邻的ModelRequest会被合并为一条(两个用户部分聚合),这正是消息历史修复管线中相邻兼容消息合并的体现。仓库内 tests/test_history_processor.py 的完整测试矩阵还包括:流式运行(run_stream/agent.iter驱动)下的替换语义、异步与同步返回协程处理器的等价行为、deps传递、混合签名、整体替换历史、空历史与以ModelResponse结尾的报错,以及可调用类(callable class)作为处理器——可据此直接复制到自己项目的测试中。何时越过 ProcessHistory,直接挂 before_model_request 钩子ProcessHistory只解决改写消息列表这一件事。若你的需求是:读取完整的ModelRequestContext(工具、模型设置等)、基于请求状态短路跳过模型调用(抛出SkipModelRequest),或为钩子配置超时,直接走 Hooks 能力更合适:from pydantic_ai.capabilities import Hooks hooks.on.before_model_request def log_request(ctx, request_context): ... # 可抛 SkipModelRequest(response) 直接短路本次模型请求 agent Agent(test, capabilities[Hooks(before_model_requestlog_request)])注意before_model_request每个模型轮次触发一次(提供方中途暂停或后台模式续跑时同样适用),且它看到的是即将发送的请求的完整状态(包括当前函数工具与模型设置)。ProcessHistory作为其薄包装,继承同样的触发时机,只是把返回修改后的request_context简化成了传入消息列表、返回消息列表。小结与延伸阅读ProcessHistory的契约一句话:一个list[ModelMessage] → list[ModelMessage]的同步/异步函数,可选首参RunContext;处理结果替换运行历史,会连带影响all_messages()/new_messages()的边界,改写尾部请求要保留元数据、新增消息要盖run_idctx.run_id;裁剪与摘要务必维护 tool call/return 配对,延迟工具场景保留ToolAvailabilityDeltaPart;处理器输出必须非空且以ModelRequest结尾,否则UserError;需要短路模型调用或访问请求级状态时,直接用Hooks(before_model_request...)。相关文档与源码入口:Process History 能力文档(本文主体)处理消息历史(完整指南)生命周期钩子、压缩能力、ReinjectSystemPrompt实现:capabilities/process_history.py、类型定义、分派与 RunContext 探测测试:tests/test_history_processor.py完整对话式应用示例:chat app【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
