Logo

菜单

发布于 2026-01-05 5 阅读
0

构建股票投资组合人工智能代理（全栈式，Pydantic AI + AG-UI）

构建股票投资组合人工智能代理（全栈式，Pydantic AI + AG-UI）

太长不看

本指南将介绍如何使用 Pydantic AI 和AG-UI 协议构建全栈 AI 代理。此外，我们还将介绍如何将 AG-UI + Pydantic AI代理与 CopilotKit 集成，以便与代理进行聊天并在前端实时显示其回复。

在正式开始之前，我们先来看看我们将要涵盖的内容：

  • AG-UI协议是什么？

  • 将 Pydantic AI 代理与 AG-UI 协议集成

  • 使用 CopilotKit 将前端集成到 AG-UI + Pydantic AI 代理中

以下是我们将要构建的内容预览：

AG-UI协议是什么？

让我们来详细了解一下这个叫做代理用户交互协议（简称 AG-UI）的酷炫工具。它由 CopilotKit 开发，是一款基于事件的超轻量级开源工具。简单来说，它可以帮助你在应用程序的前端（例如用户界面）和 AI 代理（可以理解为能够执行任务的智能机器人）之间创建流畅的实时对话。

AG-UI 可以轻松处理事件驱动型聊天、跟踪状态（例如当前正在发生的事情）、使用工具，甚至可以流式传输 AI 的回复，使其像实时对话一样一点一点地显示出来。

为了在前端和AI代理之间传递信息，AG-UI使用了不同类型的事件。以下是简要概述：

  • 生命周期事件：这些事件就像信号，表示代理的任务何时开始或结束。例如，有RUN_STARTED（嘿，我们开始了！）和RUN_FINISHED（完成了！）。

  • 短信事件：这些事件以流式方式将 AI 的回复发送到前端。您会看到类似TEXT_MESSAGE_START（开始新消息）、TEXT_MESSAGE_CONTENT（这里是一些要添加的文本）和TEXT_MESSAGE_END（消息完成）之类的内容。

  • 状态管理事件：这些事件使前端和 AI 之间的所有内容保持同步，确保每个人都能及时了解情况。例如，STATE_SNAPSHOT（当前状态的完整视图）和STATE_DELTA（自上次以来的变化）。

  • 工具调用事件：当代理需要使用工具（例如获取数据或运行函数）时，就会触发这些事件。这些事件包括TOOL_CALL_START（启动工具）、TOOL_CALL_ARGS（传递所需的详细信息）和TOOL_CALL_END（工具已完成，以下是结果）。

如果您想深入了解 AG-UI 的工作原理和设置，请查看此处的文档：AG-UI 文档。继续编码——您一定可以的！

图片来自 Notion

现在我们已经了解了 AG-UI 协议是什么，让我们看看如何将其与 Pydantic AI 代理框架集成。

我们开始吧！

先决条件

要完全理解本教程，您需要对 React 或 Next.js 有一定的了解。

我们还将利用以下资源：

  • Python 是一种流行的编程语言，可用于使用 LangGraph 构建 AI 代理；请确保您的计算机上已安装 Python。

  • Pydantic AI—— 一个Python代理框架，旨在让使用生成式AI构建生产级应用程序变得更容易。

  • OpenAI API 密钥 - 一个 API 密钥，使您能够使用 GPT 模型执行各种任务；对于本教程，请确保您有权访问 GPT-4 模型。

  • CopilotKit  - 一个开源的辅助驾驶框架，用于构建自定义 AI 聊天机器人、应用内 AI 代理和文本区域。

将 Pydantic AI 代理与 AG-UI 协议集成

首先，克隆Open AG UI Demo 存储库，该存储库包含一个基于 Python 的后端（代理）和一个 Next.js 前端（前端）。

接下来，导航到后端目录：

cd agent
Enter fullscreen mode Exit fullscreen mode

然后使用 Poetry 安装依赖项：

poetry install
Enter fullscreen mode Exit fullscreen mode

之后，创建一个 .env 包含OpenAI API密钥的文件：

OPENAI_API_KEY=<<your-OpenAI-key-here>>
Enter fullscreen mode Exit fullscreen mode

然后使用以下命令运行代理：

poetry run python main.py
Enter fullscreen mode Exit fullscreen mode

现在让我们看看如何将 AG-UI 协议与 Pydantic AI 代理框架集成。

