State streaming

This example demonstrates predictive state updates in the CopilotKit Feature Viewer.

What is this?#

A LangGraph agent's state updates discontinuosly; only across node transitions in the graph. But even a single node in the graph 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.

When should I use this?#

You can use this when you want to provide the user with feedback about what your agent is doing, specifically to:

Keep users engaged by avoiding long loading indicators
Build trust by demonstrating what the agent is working on
Enable agent steering - allowing users to course-correct the agent if needed

Important Note#

When a node in your LangGraph 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 node's final returned state. Otherwise, they will be overwritten when the node completes.

Implementation#

Custom graph

I'm using a custom graph, where I define the nodes and edges myself.

Prebuilt agent

I'm using a prebuilt agent like "create_agent" by LangGraph

Install the CopilotKit SDK#

Any LangGraph agent can be used with CopilotKit. However, creating deep agentic experiences with CopilotKit requires our LangGraph SDK.

uv add copilotkit

poetry add copilotkit

pip install copilotkit --extra-index-url https://copilotkit.gateway.scarf.sh/simple/

conda install copilotkit -c copilotkit-channel

npm npm install @copilotkit/sdk-js

Define the state#

We'll be defining a observed_steps field in the state, which will be updated as the agent writes different sections of the report.

agent.py

from copilotkit import CopilotKitState
from typing import Literal

class AgentState(CopilotKitState):
    observed_steps: list[str]  # Array of completed steps

agent-js/src/agent.ts

import { StateSchema } from "@langchain/langgraph";
import { CopilotKitStateSchema } from "@copilotkit/sdk-js/langgraph";
import { z } from "zod";

export const AgentStateSchema = new StateSchema({
    observed_steps: z.array(z.string()).default(() => []),  // Array of completed steps
    ...CopilotKitStateSchema.fields,
});
export type AgentState = typeof AgentStateSchema.State;

Emit the intermediate state#

Observe the predictions#

These predictions will be emitted as the agent runs, allowing you to track its progress before the final state is determined.

ui/app/page.tsx

import { useAgent } from '@copilotkit/react-core'; 

const YourMainContent = () => {
    const { agent } = useAgent({
        agentId: "sample_agent",
    });

    const observedSteps = (agent.state.observed_steps as string[]) ?? [];

    return (
        <div>
            <h1>Agent Progress</h1>
            {observedSteps.length > 0 && (
                <div>
                    <h3>Steps:</h3>
                    <ul>
                        {observedSteps.map((step, i) => (
                            <li key={i}>{step}</li>
                        ))}
                    </ul>
                </div>
            )}
        </div>
    )
}

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.

This video shows the result of npx copilotkit@latest init with the implementation section applied to it!

import { StateSchema } from "@langchain/langgraph"; import { CopilotKitStateSchema } from "@copilotkit/sdk-js/langgraph"; import { z } from "zod"; export const AgentStateSchema = new StateSchema({ observed_steps: z.array(z.string()).default(() => []), // Array of completed steps ...CopilotKitStateSchema.fields, }); export type AgentState = typeof AgentStateSchema.State;

from copilotkit.langgraph import copilotkit_emit_state # ... async def chat_node(state: AgentState, config: RunnableConfig) -> Command[Literal["cpk_action_node", "tool_node", "__end__"]]: # ... # Simulate executing steps one by one steps = [ "Analyzing input data...", "Identifying key patterns...", "Generating recommendations...", "Formatting final output..." ] for step in steps: self.state["observed_steps"] = self.state.get("observed_steps", []) + [step] await copilotkit_emit_state(config, state) await asyncio.sleep(1) # ...

import { copilotkitEmitState } from "@copilotkit/sdk-js/langgraph"; // ... async function chat_node(state: AgentState, config: RunnableConfig) { // ... // Simulate executing steps one by one const steps = [ "Analyzing input data...", "Identifying key patterns...", "Generating recommendations...", "Formatting final output..." ]; for (const step of steps) { state.observed_steps = [...(state.observed_steps ?? []), step]; copilotkitEmitState(config, state); await new Promise(resolve => setTimeout(resolve, 1000)); } }

import { useAgent } from '@copilotkit/react-core'; const YourMainContent = () => { const { agent } = useAgent({ agentId: "sample_agent", }); const observedSteps = (agent.state.observed_steps as string[]) ?? []; return ( <div> <h1>Agent Progress</h1> {observedSteps.length > 0 && ( <div> <h3>Steps:</h3> <ul> {observedSteps.map((step, i) => ( <li key={i}>{step}</li> ))} </ul> </div> )} </div> ) }

State streaming

What is this?#

When should I use this?#

Important Note#

Implementation#

Install the CopilotKit SDK#

Define the state#

Emit the intermediate state#

Observe the predictions#

Give it a try!#

On this page

Install the CopilotKit SDK#

Define the state#

Emit the intermediate state#

Observe the predictions#

Give it a try!#