-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Description
Description
Summary
Add a run_id field to ModelRequest and ModelResponse to enable reliable splitting and grouping of message history by agent run. This would provide a "handle" to identify which messages belong to a specific run, which is critical when manipulating, compacting, or resuming multi-turn conversations.
Problem
Currently, there is no reliable way to split or group a list[ModelMessage] history by run. While developers can work around this by iterating over new_messages() after each run and manually tracking boundaries, this approach:
- Requires strong assumptions about message order and completeness
- Is fragile when messages are manipulated, filtered, or reordered
- Becomes error-prone in complex scenarios like:
- Conversation compaction/summarization (as discussed in the message history docs)
- Resuming conversations from persistent storage
- Analyzing specific runs from long conversation histories
- Debugging tool call sequences across runs
Proposed Solution
Add a run_id: str field to both ModelRequest and ModelResponse dataclasses:
@dataclass(repr=False)
class ModelRequest:
"""A request generated by Pydantic AI and sent to a model."""
parts: Sequence[ModelRequestPart]
_: KW_ONLY
instructions: str | None = None
run_id: str | None = None # NEW: UUID identifying the run this message belongs to
kind: Literal['request'] = 'request'@dataclass(repr=False)
class ModelResponse:
"""A response from a model."""
parts: Sequence[ModelResponsePart]
_: KW_ONLY
usage: RequestUsage = field(default_factory=RequestUsage)
model_name: str | None = None
timestamp: datetime = field(default_factory=_now_utc)
kind: Literal['response'] = 'response'
provider_name: str | None = None
provider_details: dict[str, Any] | None = None
provider_response_id: str | None = None
finish_reason: FinishReason | None = None
run_id: str | None = None # NEW: UUID identifying the run this message belongs toImplementation Details
1. Run ID Generation
The run_id should be generated once per agent run, likely when the graph execution begins or in the AgentRun initialization:
import uuid
# Generate once per run
run_id = str(uuid.uuid4())2. Propagation
The run_id should be:
- Passed through the graph execution context (
GraphAgentState) - Stamped on all
ModelRequestmessages created during the run - Stamped on all
ModelResponsemessages received during the run - Preserved when messages are serialized/deserialized via
ModelMessagesTypeAdapter
3. Backward Compatibility
Since run_id would be optional (str | None), existing code and serialized messages will continue to work:
- Old messages without
run_idwill deserialize withrun_id=None - New messages will have
run_idpopulated - No breaking changes to the public API
Benefits
- Reliable filtering:
[msg for msg in history if msg.run_id == target_run_id] - Run-aware grouping:
itertools.groupby(history, key=lambda msg: msg.run_id) - Safer history processing: History processors can group by run and preserve run boundaries
- Better debugging: Easily trace all messages from a problematic run
- Cleaner conversation management: No need for manual index tracking when continuing conversations
- Improved observability: Correlate messages with OpenTelemetry spans using the same run ID
Example Usage
from pydantic_ai import Agent
agent = Agent('openai:gpt-4o')
# Run 1
result1 = agent.run_sync('What is 2+2?')
run1_id = result1.new_messages()[0].run_id
# Run 2 (continuing conversation)
result2 = agent.run_sync('What about 3+3?', message_history=result1.new_messages())
run2_id = result2.new_messages()[0].run_id
# Now we can reliably filter by run
all_history = result2.all_messages()
run1_messages = [msg for msg in all_history if msg.run_id == run1_id]
run2_messages = [msg for msg in all_history if msg.run_id == run2_id]Alternative Considered
As mentioned in the original Slack discussion, the current workaround is to use provider_metadata: dict[str, Any] on ModelResponse to manually add run_id after each run:
result = agent.run_sync('Hello')
for msg in result.new_messages():
if isinstance(msg, ModelResponse):
if msg.provider_metadata is None:
msg.provider_metadata = {}
msg.provider_metadata['run_id'] = str(uuid.uuid4())However, this approach:
- Only works for
ModelResponse, notModelRequest - Requires manual intervention after every run
- Doesn't work well with immutable dataclasses (requires mutation)
- Pollutes provider-specific metadata with framework-level concerns
Related Code
pydantic_ai_slim/pydantic_ai/messages.py- Message dataclasses (lines 927-1367)pydantic_ai_slim/pydantic_ai/result.py-all_messages()andnew_messages()methodspydantic_ai_slim/pydantic_ai/_agent_graph.py- Graph execution statepydantic_ai_slim/pydantic_ai/run.py-AgentRunclass- Message History Documentation
References
- Original Slack thread discussion (October 2, 2024)
- @DouweM suggestion to add
ModelResponse.run_idfield - Feedback that all messages (Request & Response) should be stamped with a shared unique ID
References
No response