使用 Agno + AG-UI 构建您自己的 AI 股票投资组合代理
在本指南中，您将学习如何将Agno 代理与 AG-UI 协议集成。
此外，我们还将介绍如何将 AG-UI 和 Agno 代理与 CopilotKit 集成，使用户能够与代理聊天并在前端流式传输其回复。
在正式开始之前，我们先来看看我们将要涵盖的内容：
-
AG-UI协议是什么？
-
将 Agno 代理与 AG-UI 协议集成
-
使用 CopilotKit 将前端集成到 AG-UI + Agno 代理中
以下是我们将要构建的内容预览：
AG-UI协议是什么？
由 CopilotKit 开发的 Agent User Interaction Protocol (AG-UI) 是一种开源、轻量级、基于事件的协议，可促进前端和 AI 代理之间丰富的实时交互。
AG-UI 协议支持事件驱动通信、状态管理、工具使用和流式 AI 代理响应。
为了在前端和您的 AI 代理之间传递信息，AG-UI 使用以下事件：
-
生命周期事件：这些事件标志着代理任务执行的开始或结束。生命周期事件包括开始
RUN_STARTED和RUN_FINISHED结束事件。 -
短信事件：这些事件处理流式代理对前端的响应。短信事件包括
TEXT_MESSAGE_START、TEXT_MESSAGE_CONTENT和TEXT_MESSAGE_END事件。 -
工具调用事件：这些事件管理代理的工具执行。工具调用事件包括
TOOL_CALL_START、TOOL_CALL_ARGS和TOOL_CALL_END事件。 -
状态管理事件：这些事件使前端和AI代理的状态保持同步。状态管理事件包括
STATE_SNAPSHOT事件STATE_DELTA。
您可以访问 AG-UI文档，了解更多关于 AG-UI协议及其架构的信息。
现在我们已经了解了 AG-UI 协议是什么，让我们看看如何将其与 Agno 代理框架集成。
我们开始吧！
先决条件
要完全理解本教程，您需要对 React 或 Next.js 有一定的了解。
我们还将利用以下资源：
-
Python 是一种流行的编程语言，可用于使用 LangGraph 构建 AI 代理；请确保您的计算机上已安装 Python。
-
Agno—— 一个用于构建具有记忆、知识和推理能力的多智能体系统的全栈框架。
-
OpenAI API 密钥 - 用于启用 GPT 模型执行各种任务的 API 密钥；对于本教程，请确保您有权访问 GPT-4 模型。
-
CopilotKit - 一个开源的辅助驾驶框架，用于构建自定义 AI 聊天机器人、应用内 AI 代理和文本区域。
将 Agno 代理与 AG-UI 协议集成
首先，克隆Open AG UI Demo 存储库，该存储库包含一个基于 Python 的后端（代理）和一个 Next.js 前端（前端）。
接下来，导航到后端目录：
cd agent
然后使用 Poetry 安装依赖项：
poetry install
之后，创建一个 .env 包含OpenAI API密钥的文件：
OPENAI_API_KEY=<<your-OpenAI-key-here>>
然后使用以下命令运行代理：
poetry run python main.py
要测试 AG-UI + Agno 集成，请在https://reqbin.com/curl上运行以下 curl 命令。
curl -X POST "http://localhost:8000/agno-agent" \
-H "Content-Type: application/json" \
-d '{
"thread_id": "test_thread_123",
"run_id": "test_run_456",
"messages": [
{
"id": "msg_1",
"role": "user",
"content": "Analyze AAPL stock with a $10000 investment from 2023-01-01"
}
],
"tools": [],
"context": [],
"forwarded_props": {},
"state": {}
}'
现在让我们看看如何将 AG-UI 协议与 Agno 代理框架集成。
步骤 1：创建 Agno 代理工作流程
在将 AG-UI 协议与 Agno 代理集成之前，请按照文件中所示创建 Agno 代理工作流程agent/stock_analysis.py。
# Import necessary libraries and modules for stock analysis workflow
from agno.agent.agent import Agent # Core agent functionality
from agno.models.openai.chat import OpenAIChat # OpenAI chat model integration
from agno.workflow.v2 import Step, Workflow, StepOutput # Workflow management components
from ag_ui.core import EventType, StateDeltaEvent # Event handling for UI updates
from ag_ui.core import AssistantMessage, ToolMessage # Message types for chat interface
import uuid # For generating unique identifiers
import asyncio # For asynchronous operations
from openai import OpenAI # OpenAI API client
from dotenv import load_dotenv # For loading environment variables
import os # Operating system interface
import json # JSON data handling
import yfinance as yf # Yahoo Finance API for stock data
from datetime import datetime # Date and time handling
import numpy as np # Numerical computing
import pandas as pd # Data manipulation and analysis
from prompts import insights_prompt, system_prompt # Custom prompt templates
# Load environment variables from .env file (contains API keys, etc.)
load_dotenv()
// ...
# WORKFLOW DEFINITION: Complete stock analysis pipeline
# This workflow orchestrates all the steps in sequence:
# 1. Chat: Parse user input and extract parameters
# 2. Simulation: Gather historical stock data
# 3. Cash_allocation: Calculate portfolio performance and allocations
# 4. Gather_insights: Generate market insights
stock_analysis_workflow = Workflow(
name="Mixed Execution Pipeline",
steps=[chat, simultion, cash_allocation, gather_insights], # Function
)
// ...
步骤 2：使用 FastAPI 创建端点
定义好 Agno 代理工作流后，创建一个 FastAPI 端点，并按照agent/main.py文件中所示导入 Agno 代理工作流。
# Import necessary libraries for FastAPI web server and async operations
from fastapi import FastAPI # Main FastAPI framework for web API
from fastapi.responses import StreamingResponse # For streaming real-time responses to the client
import uuid # For generating unique identifiers
from typing import Any # Type hints for better code documentation
import os # Operating system interface for environment variables
import uvicorn # ASGI server for running FastAPI applications
import asyncio # Asynchronous I/O operations and event loop management
# Import event system components from ag_ui.core for real-time UI updates
from ag_ui.core import (
RunAgentInput, # Input data structure for agent requests
StateSnapshotEvent, # Event for sending current state to UI
EventType, # Enumeration of all possible event types
RunStartedEvent, # Event signaling agent run has started
RunFinishedEvent, # Event signaling agent run has completed
TextMessageStartEvent, # Event for beginning text message streaming
TextMessageEndEvent, # Event for ending text message streaming
TextMessageContentEvent, # Event for streaming text content chunks
ToolCallStartEvent, # Event for beginning tool/function calls
ToolCallEndEvent, # Event for ending tool/function calls
ToolCallArgsEvent, # Event for streaming tool arguments
StateDeltaEvent, # Event for incremental state updates
)
# Import event encoder for formatting events for streaming
from ag_ui.encoder import EventEncoder # Encodes events for client consumption
from typing import List # Type hint for list types
# Import the main stock analysis workflow from our custom module
from stock_analysis import stock_analysis_workflow
# Initialize FastAPI application instance
app = FastAPI()
# MAIN API ENDPOINT: Handle stock analysis agent requests
# This endpoint receives investment queries and streams back real-time responses
@app.post("/agno-agent")
async def agno_agent(input_data: RunAgentInput):
// ...
# SERVER STARTUP FUNCTION: Initialize and run the FastAPI server
def main():
"""Run the uvicorn server."""
# Step 1: Get port from environment variable or default to 8000
port = int(os.getenv("PORT", "8000"))
# Step 2: Start uvicorn ASGI server with configuration
uvicorn.run(
"main:app", # Module:app reference
host="0.0.0.0", # Listen on all network interfaces
port=port, # Port number
reload=True, # Auto-reload on code changes (development mode)
)
# SCRIPT ENTRY POINT: Run server when script is executed directly
if __name__ == "__main__":
main() # Start the server
步骤 3：定义事件生成器
创建 FastAPI 端点后，定义一个事件生成器，生成 AG-UI 协议事件流，初始化事件编码器，并将流式响应返回给客户端或前端，如文件中所示agent/main.py。
# MAIN API ENDPOINT: Handle stock analysis agent requests
# This endpoint receives investment queries and streams back real-time responses
@app.post("/agno-agent")
async def agno_agent(input_data: RunAgentInput):
try:
# ASYNC GENERATOR: Streams events to client in real-time
# This function generates a stream of events that get sent to the frontend
async def event_generator():
# Step 1: Initialize event streaming infrastructure
encoder = EventEncoder() # Encodes events for transmission
event_queue = asyncio.Queue() # Queue for handling events from workflow
# Step 2: Define event emission callback function
# This function gets called by workflow steps to send updates to UI
def emit_event(event):
event_queue.put_nowait(event) # Add event to queue without blocking
# Step 3: Generate unique message identifier for this conversation
message_id = str(uuid.uuid4())
// ...
except Exception as e:
# Step 23: Handle any errors during execution
print(e) # Log error for debugging
# Step 24: Return streaming response to client
# FastAPI will stream the events as Server-Sent Events (SSE)
return StreamingResponse(event_generator(), media_type="text/event-stream")
步骤 4：配置 AG-UI 协议生命周期事件
定义事件生成器后，定义 AG-UI 协议生命周期事件，这些事件表示 AG-UI + Agno 代理工作流运行的生命周期，如agent/main.py文件中所示。
# MAIN API ENDPOINT: Handle stock analysis agent requests
# This endpoint receives investment queries and streams back real-time responses
@app.post("/agno-agent")
async def agno_agent(input_data: RunAgentInput):
try:
# ASYNC GENERATOR: Streams events to client in real-time
# This function generates a stream of events that get sent to the frontend
async def event_generator():
// ...
# Step 4: Send initial "run started" event to client
# Signals to the UI that the agent has begun processing
the yield encoder.encode(
RunStartedEvent(
type=EventType.RUN_STARTED,
thread_id=input_data.thread_id, # Conversation thread identifier
run_id=input_data.run_id, # Unique run identifier
)
)
// ...
# Step 22: Send final "run finished" event
# Signal to client that the entire agent run has completed
the yield encoder.encode(
RunFinishedEvent(
type=EventType.RUN_FINISHED,
thread_id=input_data.thread_id,
run_id=input_data.run_id,
)
)
except Exception as e:
# Step 23: Handle any errors during execution
print(e) # Log error for debugging
# Step 24: Return streaming response to client
# FastAPI will stream the events as Server-Sent Events (SSE)
return StreamingResponse(event_generator(), media_type="text/event-stream")
步骤 5：配置 AG-UI 协议状态管理事件
STATE_DELTA定义 AG-UI 协议生命周期事件后，使用Agno 代理工作流步骤中的事件集成 AG-UI 协议状态管理事件，如agent/stock_analysis.py文件中所示。
# WORKFLOW STEP 1: Initial chat processing and parameter extraction
# This function handles the first interaction with the user query
async def chat(step_input):
try:
// ...
# Step 3: Emit state change event to update UI
# Uses JSON patch operations to update the frontend state
step_input.additional_data["emit_event"](
StateDeltaEvent(
type=EventType.STATE_DELTA,
delta=[
{
"op": "add", # Add new log entry
"path": "/tool_logs/-", # Append to tool_logs array
"value": {
"message": "Analyzing user query",
"status": "processing",
"id": tool_log_id,
},
}
],
)
)
await asyncio.sleep(0) # Yield control to event loop
// ...
# Step 7: Update tool log status to completed
# Find the last log entry and mark it as completed
index = len(step_input.additional_data['tool_logs']) - 1
step_input.additional_data["emit_event"](
StateDeltaEvent(
type=EventType.STATE_DELTA,
delta=[
{
"op": "replace", # Update existing value
"path": f"/tool_logs/{index}/status",
"value": "completed",
}
],
)
)
await asyncio.sleep(0) # Yield control to event loop
// ...
except Exception as e:
# Step 10: Handle errors gracefully
print(e) # Log error for debugging
# Add an empty assistant message to maintain conversation flow
a_message = AssistantMessage(id=response.id, content="", role="assistant")
step_input.additional_data["messages"].append(a_message)
return "end" # Signal workflow termination
然后，在 FastAPI 端点中，使用STATE_SNAPSHOT状态管理事件初始化 Agno 代理工作流状态，如下所示。
# MAIN API ENDPOINT: Handle stock analysis agent requests
# This endpoint receives investment queries and streams back real-time responses
@app.post("/agno-agent")
async def agno_agent(input_data: RunAgentInput):
try:
# ASYNC GENERATOR: Streams events to client in real-time
# This function generates a stream of events that get sent to the frontend
async def event_generator():
# Step 1: Initialize event streaming infrastructure
encoder = EventEncoder() # Encodes events for transmission
event_queue = asyncio.Queue() # Queue for handling events from workflow
// ...
# Step 5: Send current state snapshot to client
# Provides initial state including cash, portfolio, and logs
yield encoder.encode(
StateSnapshotEvent(
type=EventType.STATE_SNAPSHOT,
snapshot={
"available_cash": input_data.state["available_cash"], # User's cash balance
"investment_summary": input_data.state["investment_summary"], # Portfolio summary
"investment_portfolio": input_data.state[
"investment_portfolio" # Current holdings
],
"tool_logs": [], # Initialize empty tool execution logs
},
)
)
// ...
except Exception as e:
# Step 23: Handle any errors during execution
print(e) # Log error for debugging
# Step 24: Return streaming response to client
# FastAPI will stream the events as Server-Sent Events (SSE)
return StreamingResponse(event_generator(), media_type="text/event-stream")
步骤 6：使用 AG-UI 协议配置 Agno 代理工作流程
初始化 Agno 代理工作流状态后，按照agent/main.py文件中的说明，将 Agno 代理工作流与 AG-UI 协议集成。
# MAIN API ENDPOINT: Handle stock analysis agent requests
# This endpoint receives investment queries and streams back real-time responses
@app.post("/agno-agent")
async def agno_agent(input_data: RunAgentInput):
try:
# ASYNC GENERATOR: Streams events to client in real-time
# This function generates a stream of events that get sent to the frontend
async def event_generator():
// ...
# Step 6: Start the stock analysis workflow as an async task
# This runs the entire analysis pipeline in the background
agent_task = asyncio.create_task(
stock_analysis_workflow.arun( # Execute workflow asynchronously
additional_data= {
"tools": input_data.tools, # Available tools/functions
"messages": input_data.messages, # Conversation history
"emit_event": emit_event, # Callback for sending UI updates
"available_cash": input_data.state["available_cash"], # Cash balance
"investment_portfolio": input_data.state["investment_portfolio"], # Holdings
"tool_logs": [], # Initialize logs array
}
)
)
# Step 7: Stream events from workflow while it's running
# This loop processes events from the workflow and streams them to the client
while True:
try:
# Step 8: Wait for events from workflow (with timeout)
event = await asyncio.wait_for(event_queue.get(), timeout=0.1)
yield encoder.encode(event) # Send event to client
except asyncio.TimeoutError:
# Step 9: Check if workflow has completed
# Check if the agent is done
if agent_task.done():
break # Exit loop when workflow finishes
# Step 10: Clear tool logs after workflow completion
# Send event to reset tool logs in UI
yield encoder.encode(
StateDeltaEvent(
type=EventType.STATE_DELTA,
delta=[{"op": "replace", "path": "/tool_logs", "value": []}],
)
)
// ...
except Exception as e:
# Step 23: Handle any errors during execution
print(e) # Log error for debugging
# Step 24: Return streaming response to client
# FastAPI will stream the events as Server-Sent Events (SSE)
return StreamingResponse(event_generator(), media_type="text/event-stream")
步骤 7：配置 AG-UI 协议工具事件以处理人机交互断点
将 Agno 代理工作流与 AG-UI 协议集成后，将带有工具调用名称的工具调用消息附加到状态，如文件中的现金分配步骤所示agent/stock_analysis.py。
# WORKFLOW STEP 3: Cash allocation and portfolio simulation
# This function calculates how investments would perform over time
async def cash_allocation(step_input):
# Step 1: Validate that we have tool calls to process
if step_input.additional_data["messages"][-1].tool_calls is None:
return
# Step 2: Initialize tool logging for allocation calculation
tool_log_id = str(uuid.uuid4())
step_input.additional_data["tool_logs"].append(
{
"id": tool_log_id,
"message": "Calculating portfolio allocation",
"status": "processing",
}
)
// ...
# Step 31: Add tool message to conversation
step_input.additional_data["messages"].append(
ToolMessage(
role="tool",
id=str(uuid.uuid4()),
content="The relevant details had been extracted", # Confirmation message
tool_call_id=step_input.additional_data["messages"][-1].tool_calls[0].id,
)
)
# Step 32: Request chart rendering through tool call
step_input.additional_data["messages"].append(
AssistantMessage(
role="assistant",
tool_calls=[
{
"id": str(uuid.uuid4()),
"type": "function",
"function": {
"name": "render_standard_charts_and_table", # Frontend rendering function
"arguments": json.dumps(
{"investment_summary": step_input.additional_data["investment_summary"]}
),
},
}
],
id=str(uuid.uuid4()),
)
)
# Step 33: Mark allocation calculation as completed
index = len(step_input.additional_data["tool_logs"]) - 1
step_input.additional_data["emit_event"](
StateDeltaEvent(
type=EventType.STATE_DELTA,
delta=[
{
"op": "replace",
"path": f"/tool_logs/{index}/status",
"value": "completed",
}
],
)
)
await asyncio.sleep(0) # Yield control to event loop
return
然后，定义 AG-UI 协议工具调用事件，代理可以使用这些事件通过调用前端操作并使用工具名称来触发前端操作，以便请求用户反馈，如文件中所示agent/main.py。
# MAIN API ENDPOINT: Handle stock analysis agent requests
# This endpoint receives investment queries and streams back real-time responses
@app.post("/agno-agent")
async def agno_agent(input_data: RunAgentInput):
try:
# ASYNC GENERATOR: Streams events to client in real-time
# This function generates a stream of events that get sent to the frontend
async def event_generator():
// ...
# Step 11: Process final workflow results and stream appropriate response
# Check if the last message from assistant contains tool calls or text
if agent_task.result().step_responses[-1].content['messages'][-1].role == "assistant":
if agent_task.result().step_responses[-1].content['messages'][-1].tool_calls:
# BRANCH A: Handle tool call responses (charts, analysis, etc.)
# for tool_call in state['messages'][-1].tool_calls:
# Step 12: Send tool call start event
yield encoder.encode(
ToolCallStartEvent(
type=EventType.TOOL_CALL_START,
tool_call_id=agent_task.result().step_responses[-1].content['messages'][-1].tool_calls[0].id,
toolCallName=agent_task.result().step_responses[-1].content['messages'][-1]
.tool_calls[0]
.function.name, # Name of function being called (e.g., render_charts)
)
)
# Step 13: Send tool call arguments
# Stream the arguments being passed to the tool/function
yield encoder.encode(
ToolCallArgsEvent(
type=EventType.TOOL_CALL_ARGS,
tool_call_id=agent_task.result().step_responses[-1].content['messages'][-1].tool_calls[0].id,
delta=agent_task.result().step_responses[-1].content['messages'][-1]
.tool_calls[0]
.function.arguments, # JSON arguments for the function call
)
)
# Step 14: Send tool call completion event
# Signals that the tool call has finished
the yield encoder.encode(
ToolCallEndEvent(
type=EventType.TOOL_CALL_END,
tool_call_id=agent_task.result().step_responses[-1].content['messages'][-1].tool_calls[0].id,
)
)
else:
// ...
// ...
except Exception as e:
# Step 23: Handle any errors during execution
print(e) # Log error for debugging
# Step 24: Return streaming response to client
# FastAPI will stream the events as Server-Sent Events (SSE)
return StreamingResponse(event_generator(), media_type="text/event-stream")
步骤 8：配置 AG-UI 协议文本消息事件
配置好 AG-UI 协议工具事件后，定义 AG-UI 协议文本消息事件，以便处理流式代理响应到前端，如agent/main.py文件中所示。
# MAIN API ENDPOINT: Handle stock analysis agent requests
# This endpoint receives investment queries and streams back real-time responses
@app.post("/agno-agent")
async def agno_agent(input_data: RunAgentInput):
try:
# ASYNC GENERATOR: Streams events to client in real-time
# This function generates a stream of events that get sent to the frontend
async def event_generator():
// ...
# Step 11: Process final workflow results and stream appropriate response
# Check if the last message from assistant contains tool calls or text
if agent_task.result().step_responses[-1].content['messages'][-1].role == "assistant":
// ...
else:
# BRANCH B: Handle text message responses
# Step 15: Start text message streaming
# Signal to UI that a text message is beginning
yield encoder.encode(
TextMessageStartEvent(
type=EventType.TEXT_MESSAGE_START,
message_id=message_id,
role="assistant", # Message from AI assistant
)
)
# Step 16: Stream message content (if available)
# Only send content event if content is not empty
if agent_task.result().step_responses[-1].content['messages'][-1].content:
content = agent_task.result().step_responses[-1].content['messages'][-1].content
# Step 17: Split message into chunks for streaming effect
# Split content into 100 parts
n_parts = 100
part_length = max(1, len(content) // n_parts) # Ensure at least 1 char per part
parts = [
content[i : i + part_length]
for i in range(0, len(content), part_length)
]
# Step 18: Handle edge case where splitting creates too many parts
# If splitting results in more than 5 due to rounding, merge the last parts
if len(parts) > n_parts:
parts = parts[: n_parts - 1] + [
"".join(parts[n_parts - 1 :])
]
# Step 19: Stream each content chunk with a delay for typing effect
for part in parts:
yield encoder.encode(
TextMessageContentEvent(
type=EventType.TEXT_MESSAGE_CONTENT,
message_id=message_id,
delta=part, # Chunk of message content
)
)
await asyncio.sleep(0.05) # Small delay for typing effect
else:
# Step 20: Handle case where no content was generated
# Send error message if content is empty
yield encoder.encode(
TextMessageContentEvent(
type=EventType.TEXT_MESSAGE_CONTENT,
message_id=message_id,
delta="Something went wrong! Please try again.",
)
)
# Step 21: End text message streaming
# Signal to UI that text message is complete
yield encoder.encode(
TextMessageEndEvent(
type=EventType.TEXT_MESSAGE_END,
message_id=message_id,
)
)
// ...
except Exception as e:
# Step 23: Handle any errors during execution
print(e) # Log error for debugging
# Step 24: Return streaming response to client
# FastAPI will stream the events as Server-Sent Events (SSE)
return StreamingResponse(event_generator(), media_type="text/event-stream")
恭喜！您已将 Agno 代理工作流与 AG-UI 协议集成。现在让我们看看如何为 AG-UI + Agno 代理工作流添加前端。
使用 CopilotKit 将前端集成到 AG-UI + Agno 代理工作流程中
在本节中，您将学习如何使用 CopilotKit 在 AG-UI + Agno 代理工作流程和前端之间建立连接。
我们开始吧。
首先，导航到前端目录：
cd frontend
接下来，创建一个 .env 包含OpenAI API密钥的文件：
OPENAI_API_KEY=<<your-OpenAI-key-here>>
然后安装依赖项：
pnpm install
之后，启动开发服务器：
pnpm run dev
访问 http://localhost:3000，您应该可以看到 AG-UI + Agno 代理前端已启动并运行。
现在让我们看看如何使用 CopilotKit 为 AG-UI + Agno 代理构建前端 UI。
步骤 1：创建 HttpAgent 实例
在创建 HttpAgent 实例之前，让我们先了解一下 HttpAgent 是什么。
HttpAgent 是 AG-UI 库中的一个客户端，它可以将您的前端应用程序与任何 AG-UI 兼容的 AI 代理服务器连接起来。
要创建 HttpAgent 实例，请在 API 路由中定义它，如src/app/api/copilotkit/route.ts文件中所示。
// Import CopilotKit runtime components for AI agent integration
import {
CopilotRuntime, // Core runtime for managing AI agents and conversations
copilotRuntimeNextJSAppRouterEndpoint, // Next.js App Router integration helper
OpenAIAdapter, // Adapter for OpenAI-compatible API endpoints
} from "@copilotkit/runtime";
// Import Next.js request type for proper TypeScript typing
import { NextRequest } from "next/server";
// Import HttpAgent for communicating with external AI agents
import { HttpAgent } from "@ag-ui/client";
// STEP 1: Initialize HTTP Agent for Stock Analysis Backend
// Create agent connection to our FastAPI stock analysis service
const agnoAgent = new HttpAgent({
// Use environment variable for backend URL, fallback to localhost
url: process.env.NEXT_PUBLIC_AGNO_URL || "http://0.0.0.0:8000/agno-agent",
});
// STEP 2: Configure OpenAI Service Adapter
// Set up adapter for OpenAI-compatible API communication
const serviceAdapter = new OpenAIAdapter();
// STEP 3: Initialize CopilotKit Runtime
// Create the main runtime that orchestrates AI agent interactions
const runtime = new CopilotRuntime({
agents: {
// Our FastAPI endpoint URL
// @ts-ignore - Suppress TypeScript error for agent configuration
agnoAgent: agnoAgent, // Register our stock analysis agent
},
});
// Alternative simple runtime configuration (commented out)
// const runtime = new CopilotRuntime()
// STEP 4: Define POST Request Handler
// Export async function to handle incoming POST requests from CopilotKit
export const POST = async (req: NextRequest) => {
// STEP 5: Create Request Handler with CopilotKit Integration
// Configure the endpoint handler with our runtime and service adapter
const { handleRequest } = copilotRuntimeNextJSAppRouterEndpoint({
runtime, // Our configured CopilotKit runtime with agents
serviceAdapter, // OpenAI adapter for LLM communication
endpoint: "/api/copilotkit", // This API route's endpoint path
});
// STEP 6: Process and Return Request
// Delegate request handling to CopilotKit's built-in handler
// This will route requests to appropriate agents and handle responses
return handleRequest(req);
};
步骤 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="agnoAgent">
{children}
</CopilotKit>
</body>
</html>
);
}
步骤 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 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>
);
}
步骤 4：使用 CopilotKit 钩子将 AG-UI + Agno 代理状态与前端同步
在 CopilotKit 中，CoAgents 维护着一个共享状态，该状态能够将前端 UI 与代理的执行无缝连接起来。这个共享状态系统使您能够：
-
显示代理的当前进度和中间结果
-
通过用户界面交互更新代理的状态
-
对应用程序中的状态变化做出实时反应
您可以在 CopilotKit 文档中了解更多关于 CoAgents 共享状态的信息 。
要将 AG-UI + Agno 代理状态与前端同步，请使用 CopilotKit useCoAgent hook 将 AG-UI + Agno 代理状态与前端共享，如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: "agnoAgent",
initialState: {
available_cash: totalCash,
investment_summary: {} as any,
investment_portfolio: [] as InvestmentPortfolio[],
},
});
// ...
return (
<div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
{/* ... */}
</div>
);
}
然后在聊天界面中渲染 AG-UI + Agno 代理的状态，以便以更贴合上下文的方式告知用户代理的状态。
要在聊天 UI 中渲染 AG-UI + Agno 代理的状态，可以使用 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: "agnoAgent",
render: ({ state }) => <ToolLogs logs={state.tool_logs} />,
});
// ...
return (
<div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
{/* ... */}
</div>
);
}
如果在聊天中执行查询，您应该会在聊天界面中看到 AG-UI + Agno 代理的状态任务执行情况，如下所示。
步骤 5：在前端实现人机交互（HITL）
人机交互（HITL）允许智能体在执行过程中请求人类输入或批准，从而提高人工智能系统的可靠性和可信度。对于需要处理复杂决策或需要人类判断的行动的人工智能应用而言，这种模式至关重要。
您可以在CopilotKit 文档中了解更多关于“人机交互”的信息。
要在前端实现人机交互（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>
);
}
当代理通过工具/操作名称触发前端操作，以在执行过程中请求人工输入或反馈时，最终用户会看到一个选择提示（显示在聊天界面中）。然后，用户可以通过按下聊天界面中的按钮进行选择，如下图所示。
步骤 6：在前端流式传输 AG-UI + Agno 代理响应
要将 AG-UI + Agno 代理的响应或结果流式传输到前端，请将代理的状态字段值传递给前端组件，如文件中所示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>
);
}
如果您向客服人员提出询问并批准其反馈请求，您应该会在用户界面中看到客服人员的回复或结果，如下所示。
结论
在本指南中，我们逐步介绍了如何将 Agno 代理与 AG-UI 协议集成，然后使用 CopilotKit 为代理添加前端。
虽然我们已经探索了一些功能，但我们仅仅触及了 CopilotKit 无数用例的冰山一角，从构建交互式 AI 聊天机器人到构建代理解决方案——本质上，CopilotKit 可以让您在几分钟内为您的产品添加大量有用的 AI 功能。
希望本指南能帮助您更轻松地将人工智能驱动的副驾驶功能集成到您现有的应用程序中。
在Twitter上关注 CopilotKit 并打个招呼，如果你想开发一些很酷的东西，请加入 Discord 社区。
文章来源：https://dev.to/copilotkit/build-your-own-ai-stock-portfolio-agent-with-agno-ag-ui-3ldp