AG-UI 狀態管理

AG-UI 狀態是與一次執行相關聯、可供用戶端查看的 JSON。 在 .NET 中，整合提供了兩種明確機制：

• 讀取由用戶端從來源 RunAgentInput 提供的狀態。
• 將選取的工具呼叫或結果映射為 AG-UI 狀態事件。AGUIStreamOptions

狀態對應需主動選擇加入。 任意工具的結果不會自動變成共享狀態。

閱讀客戶端狀態

MapAGUIServer 儲存起源 RunAgentInput 於 ChatOptions。 如果模型需要客戶端的目前狀態，可以用輕量級的 DelegatingAIAgent 包裝基礎代理，透過 TryGetRunAgentInput 恢復該狀態，並將其加入模型內容：

using System.Text.Json;
using AGUI.Abstractions;
using AGUI.Server;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

internal sealed class RecipeStateAgent(AIAgent innerAgent)
    : DelegatingAIAgent(innerAgent)
{
    protected override Task<AgentResponse> RunCoreAsync(
        IEnumerable<ChatMessage> messages,
        AgentSession? session = null,
        AgentRunOptions? options = null,
        CancellationToken cancellationToken = default) =>
        RunCoreStreamingAsync(messages, session, options, cancellationToken)
            .ToAgentResponseAsync(cancellationToken);

    protected override IAsyncEnumerable<AgentResponseUpdate> RunCoreStreamingAsync(
        IEnumerable<ChatMessage> messages,
        AgentSession? session = null,
        AgentRunOptions? options = null,
        CancellationToken cancellationToken = default)
    {
        if (options is ChatClientAgentRunOptions { ChatOptions: { } chatOptions } &&
            chatOptions.TryGetRunAgentInput(out RunAgentInput? input) &&
            input.State is { ValueKind: JsonValueKind.Object } state)
        {
            ChatMessage stateMessage = new(
                ChatRole.System,
                $"The user's current recipe state is:\n{state.GetRawText()}");
            messages = [stateMessage, .. messages];
        }

        return InnerAgent.RunStreamingAsync(
            messages,
            session,
            options,
            cancellationToken);
    }
}

AIAgent agent = new RecipeStateAgent(baseAgent);

包裝器只處理輸入路徑。 狀態事件仍可透過 AGUIStreamOptions 以宣告方式發出，如以下各節所示。 TryGetRunAgentInput 讀取主機層儲存的 ChatOptions.AdditionalProperties輸入;應用程式碼不會直接存取該字典。

客戶端狀態屬於不可信任的請求輸入。 在提示、路由或特權操作中使用前，先驗證其形狀與值。

發出狀態快照

當工具回傳完整狀態時，將工具結果對應至 STATE_SNAPSHOT：

using AGUI.Server;

AGUIStreamOptions streamOptions = new AGUIStreamOptions()
    .MapResultAsStateSnapshot("generate_recipe");

app.MapAGUIServer("/", agent).WithMetadata(streamOptions);

MapResultAsStateSnapshot 需要 值 FunctionResultContent.Result 為 JsonElement。 在將其傳回之前，先在工具中將 POCO、字典或集合序列化為 JsonElement。 generate_recipe結果即為快照，取代用戶端目前的共享狀態。

對於其他結果類型，請使用MapResult配合自訂對應工具來建構StateSnapshotEvent。

發送狀態增量

當其傳回 RFC 6902 JSON 修補程式時，將工具結果對應至STATE_DELTA：

AGUIStreamOptions streamOptions = new AGUIStreamOptions()
    .MapResultAsStateSnapshot("create_plan")
    .MapResultAsStateDelta("update_plan_step");

app.MapAGUIServer("/", agent).WithMetadata(streamOptions);

對於增量變更，請使用快照來初始化或替換狀態和變化量。

MapResultAsStateDelta 也需要一個 JsonElement 結果。 該元素必須包含 RFC 6902 JSON 補丁 陣列。 如果工具回傳另一個表示法，就用 MapResult 自訂映射器。

將工具呼叫對應到狀態