步骤 1：定义 Pydantic AI 代理状态

使用 Pydantic AI BaseModel，定义一个AgentState类来保存智能体所需的所有数据。在我们的示例中，我们跟踪诸如 `a` available_cash、investment_portfolio`b` 和tool_logs`c` 之类的信息，如文件所示agent/stock.py。

from pydantic import BaseModel, Field

class AgentState(BaseModel):
    """
    Main application state that holds all the data for the stock portfolio agent.
    This state is managed throughout the agent's lifecycle and updated by various tools.
    """
    tools: list = []                                    # List of available tools for the agent
    be_stock_data: Any = None                          # Backend stock price data (dictionary format)
    be_arguments: dict = {}                            # Arguments passed to backend operations
    available_cash: float = 0.0                       # Amount of cash available for investment
    investment_summary: dict = {}                      # Summary of investment performance and holdings
    investment_portfolio: list = []                    # List of stocks in the portfolio with amounts
    tool_logs: list = []                              # Log of tool executions for debugging
    render_standard_charts_and_table_args: dict = {}  # Arguments for rendering charts and tables
Enter fullscreen mode Exit fullscreen mode

步骤 2：初始化您的 Pydantic AI 代理

定义好代理状态后，初始化 Pydantic AI 代理，并将其传递AgentState给其共享状态容器，如文件中所示agent/stock.py。

# Pydantic AI imports for agent creation and context management
from pydantic_ai import Agent, RunContext
from pydantic_ai.ag_ui import StateDeps

agent = Agent(
    "openai:gpt-4o-mini",  # Specify the AI model to use
    deps_type=StateDeps[AgentState],  # Specify the state dependency type
)
Enter fullscreen mode Exit fullscreen mode

步骤 3：定义您的 Pydantic AI 代理工具

首先，定义核心代理工具，例如股票数据工具，如agent/stock.py文件中所示。

@agent.tool
async def gather_stock_data(
    ctx: RunContext[StateDeps[AgentState]],
    stock_tickers_list: list[str],
    investment_date: str,
    interval_of_investment: str,
    amount_of_dollars_to_be_invested: list[float],
    operation: Literal["add", "replace", "delete"],
    to_be_replaced : list[str]
) -> list[StateSnapshotEvent, StateDeltaEvent]:
    """
    Gathers historical stock data and manages the investment portfolio based on the specified operation.
    This is the primary tool for setting up and modifying stock portfolios.

    This tool is used for chat purposes. If the user query is not related to the stock portfolio, 
    you should use this tool to answer the question. The answers should be generic and should be relevant to the user's query.

    Args:
        ctx: The current run context containing the agent state
        stock_tickers_list: List of stock ticker symbols (e.g., ['AAPL', 'GOOGL', 'MSFT'])
        investment_date: Starting date for the investment in YYYY-MM-DD format
        interval_of_investment: Frequency of investment (currently supports "single_shot")
        amount_of_dollars_to_be_invested: List of dollar amounts to invest in each corresponding stock
        operation: Type of portfolio operation - "add" (append to existing), "replace" (override existing), or "delete" (remove specified)
        to_be_replaced: List of tickers to be replaced (used with "replace" operation)

    Returns:
        list: Contains StateSnapshotEvent and StateDeltaEvent for updating the UI state
    """

    // ...
Enter fullscreen mode Exit fullscreen mode

步骤 4：配置 AG-UI 状态管理事件

要配置 AG-UI 状态管理事件，首先要定义一个 JSON Patch 对象，用于对应用程序状态进行增量更新，如agent/stock.py文件中所示。

class JSONPatchOp(BaseModel):
    """
    A class representing a JSON Patch operation (RFC 6902).
    Used for making incremental updates to the application state.
    """
    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)",
    )
Enter fullscreen mode Exit fullscreen mode

然后配置 AG-UI 状态管理事件，以便使用 UI 更改更新前端，如agent/stock.py文件中所示。

