智能体用户交互 (AG-UI)
结合使用 Pydantic AI 智能体和 AG-UI Dojo 示例应用的示例。
有关 AG-UI 集成的更多信息，请参阅 AG-UI 文档。
演示内容：
先决条件
运行示例
安装好依赖项并设置好环境变量后，您将需要两个命令行窗口。
Pydantic AI AG-UI 后端
设置您的 OpenAI API 密钥
export OPENAI_API_KEY=<your api key>
启动 Pydantic AI AG-UI 示例后端。
python -m pydantic_ai_examples.ag_ui
uv run -m pydantic_ai_examples.ag_ui
AG-UI Dojo 示例前端
接下来运行 AG-UI Dojo 示例前端。
-
克隆 AG-UI 代码仓库
git clone https://github.com/ag-ui-protocol/ag-ui.git -
切换到
ag-ui/typescript-sdk目录cd ag-ui/typescript-sdk -
按照官方说明运行 Dojo 应用
- 访问 https://:3000/pydantic-ai
- 从侧边栏选择查看
Pydantic AI
功能示例
智能体聊天
这演示了一个基本的智能体交互，包括 Pydantic AI 服务器端工具和 AG-UI 客户端工具。
如果您已经运行了示例，可以在 https://:3000/pydantic-ai/feature/agentic_chat 查看。
智能体工具
time- 用于检查特定时区当前时间的 Pydantic AI 工具background- 用于设置客户端窗口背景颜色的 AG-UI 工具
智能体提示
What is the time in New York?
Change the background to blue
一个混合了 AG-UI 和 Pydantic AI 工具的复杂示例
Perform the following steps, waiting for the response of each step before continuing:
1. Get the time
2. Set the background to red
3. Get the time
4. Report how long the background set took by diffing the two times
智能体聊天 - 代码
ag_ui/api/agentic_chat.py
"""Agentic Chat feature."""
from __future__ import annotations
from datetime import datetime
from zoneinfo import ZoneInfo
from pydantic_ai import Agent
agent = Agent('openai:gpt-4o-mini')
app = agent.to_ag_ui()
@agent.tool_plain
async def current_time(timezone: str = 'UTC') -> str:
"""Get the current time in ISO format.
Args:
timezone: The timezone to use.
Returns:
The current time in ISO format string.
"""
tz: ZoneInfo = ZoneInfo(timezone)
return datetime.now(tz=tz).isoformat()
智能体生成式 UI
演示一个长时间运行的任务，其中智能体向前端发送更新，让用户了解正在发生的事情。
如果您已经运行了示例，可以在 https://:3000/pydantic-ai/feature/agentic_generative_ui 查看。
计划提示
Create a plan for breakfast and execute it
智能体生成式 UI - 代码
ag_ui/api/agentic_generative_ui.py
"""Agentic Generative UI feature."""
from __future__ import annotations
from textwrap import dedent
from typing import Any, Literal
from pydantic import BaseModel, Field
from ag_ui.core import EventType, StateDeltaEvent, StateSnapshotEvent
from pydantic_ai import Agent
StepStatus = Literal['pending', 'completed']
class Step(BaseModel):
"""Represents a step in a plan."""
description: str = Field(description='The description of the step')
status: StepStatus = Field(
default='pending',
description='The status of the step (e.g., pending, completed)',
)
class Plan(BaseModel):
"""Represents a plan with multiple steps."""
steps: list[Step] = Field(default_factory=list, description='The steps in the plan')
class JSONPatchOp(BaseModel):
"""A class representing a JSON Patch operation (RFC 6902)."""
op: Literal['add', 'remove', 'replace', 'move', 'copy', 'test'] = Field(
description='The operation to perform: add, remove, replace, move, copy, or test',
)
path: str = Field(description='JSON Pointer (RFC 6901) to the target location')
value: Any = Field(
default=None,
description='The value to apply (for add, replace operations)',
)
from_: str | None = Field(
default=None,
alias='from',
description='Source path (for move, copy operations)',
)
agent = Agent(
'openai:gpt-4o-mini',
instructions=dedent(
"""
When planning use tools only, without any other messages.
IMPORTANT:
- Use the `create_plan` tool to set the initial state of the steps
- Use the `update_plan_step` tool to update the status of each step
- Do NOT repeat the plan or summarise it in a message
- Do NOT confirm the creation or updates in a message
- Do NOT ask the user for additional information or next steps
Only one plan can be active at a time, so do not call the `create_plan` tool
again until all the steps in current plan are completed.
"""
),
)
@agent.tool_plain
async def create_plan(steps: list[str]) -> StateSnapshotEvent:
"""Create a plan with multiple steps.
Args:
steps: List of step descriptions to create the plan.
Returns:
StateSnapshotEvent containing the initial state of the steps.
"""
plan: Plan = Plan(
steps=[Step(description=step) for step in steps],
)
return StateSnapshotEvent(
type=EventType.STATE_SNAPSHOT,
snapshot=plan.model_dump(),
)
@agent.tool_plain
async def update_plan_step(
index: int, description: str | None = None, status: StepStatus | None = None
) -> StateDeltaEvent:
"""Update the plan with new steps or changes.
Args:
index: The index of the step to update.
description: The new description for the step.
status: The new status for the step.
Returns:
StateDeltaEvent containing the changes made to the plan.
"""
changes: list[JSONPatchOp] = []
if description is not None:
changes.append(
JSONPatchOp(
op='replace', path=f'/steps/{index}/description', value=description
)
)
if status is not None:
changes.append(
JSONPatchOp(op='replace', path=f'/steps/{index}/status', value=status)
)
return StateDeltaEvent(
type=EventType.STATE_DELTA,
delta=changes,
)
app = agent.to_ag_ui()
人在回路
演示一个简单的人在回路工作流，其中智能体提出一个计划，用户可以使用复选框来批准它。
任务规划工具
generate_task_steps- 用于生成和确认步骤的 AG-UI 工具
任务规划提示
Generate a list of steps for cleaning a car for me to review
人在回路 - 代码
ag_ui/api/human_in_the_loop.py
"""Human in the Loop Feature.
No special handling is required for this feature.
"""
from __future__ import annotations
from textwrap import dedent
from pydantic_ai import Agent
agent = Agent(
'openai:gpt-4o-mini',
instructions=dedent(
"""
When planning tasks use tools only, without any other messages.
IMPORTANT:
- Use the `generate_task_steps` tool to display the suggested steps to the user
- Never repeat the plan, or send a message detailing steps
- If accepted, confirm the creation of the plan and the number of selected (enabled) steps only
- If not accepted, ask the user for more information, DO NOT use the `generate_task_steps` tool again
"""
),
)
app = agent.to_ag_ui()
预测性状态更新
演示如何使用预测性状态更新功能，根据智能体的响应来更新 UI 状态，包括通过用户确认实现的用户交互。
如果您已经运行了示例，可以在 https://:3000/pydantic-ai/feature/predictive_state_updates 查看。
故事工具
write_document- 用于将文档写入窗口的 AG-UI 工具document_predict_state- 为write_document工具启用文档状态预测的 Pydantic AI 工具
这也展示了如何根据共享状态信息使用自定义指令。
故事示例
起始文档文本
Bruce was a good dog,
智能体提示
Help me complete my story about bruce the dog, is should be no longer than a sentence.
预测性状态更新 - 代码
ag_ui/api/predictive_state_updates.py
"""Predictive State feature."""
from __future__ import annotations
from textwrap import dedent
from pydantic import BaseModel
from ag_ui.core import CustomEvent, EventType
from pydantic_ai import Agent, RunContext
from pydantic_ai.ag_ui import StateDeps
class DocumentState(BaseModel):
"""State for the document being written."""
document: str = ''
agent = Agent('openai:gpt-4o-mini', deps_type=StateDeps[DocumentState])
# Tools which return AG-UI events will be sent to the client as part of the
# event stream, single events and iterables of events are supported.
@agent.tool_plain
async def document_predict_state() -> list[CustomEvent]:
"""Enable document state prediction.
Returns:
CustomEvent containing the event to enable state prediction.
"""
return [
CustomEvent(
type=EventType.CUSTOM,
name='PredictState',
value=[
{
'state_key': 'document',
'tool': 'write_document',
'tool_argument': 'document',
},
],
),
]
@agent.instructions()
async def story_instructions(ctx: RunContext[StateDeps[DocumentState]]) -> str:
"""Provide instructions for writing document if present.
Args:
ctx: The run context containing document state information.
Returns:
Instructions string for the document writing agent.
"""
return dedent(
f"""You are a helpful assistant for writing documents.
Before you start writing, you MUST call the `document_predict_state`
tool to enable state prediction.
To present the document to the user for review, you MUST use the
`write_document` tool.
When you have written the document, DO NOT repeat it as a message.
If accepted briefly summarize the changes you made, 2 sentences
max, otherwise ask the user to clarify what they want to change.
This is the current document:
{ctx.deps.state.document}
"""
)
app = agent.to_ag_ui(deps=StateDeps(DocumentState()))
共享状态
演示如何在 UI 和智能体之间使用共享状态。
发送给智能体的状态由一个基于函数的指令检测到。然后，该指令使用自定义的 Pydantic 模型验证数据，之后用它来创建要遵循的指令，并通过一个 AG-UI 工具发送给客户端。
如果您已经运行了示例，可以在 https://:3000/pydantic-ai/feature/shared_state 查看。
食谱工具
display_recipe- 用于以图形格式显示食谱的 AG-UI 工具
食谱示例
- 自定义您的食谱基本设置
- 点击
用 AI 改进
共享状态 - 代码
ag_ui/api/shared_state.py
"""Shared State feature."""
from __future__ import annotations
from enum import StrEnum
from textwrap import dedent
from pydantic import BaseModel, Field
from ag_ui.core import EventType, StateSnapshotEvent
from pydantic_ai import Agent, RunContext
from pydantic_ai.ag_ui import StateDeps
class SkillLevel(StrEnum):
"""The level of skill required for the recipe."""
BEGINNER = 'Beginner'
INTERMEDIATE = 'Intermediate'
ADVANCED = 'Advanced'
class SpecialPreferences(StrEnum):
"""Special preferences for the recipe."""
HIGH_PROTEIN = 'High Protein'
LOW_CARB = 'Low Carb'
SPICY = 'Spicy'
BUDGET_FRIENDLY = 'Budget-Friendly'
ONE_POT_MEAL = 'One-Pot Meal'
VEGETARIAN = 'Vegetarian'
VEGAN = 'Vegan'
class CookingTime(StrEnum):
"""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):
"""A class representing an ingredient in a recipe."""
icon: str = Field(
default='ingredient',
description="The icon emoji (not emoji code like '\x1f35e', but the actual emoji like 🥕) of the ingredient",
)
name: str
amount: str
class Recipe(BaseModel):
"""A class representing a recipe."""
skill_level: SkillLevel = Field(
default=SkillLevel.BEGINNER,
description='The skill level required for the recipe',
)
special_preferences: list[SpecialPreferences] = Field(
default_factory=list,
description='Any special preferences for the recipe',
)
cooking_time: CookingTime = Field(
default=CookingTime.FIVE_MIN, description='The cooking time of the recipe'
)
ingredients: list[Ingredient] = Field(
default_factory=list,
description='Ingredients for the recipe',
)
instructions: list[str] = Field(
default_factory=list, description='Instructions for the recipe'
)
class RecipeSnapshot(BaseModel):
"""A class representing the state of the recipe."""
recipe: Recipe = Field(
default_factory=Recipe, description='The current state of the recipe'
)
agent = Agent('openai:gpt-4o-mini', deps_type=StateDeps[RecipeSnapshot])
@agent.tool_plain
async def display_recipe(recipe: Recipe) -> StateSnapshotEvent:
"""Display the recipe to the user.
Args:
recipe: The recipe to display.
Returns:
StateSnapshotEvent containing the recipe snapshot.
"""
return StateSnapshotEvent(
type=EventType.STATE_SNAPSHOT,
snapshot={'recipe': recipe},
)
@agent.instructions
async def recipe_instructions(ctx: RunContext[StateDeps[RecipeSnapshot]]) -> str:
"""Instructions for the recipe generation agent.
Args:
ctx: The run context containing recipe state information.
Returns:
Instructions string for the recipe generation agent.
"""
return dedent(
f"""
You are a helpful assistant for creating recipes.
IMPORTANT:
- Create a complete recipe using the existing ingredients
- Append new ingredients to the existing ones
- Use the `display_recipe` tool to present the recipe to the user
- Do NOT repeat the recipe in the message, use the tool instead
- Do NOT run the `display_recipe` tool multiple times in a row
Once you have created the updated recipe and displayed it to the user,
summarise the changes in one sentence, don't describe the recipe in
detail or send it as a message to the user.
The current state of the recipe is:
{ctx.deps.state.recipe.model_dump_json(indent=2)}
""",
)
app = agent.to_ag_ui(deps=StateDeps(RecipeSnapshot()))
基于工具的生成式 UI
演示对工具输出进行自定义渲染，并需要用户确认。
如果您已经运行了示例，可以在 https://:3000/pydantic-ai/feature/tool_based_generative_ui 查看。
俳句工具
generate_haiku- 用于显示英文和日文俳句的 AG-UI 工具
俳句提示
Generate a haiku about formula 1
基于工具的生成式 UI - 代码
ag_ui/api/tool_based_generative_ui.py
"""Tool Based Generative UI feature.
No special handling is required for this feature.
"""
from __future__ import annotations
from pydantic_ai import Agent
agent = Agent('openai:gpt-4o-mini')
app = agent.to_ag_ui()