AGUIStreamOptions.MapCall會將所選的FunctionCallContent對應到在一般工具呼叫事件後所發出的其他 AG-UI 事件。 當狀態從工具參數而非工具結果衍生時，請使用此方法：

AGUIStreamOptions streamOptions = new AGUIStreamOptions()
    .MapCall("write_document", call =>
    {
        if (call.Arguments?.TryGetValue("document", out object? document) is not true)
        {
            return [];
        }

        JsonElement snapshot = JsonSerializer.SerializeToElement(new { document });
        return [new StateSnapshotEvent { Snapshot = snapshot }];
    });

app.MapAGUIServer("/", agent).WithMetadata(streamOptions);

應用程式會定義對應關係和狀態結構。 MapCall 不會從任意工具參數推斷狀態，也不會抑制正常工具執行。 增量更新需要底層模型客戶端揭露串流的工具呼叫參數，並由應用程式配置相應的參數擷取。

在 .NET 用戶端接收狀態

AG-UI .NET 用戶端透過以下方式呈現狀態協定事件ChatResponseUpdate.RawRepresentation：

await foreach (AgentResponseUpdate update in agent.RunStreamingAsync(messages, session))
{
    if (update.AsChatResponseUpdate().RawRepresentation is StateSnapshotEvent snapshot)
    {
        JsonElement state = snapshot.Snapshot;
    }
    else if (update.AsChatResponseUpdate().RawRepresentation is StateDeltaEvent delta)
    {
        JsonElement changes = delta.Delta;
    }
}

用戶端負責保留並套用共享狀態，並在應用程式需要時在後續請求時傳送當前狀態。

下一步

定義狀態模型

首先，為您的狀態結構定義 Pydantic 模型。 這可確保類型安全性和驗證：

from enum import Enum
from pydantic import BaseModel, Field


class SkillLevel(str, Enum):
    """The skill level required for the recipe."""
    BEGINNER = "Beginner"
    INTERMEDIATE = "Intermediate"
    ADVANCED = "Advanced"


class CookingTime(str, Enum):
    """The cooking time of the recipe."""
    FIVE_MIN = "5 min"
    FIFTEEN_MIN = "15 min"
    THIRTY_MIN = "30 min"
    FORTY_FIVE_MIN = "45 min"
    SIXTY_PLUS_MIN = "60+ min"


class Ingredient(BaseModel):
    """An ingredient with its details."""
    icon: str = Field(..., description="Emoji icon representing the ingredient (e.g., 🥕)")
    name: str = Field(..., description="Name of the ingredient")
    amount: str = Field(..., description="Amount or quantity of the ingredient")


class Recipe(BaseModel):
    """A complete recipe."""
    title: str = Field(..., description="The title of the recipe")
    skill_level: SkillLevel = Field(..., description="The skill level required")
    special_preferences: list[str] = Field(
        default_factory=list, description="Dietary preferences (e.g., Vegetarian, Gluten-free)"
    )
    cooking_time: CookingTime = Field(..., description="The estimated cooking time")
    ingredients: list[Ingredient] = Field(..., description="Complete list of ingredients")
    instructions: list[str] = Field(..., description="Step-by-step cooking instructions")

狀態架構

定義狀態結構描述以指定狀態的結構和類型：

state_schema = {
    "recipe": {"type": "object", "description": "The current recipe"},
}

預測狀態更新

預測狀態會在 LLM 產生時將串流工具引數更新為狀態，進而實現樂觀 UI 更新：

predict_state_config = {
    "recipe": {"tool": "update_recipe", "tool_argument": "recipe"},
}

此配置將 recipe 狀態欄位對應至 recipe 工具的 update_recipe 參數。 當 Agent 呼叫工具時，引數會在 LLM 產生時即時串流到狀態。

定義狀態更新工具

建立一個接受 Pydantic 模型的工具函數：

from agent_framework import tool


@tool
def update_recipe(recipe: Recipe) -> str:
    """Update the recipe with new or modified content.

    You MUST write the complete recipe with ALL fields, even when changing only a few items.
    When modifying an existing recipe, include ALL existing ingredients and instructions plus your changes.
    NEVER delete existing data - only add or modify.

    Args:
        recipe: The complete recipe object with all details

    Returns:
        Confirmation that the recipe was updated
    """
    return "Recipe updated."