@agent.tool
async def gather_stock_data(
    ctx: RunContext[StateDeps[AgentState]],
    stock_tickers_list: list[str],
    investment_date: str,
    interval_of_investment: str,
    amount_of_dollars_to_be_invested: list[float],
    operation: Literal["add", "replace", "delete"],
    to_be_replaced : list[str]
) -> list[StateSnapshotEvent, StateDeltaEvent]:
    """
    Gathers historical stock data and manages the investment portfolio based on the specified operation.
    This is the primary tool for setting up and modifying stock portfolios.

    This tool is used for chat purposes. If the user query is not related to the stock portfolio, 
    you should use this tool to answer the question. The answers should be generic and should be relevant to the user's query.

    Args:
        ctx: The current run context containing the agent state
        stock_tickers_list: List of stock ticker symbols (e.g., ['AAPL', 'GOOGL', 'MSFT'])
        investment_date: Starting date for the investment in YYYY-MM-DD format
        interval_of_investment: Frequency of investment (currently supports "single_shot")
        amount_of_dollars_to_be_invested: List of dollar amounts to invest in each corresponding stock
        operation: Type of portfolio operation - "add" (append to existing), "replace" (override existing), or "delete" (remove specified)
        to_be_replaced: List of tickers to be replaced (used with "replace" operation)

    Returns:
        list: Contains StateSnapshotEvent and StateDeltaEvent for updating the UI state
    """
    // ...

    # Initialize list to track state changes for the UI
    changes = []

    # Add initial tool log entry to show the tool has started
    tool_log_start_id = str(uuid.uuid4())
    changes.append(
        JSONPatchOp(
            op="add",
            path="/tool_logs/-",
            value={
                "message": "Starting stock data gathering...",
                "status": "in_progress",
                "id": tool_log_start_id,
            },
        )
    )

    // ...

    # STEP 2: Update the investment portfolio in the application state
    changes.append(
        JSONPatchOp(
            op="replace",
            path="/investment_portfolio",
            value=[
                {
                    "ticker": ticker,
                    "amount": amount_of_dollars_to_be_invested[index],
                }
                for index, ticker in enumerate(stock_tickers_list)
            ],
        )
    )

    // ...

    # STEP 5: Store stock data in application state
    # Convert the closing prices to dictionary format for easier handling
    changes.append(
        JSONPatchOp(
            op="replace",
            path="/be_stock_data",
            value=data["Close"].to_dict(),  # Extract closing prices and convert to dict
        )
    )
    ctx.deps.state.be_stock_data = data["Close"].to_dict()  # Update local state

    # STEP 6: Store the arguments used for this data gathering operation
    changes.append(
        JSONPatchOp(
            op="replace",
            path="/be_arguments",
            value={
                "ticker_symbols": stock_tickers_list,
                "investment_date": investment_date,
                "amount_of_dollars_to_be_invested": amount_of_dollars_to_be_invested,
                "interval_of_investment": interval_of_investment,
            },
        )
    )

    # Generate a unique ID for completion tool logging
    tool_log_id = str(uuid.uuid4())
    # Add completion tool log entry for data gathering
    changes.append(
        JSONPatchOp(
            op="add",
            path="/tool_logs/-",
            value={
                "message": "Stock data gathering completed successfully",
                "status": "completed",
                "id": tool_log_id,
            },
        )
    )

    // ...

    # STEP 7: Return state events for UI updates
    return [
        # Complete state snapshot for full UI refresh
        StateSnapshotEvent(
            type=EventType.STATE_SNAPSHOT,
            snapshot=(ctx.deps.state).model_dump(),
        ),
        # Incremental changes for efficient UI updates
        StateDeltaEvent(type=EventType.STATE_DELTA, delta=changes),
    ]
Enter fullscreen mode Exit fullscreen mode

步骤 5：配置人机交互（HITL）功能

要配置人机交互功能，请定义一个@agent.instructions异步函数，指示 Pydantic AI 代理按名称调用前端操作工具，以向用户征求反馈，如agent/stock.py文件中所示。

@agent.instructions
async def instructions(ctx: RunContext[StateDeps[AgentState]]) -> str:
    """
    Dynamic instructions for the agent that can access the current state context.
    These instructions guide the agent's behavior and tool usage patterns.

    Args:
        ctx: The current run context containing the agent state

    Returns:
        str: Formatted instructions for the agent
    """
    return dedent(f"""You are a stock portfolio analysis agent. 
                  Use the tools provided effectively to answer the user query.
                  When a user asks something related to the stock investment, make 
                  sure to call the frontend tool render_standard_charts_and_table 
                  with the tool argument render_standard_charts_and_table_args as 
                  the tool argument to the frontend after running the generate_insights tool""")
