-
Notifications
You must be signed in to change notification settings - Fork 15
Update framework documentation #184
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
d8aa6c5
9174db3
bcd5cf1
edaaf07
332b8a7
95b95f6
4ffc155
33b7c51
d5ce4e0
95e96cf
b388e9f
948bcc5
92e4dc1
402ecb0
cdc0083
1ee594b
aee6d90
c63cb8f
48c25e2
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,42 +5,18 @@ type: guide | |
| summary: Parse UI messages, apply approvals, stream SSE responses, set headers, and rebuild UI history. | ||
| --- | ||
|
|
||
| The AI SDK UI adapter converts between AI SDK UI messages and the Python | ||
| runtime message/event types. | ||
| This adapter provides integration with the UI part of the sister AI SDK | ||
| in TypeScript. | ||
|
|
||
| ## Parse UI messages | ||
| AI SDK UI provides `useChat`, which handles streaming complexity on the | ||
| client side, such as collecting message chunks and optimistically updating | ||
| the UI when the user submits a new message. | ||
|
|
||
| Accept `UIMessage` values from an AI SDK UI client, then convert them to | ||
| runtime messages: | ||
| The AI SDK UI's data model is significantly different from the AI SDK for | ||
| Python's, which is why you need to convert events and messages back and forth | ||
| using the adapter. | ||
|
|
||
| ```python | ||
| class ChatRequest(pydantic.BaseModel): | ||
| messages: list[ai.agents.ui.ai_sdk.UIMessage] | ||
|
|
||
|
|
||
| @app.post("/chat") | ||
| async def chat(request: ChatRequest): | ||
| messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages) | ||
| ``` | ||
|
|
||
| `to_messages` also extracts approval responses from tool parts. | ||
|
|
||
| ## Apply approval responses | ||
|
|
||
| Register extracted approvals before resuming the agent: | ||
|
|
||
| ```python | ||
| messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages) | ||
| ai.agents.ui.ai_sdk.apply_approvals(approvals) | ||
|
|
||
| async with chat_agent.run(model, messages) as stream: | ||
| ... | ||
| ``` | ||
|
|
||
| The hook registry stores each approval until the matching tool-approval hook | ||
| runs. | ||
|
|
||
| ## Stream SSE responses | ||
| ## Outbound streaming and message history | ||
|
|
||
| Use `to_sse` to convert agent events to AI SDK UI stream chunks: | ||
|
|
||
|
|
@@ -61,18 +37,6 @@ async def chat(request: ChatRequest) -> fastapi.responses.StreamingResponse: | |
| ) | ||
| ``` | ||
|
|
||
| ## Set response headers | ||
|
|
||
| Return the adapter headers on every streamed response: | ||
|
|
||
| ```python | ||
| headers = ai.agents.ui.ai_sdk.UI_MESSAGE_STREAM_HEADERS | ||
| ``` | ||
|
|
||
| These headers identify the response as an AI SDK UI message stream. | ||
|
|
||
| ## Rebuild UI history | ||
|
|
||
| Convert stored runtime messages back to AI SDK UI messages for history | ||
| endpoints: | ||
|
|
||
|
|
@@ -86,8 +50,6 @@ async def get_chat(session_id: str): | |
| The adapter groups assistant, tool, and internal hook messages into one | ||
| assistant UI message. | ||
|
|
||
| ## Support streaming tool output | ||
|
|
||
| Generator tools and subagents emit `PartialToolCallResult` events. The UI | ||
| adapter folds those partial values into the corresponding tool output: | ||
|
|
||
|
|
@@ -101,4 +63,37 @@ async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool: | |
| ``` | ||
|
|
||
| For subagents, `ai.SubAgentTool` streams nested events and stores the nested | ||
| assistant message as the tool output. | ||
| transcript as a `MessageBundle`. The model sees the nested agent's final | ||
| assistant text. | ||
|
|
||
| ## Inbound message updates | ||
|
|
||
| Accept `UIMessage` values from an AI SDK UI client, then convert them to | ||
| runtime messages: | ||
|
|
||
| ```python | ||
| class ChatRequest(pydantic.BaseModel): | ||
| messages: list[ai.agents.ui.ai_sdk.UIMessage] | ||
|
|
||
|
|
||
| @app.post("/chat") | ||
| async def chat(request: ChatRequest): | ||
| messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we want to mention the new |
||
| ``` | ||
|
|
||
| `to_messages` also extracts approval responses from tool parts. | ||
|
|
||
| ## Tool approvals | ||
|
|
||
| Register extracted approvals before resuming the agent: | ||
|
|
||
| ```python | ||
| messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages) | ||
| ai.agents.ui.ai_sdk.apply_approvals(approvals) | ||
|
|
||
| async with chat_agent.run(model, messages) as stream: | ||
| ... | ||
| ``` | ||
|
|
||
| The hook registry stores each approval until the matching tool-approval hook | ||
| runs. | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,104 +5,51 @@ type: guide | |
| summary: Override the agent loop to customize scheduling, routing, history updates, logging, and tool execution. | ||
| --- | ||
|
|
||
| Custom loops use the same primitives as the default loop. Override `Agent.loop` | ||
| when you need to change scheduling, logging, routing, or persistence. | ||
| Override `loop` when you need custom tool dispatch, logging, durability, or | ||
| branching. While the SDK defines the loop as a Python async generator, it still | ||
| uses a few framework components. Reuse those to retain framework-controlled | ||
| behavior. | ||
|
|
||
| ## When to customize the loop | ||
| ## Explore the standard loop shape | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I wonder if we should mention somewhere explicitly that |
||
|
|
||
| Override `loop` when you need custom scheduling, logging, durability, or | ||
| branching. Keep the same pattern when you still want SDK-managed history and | ||
| tool resolution. | ||
|
|
||
| ## Use the standard loop shape | ||
|
|
||
| The default loop keeps running while the last message needs more work. On each | ||
| The default loop keeps running while there is more work to do. On each | ||
| turn, it streams the model response and schedules tool calls: | ||
|
|
||
| ```python | ||
| class CustomAgent(ai.Agent): | ||
| async def loop(self, context: ai.Context): | ||
| # ai.Context keeps track of message history, run inputs, and other per-run data | ||
| async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent]: | ||
| while context.keep_running(): | ||
| # call ai.stream with whatever model, tools, and messages the user has | ||
| # passed to agent.run | ||
| async with ( | ||
| ai.stream(context=context) as stream, | ||
| # ai.ToolRunner is responsible for concurrent tool scheduling and | ||
| # streaming, as well as graceful handling of hook interruptions | ||
| ai.ToolRunner() as tool_runner, | ||
|
|
||
| ): | ||
| # ai.util.merge interleaves two streams together, so the agent | ||
| # can fire off tools as they arrive in the stream rather than | ||
| # waiting for the model to stop streaming; and also start streaming | ||
| # the streaming tools. | ||
| async for event in ai.util.merge(stream, tool_runner.events()): | ||
| yield event | ||
|
|
||
| if isinstance(event, ai.events.ToolEnd): | ||
| # context.resolve looks up the Python function for the | ||
| # tool name in the model's tool call | ||
| tool_call = context.resolve(event.tool_call) | ||
| # schedule the tool for concurrent execution | ||
| # using the tool runner | ||
| tool_runner.schedule(tool_call) | ||
|
|
||
| # add new messages to message history stored in the context. | ||
| # this works with internal replay machinery to enable serverless | ||
| # execution. | ||
| context.add(stream.message) | ||
| context.add(tool_runner.get_tool_message()) | ||
| ``` | ||
|
|
||
| ## Resolve and schedule tool calls | ||
|
|
||
| `ToolEnd` means the model finished emitting a tool call. Resolve it through the | ||
| context, then schedule it with `ToolRunner`: | ||
|
|
||
| ```python | ||
| if isinstance(event, ai.events.ToolEnd): | ||
| tool_call = context.resolve(event.tool_call) | ||
| tool_runner.schedule(tool_call) | ||
| ``` | ||
|
|
||
| `context.resolve` validates that the agent has an executable tool registered | ||
| for the model's `tool_name`. | ||
|
|
||
| ## Add messages to history | ||
|
|
||
| After a stream turn finishes, add the assistant message and any tool-result | ||
| message: | ||
|
|
||
| ```python | ||
| context.add(stream.message) | ||
| context.add(tool_runner.get_tool_message()) | ||
| ``` | ||
|
|
||
| `context.add` skips replayed assistant messages, so resume flows can call it | ||
| without duplicating history. | ||
|
|
||
| ## Run tools sequentially | ||
|
|
||
| `ToolRunner.schedule` runs active tools concurrently. To run tools one at a | ||
| time, collect the calls during the model stream, then execute them in order: | ||
|
|
||
| ```python | ||
| pending: list[ai.ToolCall] = [] | ||
|
|
||
| async for event in stream: | ||
| yield event | ||
| if isinstance(event, ai.events.ToolEnd): | ||
| pending.append(context.resolve(event.tool_call)) | ||
|
|
||
| for tool_call in pending: | ||
| result = await tool_call() | ||
| yield result | ||
| tool_runner.add_result(result) | ||
| ``` | ||
|
|
||
| This keeps the same tool-message aggregation path while preserving execution | ||
| order. | ||
|
|
||
| ## Add logging or routing | ||
|
|
||
| Custom loops can inspect events before yielding or scheduling: | ||
|
|
||
| ```python | ||
| if isinstance(event, ai.events.ToolEnd): | ||
| call = event.tool_call | ||
| print(f"tool: {call.tool_name}({call.tool_args})") | ||
| tool_runner.schedule(context.resolve(call)) | ||
| ``` | ||
|
|
||
| You can also route specific tools through custom wrappers: | ||
|
|
||
| ```python | ||
| if tool_call.name == "contact_mothership": | ||
| tool_runner.schedule(GatedToolCall(tool_call)) | ||
| else: | ||
| tool_runner.schedule(tool_call) | ||
| ``` | ||
| You can modify most of the loop and extend `CustomAgent` to fit your application's needs, | ||
| and the SDK will keep working as long as the loop is still an async generator of events. | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Probably we want to put some links in here