使用狀態管理建立代理程式

以下是具有狀態管理的完整伺服器實作：

"""AG-UI server with state management."""

from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework_ag_ui import (
    AgentFrameworkAgent,
    add_agent_framework_fastapi_endpoint,
)
from azure.identity import AzureCliCredential
from fastapi import FastAPI

# Create the chat agent with tools
agent = Agent(
    name="recipe_agent",
    instructions="""You are a helpful recipe assistant that creates and modifies recipes.

    CRITICAL RULES:
    1. You will receive the current recipe state in the system context
    2. To update the recipe, you MUST use the update_recipe tool
    3. When modifying a recipe, ALWAYS include ALL existing data plus your changes in the tool call
    4. NEVER delete existing ingredients or instructions - only add or modify
    5. After calling the tool, provide a brief conversational message (1-2 sentences)

    When creating a NEW recipe:
    - Provide all required fields: title, skill_level, cooking_time, ingredients, instructions
    - Use actual emojis for ingredient icons (🥕 🧄 🧅 🍅 🌿 🍗 🥩 🧀)
    - Leave special_preferences empty unless specified
    - Message: "Here's your recipe!" or similar

    When MODIFYING or IMPROVING an existing recipe:
    - Include ALL existing ingredients + any new ones
    - Include ALL existing instructions + any new/modified ones
    - Update other fields as needed
    - Message: Explain what you improved (e.g., "I upgraded the ingredients to premium quality")
    - When asked to "improve", enhance with:
      * Better ingredients (upgrade quality, add complementary flavors)
      * More detailed instructions
      * Professional techniques
      * Adjust skill_level if complexity changes
      * Add relevant special_preferences

    Example improvements:
    - Upgrade "chicken" → "organic free-range chicken breast"
    - Add herbs: basil, oregano, thyme
    - Add aromatics: garlic, shallots
    - Add finishing touches: lemon zest, fresh parsley
    - Make instructions more detailed and professional
    """,
    client=OpenAIChatCompletionClient(
        model=deployment_name,
        azure_endpoint=endpoint,
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        credential=AzureCliCredential(),
    ),
    tools=[update_recipe],
)

# Wrap agent with state management
recipe_agent = AgentFrameworkAgent(
    agent=agent,
    name="RecipeAgent",
    description="Creates and modifies recipes with streaming state updates",
    state_schema={
        "recipe": {"type": "object", "description": "The current recipe"},
    },
    predict_state_config={
        "recipe": {"tool": "update_recipe", "tool_argument": "recipe"},
    },
)

# Create FastAPI app
app = FastAPI(title="AG-UI Recipe Assistant")
add_agent_framework_fastapi_endpoint(app, recipe_agent, "/")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8888)

關鍵概念

• Pydantic 模型：定義具有類型安全和驗證的結構化狀態
• 狀態結構描述：指定狀態欄位類型的簡單格式
• 預測狀態配置：將狀態欄位映射至用於流更新的工具參數
• 狀態注入：當前狀態會自動注入為系統訊息以提供上下文
• 完整更新：工具必須寫入完整的狀態，而不僅是變化量
• 確認策略：為您的網域自訂批准訊息（食譜、文件、任務計劃等）

了解狀態事件

狀態快照事件

目前狀態的完整快照集，在工具完成時發出：

{
    "type": "STATE_SNAPSHOT",
    "snapshot": {
        "recipe": {
            "title": "Classic Pasta Carbonara",
            "skill_level": "Intermediate",
            "special_preferences": ["Authentic Italian"],
            "cooking_time": "30 min",
            "ingredients": [
                {"icon": "🍝", "name": "Spaghetti", "amount": "400g"},
                {"icon": "🥓", "name": "Guanciale or bacon", "amount": "200g"},
                {"icon": "🥚", "name": "Egg yolks", "amount": "4"},
                {"icon": "🧀", "name": "Pecorino Romano", "amount": "100g grated"},
                {"icon": "🧂", "name": "Black pepper", "amount": "To taste"}
            ],
            "instructions": [
                "Bring a large pot of salted water to boil",
                "Cut guanciale into small strips and fry until crispy",
                "Beat egg yolks with grated Pecorino and black pepper",
                "Cook spaghetti until al dente",
                "Reserve 1 cup pasta water, then drain pasta",
                "Remove pan from heat, add hot pasta to guanciale",
                "Quickly stir in egg mixture, adding pasta water to create creamy sauce",
                "Serve immediately with extra Pecorino and black pepper"
            ]
        }
    }
}