Enter fullscreen mode Exit fullscreen mode

步骤 6：设置 FastAPI 服务器并挂载代理

最后，按照文件中所示，设置 FastAPI 服务器并配置/pydantic-agent使用 pydantic AI 代理和 AG UI 集成处理请求的端点agent/stock.py。

# Import the AG UI request handler
from pydantic_ai.ag_ui import handle_ag_ui_request

# Load environment variables from .env file (must be called before using env vars)
load_dotenv()

# Create FastAPI application instance
app = FastAPI()

@app.post('/pydantic-agent')
async def run_agent(request: Request) -> Response:
    """
    Handle POST requests to the /pydantic-agent endpoint.

    This endpoint processes requests using the Pydantic AI agent with AG UI integration.
    It creates a new agent state for each request and delegates processing to the
    AG UI request handler.

    Args:
        request: The incoming HTTP request containing the user's input

    Returns:
        Response: The agent's response, typically streamed back to the client
    """
    return await handle_ag_ui_request(agent = agent, deps = StateDeps(AgentState()), request=request)

def main():
    port = int(os.getenv("PORT", "8000"))  # Get port from environment or use default
    uvicorn.run(
        "main:app",  # Module and app instance to run
        host="0.0.0.0",  # Listen on all network interfaces
        port=port,
        reload=True,  # Enable auto-reload for development
    )

if __name__ == "__main__":
    # Entry point: run the server when this script is executed directly
    main()
Enter fullscreen mode Exit fullscreen mode

恭喜！您已将 Pydantic AI 代理工作流程与 AG-UI 协议集成。现在让我们看看如何为 AG-UI + Pydantic AI 代理工作流程添加前端。

使用 CopilotKit 将前端集成到 AG-UI + Pydantic AI 代理工作流程中

在本节中，您将学习如何使用 CopilotKit 在 AG-UI + Pydantic AI 代理工作流程和前端之间建立连接。

我们开始吧。

首先，导航到前端目录：

cd frontend
Enter fullscreen mode Exit fullscreen mode

接下来，创建一个 .env 包含OpenAI API密钥的文件：

OPENAI_API_KEY=<<your-OpenAI-key-here>>
Enter fullscreen mode Exit fullscreen mode

然后安装依赖项：

pnpm install
Enter fullscreen mode Exit fullscreen mode

之后，启动开发服务器：

pnpm run dev
Enter fullscreen mode Exit fullscreen mode

访问 http://localhost:3000，您应该可以看到 AG-UI + Pydantic 代理前端已启动并运行。

图片来自 Notion

现在让我们看看如何使用 CopilotKit 为 AG-UI + Pydantic AI 代理构建前端 UI。

步骤 1：创建 HttpAgent 实例

在创建 HttpAgent 实例之前，我们先来了解一下 HttpAgent 是什么。

HttpAgent 是 AG-UI 库中的一个客户端，它可以将您的前端应用程序与任何 AG-UI 兼容的 AI 代理服务器连接起来。

要创建 HttpAgent 实例，请在 API 路由中定义它，如src/app/api/copilotkit/route.ts文件中所示。

// CopilotKit runtime imports for building AI-powered applications
import {
  CopilotRuntime, // Core runtime that manages AI agents and conversations
  copilotRuntimeNextJSAppRouterEndpoint, // Next.js App Router integration helper
  OpenAIAdapter, // Adapter for OpenAI API communication
} from "@copilotkit/runtime";

// Next.js server-side request handling
import { NextRequest } from "next/server";

// AG UI client for communicating with external agent services
import { HttpAgent } from "@ag-ui/client";

// Step 1: Configure the external Pydantic AI agent connection
// This HttpAgent connects to our Python FastAPI backend that handles stock analysis
const pydanticAgent = new HttpAgent({
  // Use environment variable for agent URL, fallback to localhost development server
  url:
    process.env.NEXT_PUBLIC_PYDANTIC_URL ||
    "http://0.0.0.0:8000/pydantic-agent",
});

// Step 2: Initialize OpenAI adapter for AI model communication
// This adapter handles the integration between CopilotKit and OpenAI's language models
const serviceAdapter = new OpenAIAdapter();

