Pydantic AI 与 CopilotKit

通过 Pydantic AI 可以很快的编写一个 Agent 应用，也可以像用 Pydantic-AI 打造你的专属 CLI 编码助手一样快速的实现一个命令行的界面，但是当与 Web 界面结合时，我们发现还有许多工作要做。

我们需要搞好后端 Agent 与前端界面的通讯，需要将 Agent 的返回数据展示到前端页面。虽然 Streamlit 和 Gradio 这样的框架能快速的创建一些美观的界面，但是似乎不太适合用于生产环境。

所以有人发明了 CopilotKit 以及其背后的 AG-UI 协议。Agent 和工具之间使用 MCP，Agent 之间用 A2A ，而 Agent 和用户之间则使用 AG-UI：

AG-UI

通过 AG-UI，我们可以直接将 Agent 的内容与 Web UI 结合起来。

下面以 Pydantic AI 为例，一窥 AG-UI 可以做什么。

编写 Agent

参考 Hello Pydantic AI，我们编写一个 agent.py：

import os
import logging
from pydantic_ai import Agent
from google.genai.types import HarmBlockThreshold, HarmCategory
from pydantic_ai.models.google import GoogleModel, GoogleModelSettings
from pydantic_ai.providers.google import GoogleProvider
from google.genai.types import HttpOptions, HttpRetryOptions
from google import genai


settings = GoogleModelSettings(
    temperature=0.2,
    # max_tokens=1024,
    # google_thinking_config={'thinking_budget': 2048},
    google_safety_settings=[
        {
            'category': HarmCategory.HARM_CATEGORY_HARASSMENT,
            'threshold': HarmBlockThreshold.BLOCK_NONE,
        },
        {
            'category': HarmCategory.HARM_CATEGORY_HATE_SPEECH,
            'threshold': HarmBlockThreshold.BLOCK_NONE,
        },
        {
            'category': HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
            'threshold': HarmBlockThreshold.BLOCK_NONE,
        },
        {
            'category': HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
            'threshold': HarmBlockThreshold.BLOCK_NONE,
        },
        {
            'category': HarmCategory.HARM_CATEGORY_CIVIC_INTEGRITY,
            'threshold': HarmBlockThreshold.BLOCK_NONE,
        }
    ]
)

def get_model():
    proxy_url = os.environ.get("HTTPS_PROXY")
    retry_attempts = 2  
    http_options_kwargs = {}
    if proxy_url:
        logging.info(f"正在使用代理: {proxy_url}")
        proxies = {
            "http://": proxy_url,
            "https://": proxy_url,
        }
        http_options_kwargs["client_args"] = {"proxies": proxies}
        http_options_kwargs["async_client_args"] = {"proxies": proxies}
    logging.info(f"设置 API 请求重试次数为: {retry_attempts}")
    http_options_kwargs["retry_options"] = HttpRetryOptions(attempts=retry_attempts)
    http_options = HttpOptions(**http_options_kwargs)
    google_client = genai.Client(http_options=http_options)
    provider = GoogleProvider(client=google_client)
    model = GoogleModel(model_name='gemini-2.0-flash', provider=provider)
    return model


agent = Agent(
    model=get_model(),
    system_prompt='Be fun!', # Read system prompt from .env or use default
    model_settings=settings
)

app = agent.to_ag_ui()

# If you want the server to run on invocation, you can do the following:
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

这个 agent 运行需要安装依赖：

uv add 'pydantic-ai-slim[ag-ui]' uvicorn

另外，运行前需要设置 GOOGLE_API_KEY 环境变量。执行 uv run agent.py，会在本机的 8000 端口暴露 Agent 协议服务。

编写前端

CopilotKit 的多数例子是基于 Next.js 的，各位可以自学快速入门。

在一个 Next.js 项目中，安装依赖：

npm install @copilotkit/react-ui @copilotkit/react-core @copilotkit/runtime

然后设置 CopilotKit Runtime。创建 app/api/copilotkit/route.ts：

import {
  CopilotRuntime,
  ExperimentalEmptyAdapter,
  copilotRuntimeNextJSAppRouterEndpoint,
} from "@copilotkit/runtime";
import { HttpAgent } from "@ag-ui/client";
import { NextRequest } from "next/server";

// 1. You can use any service adapter here for multi-agent support. We use
//    the empty adapter since we're only using one agent.
const serviceAdapter = new ExperimentalEmptyAdapter();

// 2. Create the CopilotRuntime instance and utilize the PydanticAI AG-UI
//    integration to setup the connection.
const runtime = new CopilotRuntime({
  agents: {
    // Our AG-UI endpoint URL
    "my_agent": new HttpAgent({ url: "http://localhost:8000/" }),
  }   
});

// 3. Build a Next.js API route that handles the CopilotKit runtime requests.
export const POST = async (req: NextRequest) => {
  const { handleRequest } = copilotRuntimeNextJSAppRouterEndpoint({
    runtime, 
    serviceAdapter,
    endpoint: "/api/copilotkit",
  });

  return handleRequest(req);
};

这创建了一个 /api/copilotkit 的 POST API，这个 API 与前面创建的 Agent 后端关联。

然后我们修改 app/layout.tsx：

import type { Metadata } from "next";
import { Geist, Geist_Mono } from "next/font/google";
import "./globals.css";
import { CopilotKit } from "@copilotkit/react-core"; 

const geistSans = Geist({
  variable: "--font-geist-sans",
  subsets: ["latin"],
});

const geistMono = Geist_Mono({
  variable: "--font-geist-mono",
  subsets: ["latin"],
});

export const metadata: Metadata = {
  title: "Create Next App",
  description: "Generated by create next app",
};

export default function RootLayout({
  children,
}: Readonly<{
  children: React.ReactNode;
}>) {
  return (
    <html lang="en">
      <body
        className={`${geistSans.variable} ${geistMono.variable} antialiased`}
      >
        <CopilotKit runtimeUrl="/api/copilotkit" agent="my_agent"> 
          {children}
        </CopilotKit>
      </body>
    </html>
  );
}

然后编辑 app/page.tsx：

import Image from "next/image";
import "@copilotkit/react-ui/styles.css";
import { CopilotChat, CopilotSidebar, CopilotPopup } from "@copilotkit/react-ui";

export default function Home() {
  return (
    <div>
      <CopilotChat />
      <CopilotSidebar />
      <CopilotPopup />
    </div>
  );
}

然后执行 npm run dev 可以运行前端。

测试

浏览器访问 http://localhost:3000/：

alt text

可以看到一个聊天界面，发挥作用的正是刚才用 Pydantic AI 编写的 Agent。

总结

CopilotKit 有一些内置的组件，也可以自定义组件的样式，可以自定义 Action，可以为特定的 Agent 调用配置 Action，用不同的组件展示返回结果。这些都是自定义 UI 所常见的需求。

对于前端能力薄弱的开发团队， CopilotKit 无疑可以加快开发速度，而且这种标准化的前后台交互方式也能减少 AI 应用的维护成本。

不过，只有越来越多的产品支持 AG-UI 协议，它才能发挥更大的作用。比如说想使用 shadcn 的组件，不过没有 Google 到。

Pydantic AI 与 CopilotKit

编写 Agent

编写前端

测试

总结

On this page