狀態變化量是件

使用 JSON 補丁格式進行增量狀態更新，並以 LLM 串流工具參數的形式生成。

{
    "type": "STATE_DELTA",
    "delta": [
        {
            "op": "replace",
            "path": "/recipe",
            "value": {
                "title": "Classic Pasta Carbonara",
                "skill_level": "Intermediate",
                "cooking_time": "30 min",
                "ingredients": [
                    {"icon": "🍝", "name": "Spaghetti", "amount": "400g"}
                ],
                "instructions": ["Bring a large pot of salted water to boil"]
            }
        }
    ]
}

傳送客戶端狀態

用state_carrier()來從 Python AGUIChatClient傳送共享狀態。 將承運商放在自己的使用者訊息中，並附上使用者提示發送。 客戶端會將最新的電信業者移至 AG-UI 請求的 state 欄位中，並且不會將該電信業者作為聊天訊息傳送。

from agent_framework import Message
from agent_framework_ag_ui import state_carrier

messages = [
    Message(role="user", contents=["Update the sales dashboard."]),
    Message(
        role="user",
        contents=[state_carrier({"selected_tab": "sales"})],
    ),
]

await agent.run(messages, session=thread)

普通 application/json 內容仍是文件輸入。 只有當你用 來 state_carrier()建立它時，它才會成為狀態。

用戶端實作

該 agent_framework_ag_ui 套件提供 AGUIChatClient 連線到 AG-UI 伺服器，使 Python 用戶端體驗與 .NET 對等：

"""AG-UI client with state management."""

import asyncio
import json
import os
from typing import Any

from agent_framework import Agent, Message, Role
from agent_framework_ag_ui import AGUIChatClient


async def main():
    """Example client with state tracking."""
    server_url = os.environ.get("AGUI_SERVER_URL", "http://127.0.0.1:8888/")
    print(f"Connecting to AG-UI server at: {server_url}\n")

    # Create AG-UI chat client
    chat_client = AGUIChatClient(endpoint=server_url)

    # Wrap with Agent for convenient API
    agent = Agent(
        name="ClientAgent",
        client=chat_client,
        instructions="You are a helpful assistant.",
    )

    # Get a thread for conversation continuity
    thread = agent.create_session()

    # Track state locally
    state: dict[str, Any] = {}

    try:
        while True:
            message = input("\nUser (:q to quit, :state to show state): ")
            if not message.strip():
                continue

            if message.lower() in (":q", "quit"):
                break

            if message.lower() == ":state":
                print(f"\nCurrent state: {json.dumps(state, indent=2)}")
                continue

            print()
            # Stream the agent response with state
            async for update in agent.run(message, session=thread, stream=True):
                # Handle text content
                if update.text:
                    print(update.text, end="", flush=True)

                # Handle state updates surfaced through AG-UI events.
                for content in update.contents:
                    if content.type == "data" and getattr(content, "media_type", None) == "application/json":
                        print("\n[JSON state payload received]")

            print(f"\n\nCurrent state: {json.dumps(state, indent=2)}")
            print()

    except KeyboardInterrupt:
        print("\n\nExiting...")


if __name__ == "__main__":
    # Install dependencies: pip install agent-framework-ag-ui --pre
    asyncio.run(main())

主要優點

AGUIChatClient 提供：

• 簡化連接：自動處理 HTTP/SSE 通訊
• 線程管理： 內置線程 ID 追蹤，確保對話連續性
• 代理整合：無縫協作以配合熟悉的 API Agent
• 狀態處理：自動解析來自伺服器的狀態事件
• 與 .NET 一致性：跨語言的一致體驗

確認預測狀態