// Step 3: Create the main CopilotRuntime instance
// This runtime orchestrates the entire AI conversation system and manages connected agents
const runtime = new CopilotRuntime({
  agents: {
    // @ts-ignore - TypeScript ignore for agent type compatibility
    // Register our Pydantic agent under the name 'pydanticAgent'
    // This makes the stock analysis agent available to the frontend
    pydanticAgent: pydanticAgent,
  },
});

// Alternative: Simple runtime without agents (commented out)
// const runtime = new CopilotRuntime()
// Step 4: Define the POST endpoint handler for CopilotKit API requests
// This endpoint processes all AI conversation requests from the frontend
export const POST = async (req: NextRequest) => {
  // Step 4a: Initialize the CopilotKit endpoint handler with our configuration
  const { handleRequest } = copilotRuntimeNextJSAppRouterEndpoint({
    runtime, // The runtime instance with our registered agents
    serviceAdapter, // OpenAI adapter for language model communication
    endpoint: "/api/copilotkit", // The API route path for this endpoint
  });

  // Step 4b: Process the incoming request through the CopilotKit system
  // This handles:
  // - Authentication and validation
  // - Message routing to appropriate agents
  // - Streaming response coordination
  // - Error handling and recovery
  return handleRequest(req);
};
Enter fullscreen mode Exit fullscreen mode

步骤 2：设置 CopilotKit 提供程序

