AG-UI 不是炒作！这是 AI 开发者迟早要学的通信协议，从原理到落地一篇搞懂

import * as readline from "readline"
import { agent } from "./agent"
import { randomUUID } from "node:crypto"
// 创建命令行交互接口
const rl = readline.createInterface({
  input: process.stdin,
  output: process.stdout,
})
async function chatLoop() {
  console.log("🤖 AG-UI助手已启动！")
  console.log("输入消息后按Enter发送，按Ctrl+D退出。\n")
  return new Promise<void>((resolve) => {
    const promptUser = () => {
      rl.question("> ", async (input) => {
        // 处理空输入
        if (input.trim() === "") {
          promptUser()
          return
        }
        console.log("")
        // 处理消息时暂停输入
        rl.pause()
        // 将用户消息添加到对话中（符合AG-UI规范）
        agent.messages.push({
          id: randomUUID(), // 生成唯一消息ID
          role: "user",     // 角色为“用户”
          content: input.trim(), // 用户输入内容
        })
        try {
          // 运行智能体并绑定AG-UI事件处理器
          await agent.runAgent(
            {}, // 无需额外配置
            {
              // 文本消息开始事件：提示“助手回复中”
              onTextMessageStartEvent() {
                process.stdout.write("🤖 助手： ")
              },
              // 文本消息内容事件：流式输出回复内容
              onTextMessageContentEvent({ event }) {
                process.stdout.write(event.delta)
              },
              // 文本消息结束事件：回复完成后换行
              onTextMessageEndEvent() {
                console.log("\n")
              },
            }
          )
        } catch (error) {
          console.error("❌ 错误：", error)
        }
        // 恢复输入，等待下一次用户提问
        rl.resume()
        promptUser()
      })
    }
    // 处理Ctrl+D退出事件
    rl.on("close", () => {
      console.log("\n👋 感谢使用AG-UI助手！")
      resolve()
    })
    // 启动用户提问循环
    promptUser()
  })
}
async function main() {
  await chatLoop()
}
main().catch(console.error)

import os
import uuid
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from ag_ui.core import (
    RunAgentInput,
    EventType,
    RunStartedEvent,
    RunFinishedEvent,
    RunErrorEvent,
)
from ag_ui.encoder import EventEncoder
from openai import OpenAI
app = FastAPI(title="AG-UI OpenAI Server")
# Initialize OpenAI client - uses OPENAI_API_KEY from environment
client = OpenAI()
@app.post("/")
async def agentic_chat_endpoint(input_data: RunAgentInput, request: Request):
    """OpenAI agentic chat endpoint"""
    accept_header = request.headers.get("accept")
    encoder = EventEncoder(accept=accept_header)
    async def event_generator():
        try:
            yield encoder.encode(
                RunStartedEvent(
                    type=EventType.RUN_STARTED,
                    thread_id=input_data.thread_id,
                    run_id=input_data.run_id
                )
            )
            # Call OpenAI's API with streaming enabled
            stream = client.chat.completions.create(
                model="gpt-4o",
                stream=True,
                # Convert AG-UI tools format to OpenAI's expected format
                tools=[
                    {
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "description": tool.description,
                            "parameters": tool.parameters,
                        }
                    }
                    for tool in input_data.tools
                ] if input_data.tools else None,
                # Transform AG-UI messages to OpenAI's message format
                messages=[
                    {
                        "role": message.role,
                        "content": message.content or "",
                        # Include tool calls if this is an assistant message with tools
                        **({"tool_calls": message.tool_calls} if message.role == "assistant" and hasattr(message, 'tool_calls') and message.tool_calls else {}),
                        # Include tool call ID if this is a tool result message
                        **({"tool_call_id": message.tool_call_id} if message.role == "tool" and hasattr(message, 'tool_call_id') else {}),
                    }
                    for message in input_data.messages
                ],
            )
            message_id = str(uuid.uuid4())
            # Stream each chunk from OpenAI's response
            for chunk in stream:
                # Handle text content chunks
                if chunk.choices[0].delta.content:
                    yield encoder.encode({
                        "type": EventType.TEXT_MESSAGE_CHUNK,
                        "message_id": message_id,
                        "delta": chunk.choices[0].delta.content,
                    })
                # Handle tool call chunks
                elif chunk.choices[0].delta.tool_calls:
                    tool_call = chunk.choices[0].delta.tool_calls[0]
                    yield encoder.encode({
                        "type": EventType.TOOL_CALL_CHUNK,
                        "tool_call_id": tool_call.id,
                        "tool_call_name": tool_call.function.name if tool_call.function else None,
                        "parent_message_id": message_id,
                        "delta": tool_call.function.arguments if tool_call.function else None,
                    })
            yield encoder.encode(
                RunFinishedEvent(
                    type=EventType.RUN_FINISHED,
                    thread_id=input_data.thread_id,
                    run_id=input_data.run_id
                )
            )
        except Exception as error:
            yield encoder.encode(
                RunErrorEvent(
                    type=EventType.RUN_ERROR,
                    message=str(error)
                )
            )
    return StreamingResponse(
        event_generator(),
        media_type=encoder.get_content_type()
    )
def main():
    """Run the uvicorn server."""
    port = int(os.getenv("PORT", "8000"))
    uvicorn.run(
        "example_server:app",
        host="0.0.0.0",
        port=port,
        reload=True
    )
if __name__ == "__main__":
    main()