當預測的狀態變更應在套用前先等待用戶端確認時，請在 require_confirmation=True 上設定 AgentFrameworkAgent：

recipe_agent = AgentFrameworkAgent(
    agent=agent,
    state_schema={"recipe": {"type": "object", "description": "The current recipe"}},
    predict_state_config={"recipe": {"tool": "update_recipe", "tool_argument": "recipe"}},
    require_confirmation=True,
)

在 AG-UI 用戶端 UI 中，可在轉譯確認事件時自訂確認文案。

互動範例

當伺服器和用戶端執行時：

User (:q to quit, :state to show state): I want to make a classic Italian pasta carbonara

[Run Started]
[Calling Tool: update_recipe]
[State Updated]
[State Updated]
[State Updated]
[Tool Result: Recipe updated.]
Here's your recipe!
[Run Finished]

============================================================
CURRENT STATE
============================================================

recipe:
  title: Classic Pasta Carbonara
  skill_level: Intermediate
  special_preferences: ['Authentic Italian']
  cooking_time: 30 min
  ingredients:
    - 🍝 Spaghetti: 400g
    - 🥓 Guanciale or bacon: 200g
    - 🥚 Egg yolks: 4
    - 🧀 Pecorino Romano: 100g grated
    - 🧂 Black pepper: To taste
  instructions:
    1. Bring a large pot of salted water to boil
    2. Cut guanciale into small strips and fry until crispy
    3. Beat egg yolks with grated Pecorino and black pepper
    4. Cook spaghetti until al dente
    5. Reserve 1 cup pasta water, then drain pasta
    6. Remove pan from heat, add hot pasta to guanciale
    7. Quickly stir in egg mixture, adding pasta water to create creamy sauce
    8. Serve immediately with extra Pecorino and black pepper

============================================================

預測狀態更新的實際應用

搭配predict_state_config使用預測性狀態更新時，用戶端會在 LLM 即時產生工具引數，並在工具執行之前，接收到STATE_DELTA事件：

// Agent starts generating tool call for update_recipe
// Client receives STATE_DELTA events as the recipe argument streams:

// First delta - partial recipe with title
{
  "type": "STATE_DELTA",
  "delta": [{"op": "replace", "path": "/recipe", "value": {"title": "Classic Pasta"}}]
}

// Second delta - title complete with more fields
{
  "type": "STATE_DELTA",
  "delta": [{"op": "replace", "path": "/recipe", "value": {
    "title": "Classic Pasta Carbonara",
    "skill_level": "Intermediate"
  }}]
}

// Third delta - ingredients starting to appear
{
  "type": "STATE_DELTA",
  "delta": [{"op": "replace", "path": "/recipe", "value": {
    "title": "Classic Pasta Carbonara",
    "skill_level": "Intermediate",
    "cooking_time": "30 min",
    "ingredients": [
      {"icon": "🍝", "name": "Spaghetti", "amount": "400g"}
    ]
  }}]
}

// ... more deltas as the LLM generates the complete recipe

這使客戶能夠在客服專員思考時即時顯示樂觀的 UI 更新，從而向使用者提供即時回饋。

具有人工介入的狀態

您可以透過設定 require_confirmation=True以下方式將狀態管理與核准工作流程結合：

recipe_agent = AgentFrameworkAgent(
    agent=agent,
    state_schema={"recipe": {"type": "object", "description": "The current recipe"}},
    predict_state_config={"recipe": {"tool": "update_recipe", "tool_argument": "recipe"}},
    require_confirmation=True,  # Require approval for state changes
)

啟用時：

1. 當 Agent 產生工具引數時，狀態會更新串流 (透過STATE_DELTA事件的預測性更新)
2. Agent 在RUN_FINISHED.outcome.interrupts中以tool_call中斷方式於執行工具前暫停
3. 如果核准，工具會執行並發出最終狀態 （透過 STATE_SNAPSHOT 事件）
4. 如果被拒絕，則會捨棄預測狀態的變更

進階狀態模式

具有多個欄位的複雜狀態

您可以使用不同的工具管理多個狀態欄位：

from pydantic import BaseModel