要设置 CopilotKit 提供程序， [<CopilotKit>](https://docs.copilotkit.ai/reference/components/CopilotKit) 组件必须包装应用程序中与 Copilot 相关的部分。

对于大多数使用场景，最好将 CopilotKit 提供程序包装在整个应用程序周围，例如，在您的 中 layout.tsx，如下面的文件中所示 src/app/layout.tsx 。

// Next.js imports for metadata and font handling
import type { Metadata } from "next";
import { Geist, Geist_Mono } from "next/font/google";
// Global styles for the application
import "./globals.css";
// CopilotKit UI styles for AI components
import "@copilotkit/react-ui/styles.css";
// CopilotKit core component for AI functionality
import { CopilotKit } from "@copilotkit/react-core";

// Configure Geist Sans font with CSS variables for consistent typography
const geistSans = Geist({
  variable: "--font-geist-sans",
  subsets: ["latin"],
});

// Configure Geist Mono font for code and monospace text
const geistMono = Geist_Mono({
  variable: "--font-geist-mono",
  subsets: ["latin"],
});

// Metadata configuration for SEO and page information
export const metadata: Metadata = {
  title: "AI Stock Portfolio",
  description: "AI Stock Portfolio",
};

// Root layout component that wraps all pages in the application
export default function RootLayout({
  children,
}: Readonly<{
  children: React.ReactNode;
}>) {
  return (
    <html lang="en">
      <body
        className={`${geistSans.variable} ${geistMono.variable} antialiased`}>
        {/* CopilotKit wrapper that enables AI functionality throughout the app */}
        {/* runtimeUrl points to the API endpoint for AI backend communication */}
        {/* agent specifies which AI agent to use (stockAgent for stock analysis) */}
        <CopilotKit runtimeUrl="/api/copilotkit" agent="pydanticAgent">
          {children}
        </CopilotKit>
      </body>
    </html>
  );
}
Enter fullscreen mode Exit fullscreen mode

步骤 3：设置 Copilot 聊天组件

CopilotKit 附带几个内置聊天组件，包括CopilotPopup、  CopilotSidebar和CopilotChat。

要设置 Copilot 聊天组件，请按照src/app/components/prompt-panel.tsx文件中所示进行定义。

// Client-side component directive for Next.js
"use client";

import type React from "react";
// CopilotKit chat component for AI interactions
import { CopilotChat } from "@copilotkit/react-ui";

// Props interface for the PromptPanel component
interface PromptPanelProps {
  // Amount of available cash for investment, displayed in the panel
  availableCash: number;
}

// Main component for the AI chat interface panel
export function PromptPanel({ availableCash }: PromptPanelProps) {
  // Utility function to format numbers as USD currency
  // Removes decimal places for cleaner display of large amounts
  const formatCurrency = (amount: number) => {
    return new Intl.NumberFormat("en-US", {
      style: "currency",
      currency: "USD",
      minimumFractionDigits: 0,
      maximumFractionDigits: 0,
    }).format(amount);
  };

  return (
    // Main container with full height and white background
    <div className="h-full flex flex-col bg-white">
      {/* Header section with title, description, and cash display */}
      <div className="p-4 border-b border-[#D8D8E5] bg-[#FAFCFA]">
        {/* Title section with icon and branding */}
        <div className="flex items-center gap-2 mb-2">
          <span className="text-xl">🪁</span>
          <div>
            <h1 className="text-lg font-semibold text-[#030507] font-['Roobert']">
              Portfolio Chat
            </h1>
            {/* Pro badge indicator */}
            <div className="inline-block px-2 py-0.5 bg-[#BEC9FF] text-[#030507] text-xs font-semibold uppercase rounded">
              PRO
            </div>
          </div>
        </div>
        {/* Description of the AI agent's capabilities */}
        <p className="text-xs text-[#575758]">
          Interact with the LangGraph-powered AI agent for portfolio
          visualization and analysis
        </p>

        {/* Available Cash Display section */}
        <div className="mt-3 p-2 bg-[#86ECE4]/10 rounded-lg">
          <div className="text-xs text-[#575758] font-medium">
            Available Cash
          </div>
          <div className="text-sm font-semibold text-[#030507] font-['Roobert']">
            {formatCurrency(availableCash)}
          </div>
        </div>
      </div>
      {/* CopilotKit chat interface with custom styling and initial message */}
      {/* Takes up the majority of the panel height for conversation */}
      <CopilotChat
        className="h-[78vh] p-2"
        labels={{
          // Initial welcome message explaining the AI agent's capabilities and limitations
          initial: `I am a Crew AI agent designed to analyze investment opportunities and track stock performance over time. How can I help you with your investment query? For example, you can ask me to analyze a stock like "Invest in Apple with 10k dollars since Jan 2023". \n\nNote: The AI agent has access to stock data from the past 4 years only.`
        }}
      />
    </div>
  );
}
Enter fullscreen mode Exit fullscreen mode

步骤 4：使用 CopilotKit 钩子将 AG-UI + Pydantic AI 代理状态与前端同步

在 CopilotKit 中，CoAgents 维护着一个共享状态，该状态能够将前端 UI 与代理的执行无缝连接起来。这个共享状态系统使您能够：

  • 显示代理的当前进度和中间结果

  • 通过用户界面交互更新代理的状态

  • 对应用程序中的状态变化做出实时反应

您可以在 CopilotKit 文档中了解更多关于 CoAgents 共享状态的信息 。

图片来自 Notion

要将 AG-UI + Pydantic AI 代理状态与前端同步，请使用 CopilotKit useCoAgent hook 将 AG-UI + Pydantic AI 代理状态与前端共享，如src/app/page.tsx文件中所示。

"use client";

import {
  useCoAgent,
} from "@copilotkit/react-core";

// ...

export interface SandBoxPortfolioState {
  performanceData: Array<{
    date: string;
    portfolio: number;
    spy: number;
  }>;
}
export interface InvestmentPortfolio {
  ticker: string;
  amount: number;
}

export default function OpenStocksCanvas() {

  // ...

  const [totalCash, setTotalCash] = useState(1000000);

  const { state, setState } = useCoAgent({
    name: "pydanticAgent",
    initialState: {
      available_cash: totalCash,
      investment_summary: {} as any,
      investment_portfolio: [] as InvestmentPortfolio[],
    },
  });

    // ...

  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
       {/* ... */}
    </div>
  );
}
Enter fullscreen mode Exit fullscreen mode

然后在聊天界面中渲染 AG-UI + Pydantic AI 代理的状态，这有助于以更贴近上下文的方式告知用户代理的状态。

要在聊天 UI 中渲染 AG-UI + Pydantic AI 代理的状态，可以使用 useCoAgentStateRender 钩子，如文件中所示src/app/page.tsx。

"use client";

import {
  useCoAgentStateRender,
} from "@copilotkit/react-core";

import { ToolLogs } from "./components/tool-logs";

// ...

export default function OpenStocksCanvas() {

  // ...

  useCoAgentStateRender({
    name: "pydanticAgent",
    render: ({ state }) => <ToolLogs logs={state.tool_logs} />,
  });

  // ...

  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
      {/* ... */}
    </div>
  );
}
Enter fullscreen mode Exit fullscreen mode

如果在聊天中执行查询，您应该会在聊天界面中看到 AG-UI + Pydantic AI 代理的状态任务执行情况，如下所示。

