AG-UI を使用した状態管理

AG-UI 状態は、実行に関連付けられた、クライアントに可視な JSON です。 .NETでは、統合には 2 つの明示的なメカニズムが用意されています。

• 元の RunAgentInputからクライアントによって提供される読み取り状態。
• 選択したツール呼び出しまたは結果を、 AGUIStreamOptionsを使用して状態イベントを AG-UI にマップします。

状態マッピングはオプトインです。 任意のツールの結果が自動的に共有状態になることはありません。

クライアントの状態の読み取り

MapAGUIServer は、元の RunAgentInput を ChatOptionsに格納します。 委任エージェントまたはチャット クライアント ミドルウェアは、 TryGetRunAgentInputを使用して復旧できます。

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

static bool TryGetClientState(ChatOptions options, out JsonElement state)
{
    if (options.TryGetRunAgentInput(out RunAgentInput? input) &&
        input.State is { ValueKind: not JsonValueKind.Undefined } value)
    {
        state = value;
        return true;
    }

    state = default;
    return false;
}

クライアントの状態は要求入力です。 プロンプト、ルーティング、または特権操作で使用する前に、その図形と値を検証します。

状態スナップショットを出力する

ツールが完全な状態を返したときに、ツールの結果を 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 の結果がスナップショットになり、クライアントの現在の共有状態が置き換えられます。

その他の結果の種類の場合は、StateSnapshotEventを構築するカスタム マッパーでMapResultを使用します。

状態デルタを出力する

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 Patch 配列が 含まれている必要があります。 ツールが別の表現を返す場合は、カスタム マッパーで 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引数にマップします。 エージェントがツールを呼び出す際、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."

State Management を使用してエージェントを作成する

状態管理を使用した完全なサーバー実装を次に示します。

"""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"
            ]
        }
    }
}

State Delta イベント

LLM ストリーム ツールの引数として出力される、JSON パッチ形式を使用した増分状態更新:

{
    "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"]
            }
        }
    ]
}

クライアントの実装

agent_framework_ag_ui パッケージは、AG-UI サーバーに接続するためのAGUIChatClientを提供し、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 の更新をリアルタイムで表示し、ユーザーに即座にフィードバックを提供できます。

Human-in-the-Loop を使用した状態

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. ツール引数をエージェントが生成する際に (STATE_DELTA イベントを通じた予測更新によって)、状態がストリームを更新します。
2. エージェントが、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 状態フィールドにマップされます。

ベスト プラクティス

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
    """,
    ...
)

確認 UI のカスタマイズ

サーバーから確認イベントをレンダリングするときに、AG-UI クライアントの承認メッセージと状態確認メッセージをカスタマイズします。

次のステップ

これで、すべてのコア AG-UI 機能について学習しました。 次に、次のことができます。

• Agent Framework のドキュメントを確認する
• すべての 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},
    },
})