class TaskStep(BaseModel):
    """A single task step."""
    description: str
    status: str = "pending"
    estimated_duration: str = "5 min"


@tool
def generate_task_steps(steps: list[TaskStep]) -> str:
    """Generate task steps for a given task."""
    return f"Generated {len(steps)} steps."


@tool
def update_preferences(preferences: dict[str, Any]) -> str:
    """Update user preferences."""
    return "Preferences updated."


# Configure with multiple state fields
agent_with_multiple_state = AgentFrameworkAgent(
    agent=agent,
    state_schema={
        "steps": {"type": "array", "description": "List of task steps"},
        "preferences": {"type": "object", "description": "User preferences"},
    },
    predict_state_config={
        "steps": {"tool": "generate_task_steps", "tool_argument": "steps"},
        "preferences": {"tool": "update_preferences", "tool_argument": "preferences"},
    },
)

使用萬用字元工具引數

當工具返回複雜的巢狀資料時，請使用 "*" 將所有工具參數對應到狀態：

@tool
def create_document(title: str, content: str, metadata: dict[str, Any]) -> str:
    """Create a document with title, content, and metadata."""
    return "Document created."


# Map all tool arguments to document state
predict_state_config = {
    "document": {"tool": "create_document", "tool_argument": "*"}
}

這會將整個工具呼叫 （所有引數） 對應至 document state 欄位。

最佳做法

使用 Pydantic 模型

定義類型安全的結構化模型：

class Recipe(BaseModel):
    """Use Pydantic models for structured, validated state."""
    title: str
    skill_level: SkillLevel
    ingredients: list[Ingredient]
    instructions: list[str]

優點：

• 類型安全：自動驗證資料類型
• 文件：欄位描述可作為文件
• IDE 支援：自動完成和類型檢查
• 序列化：自動 JSON 轉換

狀態更新完成

一律寫入完整狀態，而不僅是變化量：

@tool
def update_recipe(recipe: Recipe) -> str:
    """
    You MUST write the complete recipe with ALL fields.
    When modifying a recipe, include ALL existing ingredients and
    instructions plus your changes. NEVER delete existing data.
    """
    return "Recipe updated."

這可確保狀態一致性和適當的預測更新。

比對參數名稱

確保工具參數名稱符合 tool_argument 配置：

# Tool parameter name
def update_recipe(recipe: Recipe) -> str:  # Parameter name: 'recipe'
    ...

# Must match in predict_state_config
predict_state_config = {
    "recipe": {"tool": "update_recipe", "tool_argument": "recipe"}  # Same name
}

在說明中提供上下文

包括有關狀態管理的明確說明：

agent = Agent(
    instructions="""
    CRITICAL RULES:
    1. You will receive the current recipe state in the system context
    2. To update the recipe, you MUST use the update_recipe tool
    3. When modifying a recipe, ALWAYS include ALL existing data plus your changes
    4. NEVER delete existing ingredients or instructions - only add or modify
    """,
    ...
)

自訂確認介面

在您的 AG-UI 用戶端中，可於轉譯來自伺服器的確認事件時，自訂核准與狀態確認訊息。

後續步驟

您現在已經了解了所有核心 AG-UI 功能！ 接下來您可以：

• 探索 代理程式架構文件
• 構建一個結合所有 AG-UI 功能的完整應用程序
• 將您的 AG-UI 服務部署至生產環境

其他資源

• AG-UI 概覽
• 入門
• 後端工具渲染

Go 的 AG-UI 狀態管理可以透過中介軟體來實作，該中介軟體會在發送一般文字更新的同時，一併發出結構化的 message.DataContent 更新。

stateSnapshotMiddleware := agent.MiddlewareFunc(func(next agent.RunFunc, ctx context.Context, messages []*message.Message, opts ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
    return func(yield func(*agent.ResponseUpdate, error) bool) {
        for update, err := range next(ctx, messages, opts...) {
            if err != nil {
                yield(nil, err)
                return
            }
            if update != nil {
                // Inspect update contents and yield DataContent snapshots as needed.
            }
            if !yield(update, nil) {
                return
            }
        }
    }
})

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Middlewares: []agent.Middleware{stateSnapshotMiddleware},
    },
})