图片来自 Notion

步骤 5：在前端实现人机交互（HITL）

人机交互（HITL）允许智能体在执行过程中请求人类输入或批准，从而提高人工智能系统的可靠性和可信度。对于需要处理复杂决策或需要人类判断的行动的人工智能应用而言，这种模式至关重要。

您可以在CopilotKit 文档中了解更多关于“人机交互”的信息。

图片来自 Notion

要在前端实现人机交互（HITL），您需要使用 CopilotKit 的 useCopilotKitAction 钩子renderAndWaitForResponse方法，该方法允许从渲染函数异步返回值，如src/app/page.tsx文件中所示。

"use client";

import {
  useCopilotAction,
} from "@copilotkit/react-core";

// ...

export default function OpenStocksCanvas() {

  // ...

  useCopilotAction({
    name: "render_standard_charts_and_table",
    description:
      "This is an action to render a standard chart and table. The chart can be a bar chart or a line chart. The table can be a table of data.",
    renderAndWaitForResponse: ({ args, respond, status }) => {
      useEffect(() => {
        console.log(args, "argsargsargsargsargsaaa");
      }, [args]);
      return (
        <>
          {args?.investment_summary?.percent_allocation_per_stock &&
            args?.investment_summary?.percent_return_per_stock &&
            args?.investment_summary?.performanceData && (
              <>
                <div className="flex flex-col gap-4">
                  <LineChartComponent
                    data={args?.investment_summary?.performanceData}
                    size="small"
                  />
                  <BarChartComponent
                    data={Object.entries(
                      args?.investment_summary?.percent_return_per_stock
                    ).map(([ticker, return1]) => ({
                      ticker,
                      return: return1 as number,
                    }))}
                    size="small"
                  />
                  <AllocationTableComponent
                    allocations={Object.entries(
                      args?.investment_summary?.percent_allocation_per_stock
                    ).map(([ticker, allocation]) => ({
                      ticker,
                      allocation: allocation as a number,
                      currentValue:
                        args?.investment_summary.final_prices[ticker] *
                        args?.investment_summary.holdings[ticker],
                      totalReturn:
                        args?.investment_summary.percent_return_per_stock[
                          ticker
                        ],
                    }))}
                    size="small"
                  />
                </div>

                <button
                  hidden={status == "complete"}
                  className="mt-4 rounded-full px-6 py-2 bg-green-50 text-green-700 border border-green-200 shadow-sm hover:bg-green-100 transition-colors font-semibold text-sm"
                  onClick={() => {
                    debugger;
                    if (respond) {
                      setTotalCash(args?.investment_summary?.cash);
                      setCurrentState({
                        ...currentState,
                        returnsData: Object.entries(
                          args?.investment_summary?.percent_return_per_stock
                        ).map(([ticker, return1]) => ({
                          ticker,
                          return: return1 as number,
                        })),
                        allocations: Object.entries(
                          args?.investment_summary?.percent_allocation_per_stock
                        ).map(([ticker, allocation]) => ({
                          ticker,
                          allocation: allocation as a number,
                          currentValue:
                            args?.investment_summary?.final_prices[ticker] *
                            args?.investment_summary?.holdings[ticker],
                          totalReturn:
                            args?.investment_summary?.percent_return_per_stock[
                              ticker
                            ],
                        })),
                        performanceData:
                          args?.investment_summary?.performanceData,
                        bullInsights: args?.insights?.bullInsights || [],
                        bearInsights: args?.insights?.bearInsights || [],
                        currentPortfolioValue:
                          args?.investment_summary?.total_value,
                        totalReturns: (
                          Object.values(
                            args?.investment_summary?.returns
                          ) as number[]
                        ).reduce((acc, val) => acc + val, 0),
                      });
                      setInvestedAmount(
                        (
                          Object.values(
                            args?.investment_summary?.total_invested_per_stock
                          ) as number[]
                        ).reduce((acc, val) => acc + val, 0)
                      );
                      setState({
                        ...state,
                        available_cash: totalCash,
                      });
                      respond(
                        "Data rendered successfully. Provide a summary of the investments by not making any tool calls."
                      );
                    }
                  }}>
                  Accept
                </button>
                <button
                  hidden={status == "complete"}
                  className="rounded-full px-6 py-2 bg-red-50 text-red-700 border border-red-200 shadow-sm hover:bg-red-100 transition-colors font-semibold text-sm ml-2"
                  onClick={() => {
                    debugger;
                    if (respond) {
                      respond(
                        "Data rendering rejected. Just give a summary of the rejected investments by not making any tool calls."
                      );
                    }
                  }}>
                  Reject
                </button>
              </>
            )}
        </>
      );
    },
  });

  // ...

  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
      {/* ... */}
    </div>
  );
}
Enter fullscreen mode Exit fullscreen mode

