showcase/shell-docs/src/content/docs/integrations/pydantic-ai/shared-state/predictive-state-updates.mdx
<IframeSwitcher id="predictive-state-updates-example" exampleUrl="https://feature-viewer.copilotkit.ai/pydantic-ai/feature/shared_state?sidebar=false&chatDefaultOpen=false" codeUrl="https://feature-viewer.copilotkit.ai/pydantic-ai/feature/shared_state?view=code&sidebar=false&codeLayout=tabs" exampleLabel="Demo" codeLabel="Code" height="700px" />
<Callout type="info"> This example demonstrates predictive state updates in the [CopilotKit Feature Viewer](https://feature-viewer.copilotkit.ai/pydantic-ai/feature/shared_state). </Callout>A Pydantic AI Agent's state updates discontinuosly; only across function transitions in the flow. But even a single function in the flow often takes many seconds to run and contain sub-steps of interest to the user.
Agent-native applications reflect to the end-user what the agent is doing as continuously possible.
CopilotKit enables this through its concept of predictive state updates.
You can use this when you want to provide the user with feedback about what your agent is doing, specifically to:
When a function in your Pydantic AI agent finishes executing, its returned state becomes the single source of truth. While intermediate state updates are great for real-time feedback, any changes you want to persist must be explicitly included in the function's final returned state. Otherwise, they will be overwritten when the function completes.
Create your Pydantic AI agent with a stateful structure. Here's a complete example that tracks observed steps:
```python title="agent.py"
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.ag_ui import StateDeps
class AgentState(BaseModel):
"""State for the agent."""
observed_steps: list[str] = []
agent = Agent('openai:gpt-5.4-mini', deps_type=StateDeps[AgentState])
app = agent.to_ag_ui(deps=StateDeps(AgentState()))
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
```
</Step>
<Step>
### Emit the intermediate state
<TailoredContent
className="step"
id="state-emission"
header={
<div>
<p className="text-xl font-semibold">How would you like to emit state updates?</p>
<p className="text-base">
You can either manually emit state updates or configure specific tool calls to emit updates.
</p>
</div>
}
>
<TailoredContentOption
id="manual-emission"
title="Manual Predictive State Updates"
description="Manually emit state updates for maximum control over when updates occur."
icon={<FaArrowUp />}
>
For long-running tasks, you can emit state updates progressively as predictions of the final state. In this example, we create a tool that updates the observed steps:
```python title="agent.py"
from ag_ui.core import StateSnapshotEvent, EventType
from pydantic_ai import Agent
from pydantic_ai.ag_ui import StateDeps
@agent.tool_plain
async def update_steps(steps: list[str]) -> StateSnapshotEvent:
"""Update the steps of the agent."""
return StateSnapshotEvent(
type=EventType.STATE_SNAPSHOT,
snapshot={
"observed_steps": steps
}
)
```
</TailoredContentOption>
<TailoredContentOption
id="tool-emission"
title="Tool-Based Predictive State Updates"
description="Configure specific tool calls to automatically emit intermediate state updates."
icon={<FaWrench />}
>
For long-running tasks, you can configure CopilotKit to automatically predict state updates when specific tool calls are made. In this example, we'll configure CopilotKit to predict state updates whenever the LLM calls the step progress tool:
```python title="agent.py"
from ag_ui.core import CustomEvent, EventType
@agent.tool_plain
def enable_document_prediction() -> CustomEvent:
"""Enable document state prediction."""
return CustomEvent(
type=EventType.CUSTOM,
name='PredictState',
value=[{
'state_key': 'observed_steps',
'tool': 'update_steps',
'tool_argument': 'steps',
}]
)
```
</TailoredContentOption>
</TailoredContent>
</Step>
<Step>
### Observe the predictions
These predictions will be emitted as the agent runs, allowing you to track its progress before the final state is determined.
```tsx title="ui/app/page.tsx"
// ...
const YourMainContent = () => {
// Get access to both predicted and final states
const { agent } = useAgent({ agentId: "my_agent" });
// Add a state renderer to observe predictions
useAgent({
agentId: "my_agent",
render: ({ state }) => {
if (!state.observed_steps?.length) return null;
return (
<div>
<h3>Current Progress:</h3>
<ul>
{state.observed_steps.map((step, i) => (
<li key={i}>{step}</li>
))}
</ul>
</div>
);
},
});
return (
<div>
<h1>Agent Progress</h1>
{agent.state?.observed_steps?.length > 0 && (
<div>
<h3>Final Steps:</h3>
<ul>
{agent.state.observed_steps.map((step, i) => (
<li key={i}>{step}</li>
))}
</ul>
</div>
)}
</div>
)
}
```
<Callout type="warn" title="Important">
The `name` parameter must exactly match the agent name you defined in your CopilotRuntime configuration (e.g., `my_agent` from the quickstart).
</Callout>
</Step>
<Step>
### Give it a try!
Now you'll notice that the state predictions are emitted as the agent makes progress, giving you insight into its work before the final state is determined.
You can apply this pattern to any long-running task in your agent.
</Step>