当代理通过工具/操作名称触发前端操作，以在执行过程中请求人工输入或反馈时，最终用户会看到一个选择提示（显示在聊天界面中）。然后，用户可以通过按下聊天界面中的按钮进行选择，如下图所示。

图片来自 Notion

步骤 6：在前端流式传输 AG-UI + Pydantic AI 代理的响应

要将 AG-UI + Pydantic AI 代理的响应或结果流式传输到前端，请将代理的状态字段值传递给前端组件，如文件中所示src/app/page.tsx。

"use client";

import { useEffect, useState } from "react";
import { PromptPanel } from "./components/prompt-panel";
import { GenerativeCanvas } from "./components/generative-canvas";
import { ComponentTree } from "./components/component-tree";
import { CashPanel } from "./components/cash-panel";

// ...

export default function OpenStocksCanvas() {
  const [currentState, setCurrentState] = useState<PortfolioState>({
    id: "",
    trigger: "",
    performanceData: [],
    allocations: [],
    returnsData: [],
    bullInsights: [],
    bearInsights: [],
    currentPortfolioValue: 0,
    totalReturns: 0,
  });
  const [sandBoxPortfolio, setSandBoxPortfolio] = useState<
    SandBoxPortfolioState[]
  >([]);
  const [selectedStock, setSelectedStock] = useState<string | null>(null);


  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
      {/* Left Panel - Prompt Input */}
      <div className="w-85 border-r border-[#D8D8E5] bg-white flex-shrink-0">
        <PromptPanel availableCash={totalCash} />
      </div>

      {/* Center Panel - Generative Canvas */}
      <div className="flex-1 relative min-w-0">
        {/* Top Bar with Cash Info */}
        <div className="absolute top-0 left-0 right-0 bg-white border-b border-[#D8D8E5] p-4 z-10">
          <CashPanel
            totalCash={totalCash}
            investedAmount={investedAmount}
            currentPortfolioValue={
              totalCash + investedAmount + currentState.totalReturns || 0
            }
            onTotalCashChange={setTotalCash}
            onStateCashChange={setState}
          />
        </div>

        <div className="pt-20 h-full">
          <GenerativeCanvas
            setSelectedStock={setSelectedStock}
            portfolioState={currentState}
            sandBoxPortfolio={sandBoxPortfolio}
            setSandBoxPortfolio={setSandBoxPortfolio}
          />
        </div>
      </div>

      {/* Right Panel - Component Tree (Optional) */}
      {showComponentTree && (
        <div className="w-64 border-l border-[#D8D8E5] bg-white flex-shrink-0">
          <ComponentTree portfolioState={currentState} />
        </div>
      )}
    </div>
  );
}
Enter fullscreen mode Exit fullscreen mode

如果您向客服人员提出询问并批准其反馈请求，您应该会在用户界面中看到客服人员的回复或结果，如下所示。

结论

在本指南中，我们逐步介绍了如何将 Pydantic AI 代理与 AG-UI 协议集成，然后使用 CopilotKit 为代理添加前端。

虽然我们已经探索了一些功能，但我们仅仅触及了 CopilotKit 无数用例的冰山一角，从构建交互式 AI 聊天机器人到构建代理解决方案——本质上，CopilotKit 可以让您在几分钟内为您的产品添加大量有用的 AI 功能。

希望本指南能帮助您更轻松地将人工智能驱动的副驾驶功能集成到您现有的应用程序中。

在Twitter上关注 CopilotKit  并打个招呼，如果你想开发一些很酷的东西，请加入 Discord 社区。

文章来源：https://dev.to/copilotkit/build-a-fullstack-stock-portfolio-ai-agent-with-pydantic-ai-ag-ui-3e2e
我为什么从 Atom 转到 Vis...
🚀 提升你的开发技能：10+ 个...