使用 CrewAI 和 Tavily 构建全栈 AI 购物助手
由 Mux 主办的 DEV 全球展示挑战赛：展示你的项目！
太长不看
本指南将教您如何使用 CrewAI Flows 构建一个全栈式 AI 购物助手，并结合后端 Tavily Search API。之后，我们将逐步讲解如何使用 CopilotKit 为购物助手添加前端界面，以便用户与之交互。
这款人工智能购物助手能够搜索亚马逊、塔吉特或eBay等电商平台上的产品，提取结构化的产品数据，并提供量身定制的产品推荐。
在正式开始之前，我们先来看看我们将要涵盖的内容：
-
CrewAI流程是什么？
-
使用 CrewAI、Tavily 和 CopilotKit 构建 AI 购物助手后端。
-
使用 CopilotKit 构建 AI 购物助手前端。
以下是我们将要构建的内容预览：
CrewAI流程是什么？
流程是 CrewAI 中一个灵活的构建模块，它允许您在保持整体简洁性的同时，对细节进行精细控制。您可以使用流程来自动化各种任务，从基本的 AI 模型调用，到能够自主工作的智能 AI 代理团队。
借助流程，可以轻松创建和设置逐步流程，从而最大限度地发挥 CrewAI 的功能。
以下是一些主要特点：
-
轻松构建工作流程：快速连接不同的 AI 代理团队（称为 Crews）和单个任务，以创建高级 AI 系统。
-
处理共享信息：流程图使跟踪和共享流程中各个步骤之间的数据变得非常简单。
-
基于事件的设计：它围绕着对事件的反应而构建，这有助于创建能够快速适应和响应的工作流程。
-
可自定义路径：添加“如果发生这种情况，则执行该操作”之类的规则，在循环中重复步骤，或在工作流程中拆分为不同的分支。
您可以在CrewAI 文档中了解更多关于 CrewAI 流程的信息。
现在我们已经了解了 CrewAI 流程是什么，让我们来看看如何从后端到前端构建 CrewAI 购物助手。
我们开始吧！
先决条件
要完全理解本教程，您需要对 React 或 Next.js 有一定的了解。
我们还将利用以下资源：
-
Python 是一种流行的编程语言，可用于使用 LangGraph 构建 AI 代理；请确保您的计算机上已安装 Python。
-
OpenAI API - 使我们能够使用 GPT 模型执行各种任务；对于本教程，请确保您有权访问 GPT-4 模型。
-
CopilotKit - 一个开源的辅助驾驶框架，用于构建自定义 AI 聊天机器人、应用内 AI 代理和文本区域。
-
CrewAI 是一个 Python 框架，它使开发人员能够以高度的简洁性和精确的底层控制来创建自主 AI 代理。
-
Tavily API——专为人工智能代理（LLM）构建的搜索引擎，能够快速提供实时、准确、真实的搜索结果。
项目设置
首先，克隆CrewAI-Shopping-Assistant仓库，该仓库包含一个基于 Python 的后端（代理）和一个 Next.js 前端（前端）。
接下来，导航到后端目录：
cd agent
然后使用 Poetry 安装依赖项：
poetry install
之后，创建一个 .env 包含OpenAI API 密钥和Tavily API 密钥的文件：
OPENAI_API_KEY=<<your-OpenAI-key-here>>
TAVILY_API_KEY=<<your-tavily-key-here>>
然后使用以下命令运行代理：
poetry run python main.py
之后，导航到前端目录：
cd frontend
接下来，创建一个 .env 包含 OpenAI API 密钥的文件：
OPENAI_API_KEY=<<your-OpenAI-key-here>>
然后安装依赖项：
pnpm install
之后，启动开发服务器：
pnpm run dev
访问 http://localhost:3000，您应该可以看到 AI 购物助手界面已启动并运行。
现在让我们看看如何使用 CrewAI、Tavily 和 CopilotKit 构建 AI 购物助手后端。
使用 CrewAI、Tavily 和 CopilotKit 构建 AI 购物助手后端。
在本节中，您将学习如何构建以 CrewAI 为代理流程、Tavily 为可靠网络搜索、CopilotKit 为状态管理和 UI 更新支持的 AI 购物助手。
让我们开始吧。
步骤 1：定义智能体状态
首先，定义一个AgentState继承类，CopilotKitState以便跟踪产品数据、用户偏好、UI 更新日志等，如agent/shopping_assistant.py文件中所示。
from copilotkit.crewai import CopilotKitState
from typing import List, Dict, Any
class AgentState(CopilotKitState):
"""
Manages the complete state of the shopping workflow.
"""
# List of products currently displayed on the canvas
products: List = []
# List of users' favorite/saved products
favorites: List = []
# Temporary buffer to store products before confirmation
buffer_products: List = []
# User's wishlist of products
wishlist: List = []
# Activity logs to show processing status to the user
logs: List = []
# Generated comparison report for products
report: Any | None = None
# Flag to control when to show results in UI
show_results: bool = False
# Canvas logging information with title and subtitle for UI updates
canvas_logs: dict = { "title" : "", "subtitle" : "" }
步骤 2：创建购物代理工作流程
定义代理状态后，定义ShoppingAgentFlow继承自该类的类Flow[AgentState]。使用该start()方法处理整个流程，其中初始化状态、验证环境变量、处理用户查询，并通过 CopilotKit 发出状态更新，如./agent/shopping_assistant.py文件中所示。
class ShoppingAgentFlow(Flow[AgentState]):
@start()
async def start(self):
"""
Main entry point for the shopping assistant workflow.
This method handles the entire flow from user request to product results.
"""
try:
# Step 1: Initialize the workflow
print("Starting Shopping Agent Flow")
# Update canvas with initial status
self.state.canvas_logs = {
"title" : f"Parsing your request",
"subtitle" : "Deciding to run product search or not"
}
await copilotkit_emit_state(self.state)
await asyncio.sleep(0)
# Step 2: Validate required environment variables
if not os.getenv("TAVILY_API_KEY"):
raise RuntimeError("Missing TAVILY_API_KEY")
if not os.getenv("OPENAI_API_KEY"):
raise RuntimeError("Missing OPENAI_API_KEY")
# Step 3: Check if this is a report generation request (assistant message)
if self.state.messages[-1]['role'] == 'assistant':
# Generate and return product comparison report
result =await generate_report(self.state.products)
print(result, "result")
self.state.report = json.loads(result)
await copilotkit_emit_state(self.state)
return
# Step 4: Add initial processing log
self.state.logs.append({
"message" : "Analyzing user query",
"status" : "processing"
})
await copilotkit_emit_state(self.state)
# Step 5: Mark analysis as completed
self.state.logs[-1]["status"] = "completed"
await copilotkit_emit_state(self.state)
// ...
步骤 3：使用 Tavily API 进行多零售商产品搜索
创建购物代理流程后，初始化 Tavily 客户端并准备进行多零售商搜索，如下所示。
class ShoppingAgentFlow(Flow[AgentState]):
@start()
async def start(self):
"""
Main entry point for the shopping assistant workflow.
This method handles the entire flow from user request to product results.
"""
try:
// ...
# Step 13: Set up Tavily search client and result containers
tv = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
results_all: List[Dict[str, Any]] = []
total_mappings_list = []
# Step 14: Update user with search progress
self.state.logs.append({
"message" : "Identifying the sites to search",
"status" : "processing"
})
self.state.canvas_logs={
"title" : "Identifying the sites to search",
"subtitle" : "Tavily search in progress...."
}
await copilotkit_emit_state(self.state)
await asyncio.sleep(1)
self.state.logs[-1]["status"] = "completed"
await copilotkit_emit_state(self.state)
// ...
然后使用 Tavily 的特定领域搜索功能搜索每个主要零售商，以获取每个主要零售商的相关产品 URL，如下所示。
class ShoppingAgentFlow(Flow[AgentState]):
@start()
async def start(self):
"""
Main entry point for the shopping assistant workflow.
This method handles the entire flow from user request to product results.
"""
try:
// ...
# Step 15: Search across major retailers for product URLs
urls = {}
for retailer in RETAILERS: # RETAILERS = ["target.com", "amazon.com", "ebay.com"]
# Search each retailer's domain for products matching the query
search = tv.search(
query=query, # User's search query, e.g., "wireless headphones"
include_domains=[retailer], # Limit search to specific retailer
include_answer=False, # Don't include general answers
include_images=False, # Don't include image results
include_raw_content=False, # We'll extract content separately
search_depth="advanced", # Use advanced search for better results
max_results=max_search_results, # Typically 6 results per retailer
)
# Extract URLs from search results
urls[retailer] = [r["url"] for r in search.get("results", []) if r.get("url")]
if not urls[retailer]:
continue
// ...
收集 URL 后，使用 Tavily 的提取功能获取包含图像和元数据的详细网页内容，如下所示。
class ShoppingAgentFlow(Flow[AgentState]):
@start()
async def start(self):
"""
Main entry point for the shopping assistant workflow.
This method handles the entire flow from user request to product results.
"""
try:
// ...
# Step 17: Define a function for parallel URL content extraction
def extract_urls(urls: List[str], retailer: str) -> Dict[str, Any]:
"""Extract content from URLs for a specific retailer"""
try:
print(f"Extracting urls for {retailer}. Started at {datetime.now()}")
ext1 = tv.extract(urls, extract_depth="advanced", include_images=True, timeout=120)
return [ext1, retailer]
except Exception as e:
print(f"Error extracting urls: {e}")
return None
# Step 18: Execute parallel extraction for all retailers
ext_results = {}
with ThreadPoolExecutor(max_workers=3) as executor:
# Submit extraction tasks for each retailer
futures = {executor.submit(extract_urls, urls[retailer], retailer) : retailer for retailer in RETAILERS}
# Collect results as they complete
for future in as_completed(futures):
result = future.result()
ext_results[result[1]] = result[0].get("results", [])
if result == None:
print("Condition met! Cancelling remaining tasks...")
# Cancel remaining futures if an error occurs
for f in futures:
f.cancel()
break
// ...
最后，使用 LLM 处理提取的产品数据以提取产品信息，如下所示。
class ShoppingAgentFlow(Flow[AgentState]):
@start()
async def start(self):
"""
Main entry point for the shopping assistant workflow.
This method handles the entire flow from user request to product results.
"""
try:
// ...
# Step 20: Initialize data structures for product processing
target_listing_pdps: List[str] = []
done = False
self.state.logs.append({
"message" : "Processing the data",
"status" : "processing"
})
await copilotkit_emit_state(self.state)
# Initialize product containers for each retailer
products_from_each_site= {
"target.com" : [],
"amazon.com" : [],
"ebay.com" : []
}
# Initialize URL replacement counters for each retailer
retailer_counters = {
"target.com": {"product": 0, "image": 0},
"amazon.com": {"product": 0, "image": 0},
"ebay.com": {"product": 0, "image": 0}
}
# Step 21: Define async function for processing extracted data
async def process_data(ext_results1: Dict[str, Any], retailer: str, retailer_counters: Dict[str, Dict[str, int]]) -> str:
"""Process extracted web content and convert to structured product data"""
print(f"Processing data for {retailer}. Started at {datetime.now()}")
for item in ext_results1:
url = item["url"]
raw = item.get("raw_content") or ""
if not raw:
return None
# Set up URL replacement patterns for each retailer
product_base = ""
image_base = ""
if retailer == "target.com":
product_base = "https://tgt.com/url{}"
image_base = "https://tgt.com/img/url{}"
elif retailer == "amazon.com":
product_base = "https://amzn.com/url{}"
image_base = "https://amzn.com/img/url{}"
elif retailer == "ebay.com":
product_base = "https://ebay.com/url{}"
image_base = "https://ebay.com/img/url{}"
# Replace URLs with standardized product and image links
modiefied_text, mappings_list, updated_product_counter, updated_image_counter = replace_urls_with_product_and_image_links(text= raw, product_base= product_base, image_base=image_base, product_counter=retailer_counters[retailer]["product"], image_counter=retailer_counters[retailer]["image"])
# Update counters
retailer_counters[retailer]["product"] = updated_product_counter
retailer_counters[retailer]["image"] = updated_image_counter
total_mappings_list.extend(mappings_list)
# Determine retailer and page type
dom = retailer_of(url)
detail_hint = is_pdp(url)
# Get structured data assistance for Target specifically
assist = parse_target_structured(modified_text) if "target.com" in dom else None
# Build prompt for LLM product extraction
prompt = build_llm_prompt(modiefied_text, url, assist=assist, detail_hint=detail_hint)
try:
# Limit products per retailer to avoid overwhelming results
if len(products_from_each_site[retailer]) > 2:
break
print(f"Calling LLM for {url}")
data = await call_llm(prompt)
print(f"Completed extracting {url}")
except Exception as e:
# Skip this page if LLM extraction fails
print(f"LLM 1st-pass failed for {url}: {e}")
continue
# Add metadata to extracted data
data.setdefault("source_url", url)
data.setdefault("retailer", dom)
products_from_each_site[retailer] += data["products"]
return "Completed"
// ...
步骤 4：配置人机交互功能
产品提取和处理完成后，通过工具调用前端将产品呈现给用户，如下所示，代理会暂停以允许用户查看产品并做出决定。
class ShoppingAgentFlow(Flow[AgentState]):
@start()
async def start(self):
"""
Main entry point for the shopping assistant workflow.
This method handles the entire flow from user request to product results.
"""
try:
// ...
# Step 24: Combine and finalize product results
results_all = combine_products_from_sites(products_from_each_site)
print(len(results_all), "results_all here")
# Add unique IDs to each product
for item in results_all:
item["id"] = str(uuid.uuid4())
# Mark data processing as completed
self.state.logs[-1]["status"] = "completed"
await copilotkit_emit_state(self.state)
# Step 25: Apply URL mappings and prepare final results
updated_products = apply_url_mappings_to_products(results_all, total_mappings_list)
print(len(updated_products), "updated_products here")
self.state.buffer_products = updated_products
print("HERE")
# Generate a descriptive name for this chat session
chat_name = await generate_name_for_chat(query)
print(chat_name, "chat_name here")
# Step 26: Send product list to user for confirmation
self.state.messages.append({
"role" : "assistant",
"content" : "",
"tool_calls" : [
{
"id" : str(uuid.uuid4()),
"function" : {
"name": "list_products",
"arguments": json.dumps({
"products": self.state.buffer_products[:5], # Show first 5 products
"buffer_products" : self.state.buffer_products, # Keep all products in buffer
"chat_name" : chat_name
})
}
}
]
})
# Step 27: Reset state for user interaction
self.state.logs = []
self.state.report = None
self.state.canvas_logs = {
"title" : "Awaiting confirmation from the user",
"subtitle" : "Choose to accept, reject or show all products"
}
await copilotkit_emit_state(self.state)
// ...
之后，当用户做出响应（通过工具调用）时，处理用户可能请求更多选项、接受或拒绝产品推荐的不同情况。
class ShoppingAgentFlow(Flow[AgentState]):
@start()
async def start(self):
"""
Main entry point for the shopping assistant workflow.
This method handles the entire flow from user request to product results.
"""
try:
// ...
# Step 7: Handle tool response messages
if(self.state.messages[-1]['role'] == 'tool'):
# Handle "Show more products" tool response
if(self.state.messages[-1]['content'] == "Show more products"):
self.state.messages.append({
"role" : "assistant",
"content" : "Some more products have also been added to be shown in the canvas",
"id" : self.state.messages[-2]['tool_calls'][0]['id']
})
self.state.logs = []
self.state.show_results = True
await copilotkit_emit_state(self.state)
return self.state
# Handle "Rejected" tool response
if(self.state.messages[-1]['content'] == "Rejected"):
self.state.messages.append({
"role" : "assistant",
"content" : "You have rejected the products. Please try any other product search.",
"id" : self.state.messages[-2]['tool_calls'][0]['id']
})
self.state.logs = []
await copilotkit_emit_state(self.state)
return self.state
# Handle "Accepted" tool response
if(self.state.messages[-1]['content'] == "Accepted"):
self.state.messages.append({
"role" : "assistant",
"content" : "The top 5 products have been added to the canvas.",
"id" : self.state.messages[-2]['tool_calls'][0]['id']
})
self.state.logs = []
self.state.show_results = True
await copilotkit_emit_state(self.state)
return self.state
// ...
步骤 5：设置 FastAPI 端点
创建完成后ShoppingAgentFlow，设置一个 FastAPI 服务器，作为我们的 CrewAI 购物代理和前端之间的桥梁。
为此，请使用 CopilotKitRemoteEndpoint 该工具管理代理的生命周期，然后使用该 add_fastapi_endpoint 函数将 CopilotKit 的路由挂载到我们的 FastAPI 应用程序，如agent/main.py文件中所示。
# Step 1: Import required dependencies
from fastapi import FastAPI # FastAPI framework for building the API server
import uvicorn # ASGI server for running FastAPI applications
# Step 2: Import CopilotKit components for agent integration
from copilotkit.integrations.fastapi import add_fastapi_endpoint # FastAPI integration
from copilotkit import CopilotKitRemoteEndpoint # Remote endpoint for agent hosting
from copilotkit.crewai.crewai_agent import CrewAIAgent # CrewAI agent wrapper
# Step 3: Import system dependencies
import os # Operating system interface for environment variables
# Step 4: Import the shopping assistant flow
from shopping_assistant import ShoppingAgentFlow # Custom shopping agent implementation
# Step 5: Initialize FastAPI application
app = FastAPI()
# Step 6: Configure CopilotKit Remote Endpoint with Shopping Agent
# This creates a remote endpoint that hosts the shopping assistant agent
sdk = CopilotKitRemoteEndpoint(
agents=[
# Step 7: Create CrewAI agent wrapper for the shopping flow
CrewAIAgent(
name="shopping_agent_crewai", # Unique identifier for the agent
description="A shopping agent that can help you find the best products for your needs by searching various retailers", # Agent description for CopilotKit
flow=ShoppingAgentFlow() # The actual shopping assistant workflow implementation
)
]
)
# Step 8: Add CopilotKit endpoint to FastAPI application
# This mounts the CopilotKit SDK at the "/copilotkit" path
add_fastapi_endpoint(app, sdk, "/copilotkit")
# Step 10: Main server configuration and startup function
def main():
"""
Configure and run the uvicorn server with development-friendly settings.
"""
# Step 11: Get server port from environment variable with fallback
port = int(os.getenv("PORT", "8000")) # Default to port 8000 if PORT not set
# Step 12: Start uvicorn server with comprehensive configuration
uvicorn.run(
"main:app", # Application module and instance
host="0.0.0.0", # Listen on all network interfaces
port=port, # Use configured port
reload=True, # Enable hot reload for development
timeout_keep_alive=900, # 15 minutes = 900 seconds (for long operations)
timeout_graceful_shutdown=900, # 15 minutes graceful shutdown
reload_dirs=(
# Step 13: Configure reload directories for development
["."] + # Current directory (always included)
(["../../../sdk-python/copilotkit"] # CopilotKit SDK directory if exists
if os.path.exists("../../../sdk-python/copilotkit")
else [] # Empty list if SDK directory doesn't exist
)
)
)
# Step 14: Entry point for running the server
if __name__ == "__main__":
# This ensures the server only starts when the script is run directly
# (not when imported as a module)
main()
恭喜！您已使用 CrewAI 实现代理流程，使用 Tavily 实现网络搜索，使用 CopilotKit 实现状态管理，构建了一个由人工智能驱动的购物助手后端。
使用 CopilotKit 构建 CrewAI 购物代理前端
在本节中，您将学习如何使用 CopilotKit 为 CrewAI 购物代理构建前端。
我们开始吧！
步骤 1：配置 Copilot 运行时实例
首先，设置 Copilot 运行时实例，该实例将作为前端和 CrewAI 购物代理后端之间的桥梁，如下面的文件所示frontend/app/api/copilotkit/route.ts。
/**
* CopilotKit API Route for Next.js App Router
*
* This file sets up the CopilotKit runtime endpoint for the AI Shopping Agent frontend.
* It handles POST requests to integrate with the CopilotKit service.
*/
import {
CopilotRuntime,
copilotRuntimeNextJSAppRouterEndpoint,
OpenAIAdapter,
} from "@copilotkit/runtime";
import { NextRequest } from "next/server";
// Create an OpenAI adapter for the CopilotKit service
const serviceAdapter = new OpenAIAdapter();
// Initialize the CopilotRuntime with remote endpoints
// This connects to the shopping assistant backend
const runtime = new CopilotRuntime({
remoteEndpoints: [
{
url:
process.env.NEXT_PUBLIC_SHOPPING_AGENT_URL ||
"http://localhost:8000/copilotkit",
},
],
});
// POST handler for the CopilotKit API endpoint
// This function processes incoming requests and delegates to the CopilotKit runtime
export const POST = async (req: NextRequest) => {
const { handleRequest } = copilotRuntimeNextJSAppRouterEndpoint({
runtime,
serviceAdapter,
endpoint: "/api/copilotkit",
});
return handleRequest(req);
};
步骤 2：设置 CopilotKit 提供程序
设置好 Copilot Runtime 实例后，设置管理 ADK 代理会话的 CopilotKit 提供程序组件。
要设置 CopilotKit 提供程序， CopilotKit 组件必须封装应用程序中与 Copilot 相关的部分。
对于大多数使用场景，最好将 CopilotKit 提供程序包装在整个应用程序周围，例如，在您的 layout.tsx文件中。
/**
* Root Layout Component for the AI Shopping Agent
*
* This is the main layout component for the Next.js application.
* It sets up the CopilotKit provider and global styles for the entire app.
*/
import type { Metadata } from "next";
import { GeistSans } from "geist/font/sans";
import { GeistMono } from "geist/font/mono";
import "./globals.css";
import "@copilotkit/react-ui/styles.css";
import { CopilotKit } from "@copilotkit/react-core";
// Metadata configuration for the application
export const metadata: Metadata = {
title: "Shopping Assistant",
description: "Powered by LangGraph and CopilotKit",
generator: "v0.app",
};
/**
* Root Layout Component
*
* This component wraps the entire application and provides:
* - CopilotKit context for AI-powered features
* - Global font configuration
* - Base HTML structure
*/
export default function RootLayout({
children,
}: Readonly<{
children: React.ReactNode;
}>) {
return (
<html lang="en">
<head>
{/* Custom font configuration using Geist fonts */}
<style>{`
html {
font-family: ${GeistSans.style.fontFamily};
--font-sans: ${GeistSans.variable};
--font-mono: ${GeistMono.variable};
}
`}</style>
</head>
<body>
{/* CopilotKit Provider - enables AI chat and agent functionality */}
<CopilotKit
runtimeUrl="/api/copilotkit" // Points to our CopilotKit API route
showDevConsole={false} // Hide development console in production
agent="shopping_agent_crewai" // Specify which backend agent to use
>
{children}
</CopilotKit>
</body>
</html>
);
}
步骤 3：设置 Copilot 聊天组件
CopilotKit 附带几个内置聊天组件，包括 CopilotPopup、 CopilotSidebar和 CopilotChat。
要设置 Copilot 聊天组件，请按照 frontend/components/sidebar.tsx 文件中所示进行定义。
/**
* Sidebar Component for AI Shopping Assistant
*
* This component provides the main navigation and chat interface for the shopping assistant.
* It includes chat session management, view switching, and integrates with CopilotKit for AI interactions.
*/
"use client";
import type React from "react";
import { Button } from "@/components/ui/button";
import { Input } from "@/components/ui/input";
import { Badge } from "@/components/ui/badge";
import { ScrollArea } from "@/components/ui/scroll-area";
import { CopilotChat } from "@copilotkit/react-ui";
import { useCopilotChat } from "@copilotkit/react-core";
// ...
/**
* Main Sidebar Component
*
* Features:
* - Chat session management (create, switch, rename, delete)
* - View switching between products, wishlist, and reports
* - Integrated CopilotChat for AI interactions
* - Product comparison analysis generation
*/
export function Sidebar({
setQuery,
isLoading,
clearState,
onSearch,
suggestions,
currentQuery,
isSearching,
currentView,
wishlistCount,
goToProducts,
currentChatId,
chatSessions,
onSwitchChat,
onCreateNewChat,
onRenameChat,
onDeleteChat,
}: SidebarProps) {
// ...
return (
<div className="flex flex-col min-h-screen w-80 bg-[#FAFCFA] border-r border-[#D8D8E5]">
{/* Header */}
<div className="p-4 border-b border-[#D8D8E5] bg-white">
<div className="flex items-center gap-3 mb-3">
<div className="text-2xl">🪁</div>
<div className="flex-1">
<h1 className="text-lg font-semibold text-[#030507] font-['Roobert']">
Shopping Assistant
</h1>
<Badge
variant="secondary"
className="text-xs bg-[#BEC9FF] text-[#030507] font-semibold">
PRO
</Badge>
</div>
</div>
// ...
<div className="flex-1 overflow-auto">
<CopilotChat
className="h-full"
labels={{
initial:
"Hi! I'm your AI shopping assistant. I can help you find and compare products across multiple websites. What are you looking for today? Laptops, Phones, Headphones, etc.",
}}
/>
</div>
</div>
);
}
步骤 4：使用 CopilotKit 钩子将 CrewAI 购物代理的状态与前端同步
在 CopilotKit 中，CoAgents 维护着一个共享状态，该状态能够将前端 UI 与代理的执行无缝连接起来。这个共享状态系统使您能够：
-
显示代理的当前进度和中间结果
-
通过用户界面交互更新代理的状态
-
对应用程序中的状态变化做出实时反应
您可以在 CopilotKit 文档中了解更多关于 CoAgents 共享状态的信息 。
要将您的 CrewAI 购物代理状态与前端同步，请使用 CopilotKit useCoAgent hook，如文件中所示 frontend/components/shopping-assistant.tsx 。
"use client"
import { useCoAgent } from "@copilotkit/react-core"
export function ShoppingAssistant() {
// ...
const { state, setState, start, run } = useCoAgent({
name: "shopping_agent_crewai",
initialState: conversationHistory.length > 0 ? { ...conversationHistory[0]?.state, show_results: (conversationHistory[0]?.state?.products?.length > 0 ? true : false) } : {
products: [],
favorites: typeof window !== 'undefined' && window.localStorage.getItem("wishlist") ? JSON.parse(window.localStorage.getItem("wishlist") || "[]") : [],
wishlist: typeof window !== 'undefined' && window.localStorage.getItem("wishlist") ? JSON.parse(window.localStorage.getItem("wishlist") || "[]") : [],
buffer_products: [],
logs: [] as ToolLog[],
report: null,
show_results: false,
canvas_logs: {
title: "",
subtitle: ""
}
}
})
return (
<div className="flex h-screen bg-[#FAFCFA] overflow-hidden">
// ...
<div className="flex-1 flex flex-col min-w-0">
{currentView === "report" ? (
<ReportView isLoading={isLoading} products={state?.products} onExit={exitToProducts} searchQuery={query} report={state?.report} />
) : currentView === "wishlist" ? (
// ...
) : (
// ...
)}
</div>
</div>
)
}
然后在聊天界面中渲染 CrewAI 购物代理的状态，这有助于以更贴合上下文的方式告知用户代理的状态。
要在聊天 UI 中渲染 CrewAI 购物代理的状态，您可以使用 文件 中的 useCoAgentStateRenderfrontend/components/shopping-assistant.tsx钩子 。
"use client"
import { useCoAgentStateRender} from "@copilotkit/react-core"
import { ToolLog, ToolLogs } from "./tool-logs"
export function ShoppingAssistant() {
// ...
useCoAgentStateRender({
name: "shopping_agent_crewai",
render: (state1: any) => {
// useEffect(() => {
// console.log(state1, "state1")
// }, [state1])
return <ToolLogs logs={state1?.state?.logs || []} />
}
})
return (
<div className="flex h-screen bg-[#FAFCFA] overflow-hidden">
// ...
<div className="flex-1 flex flex-col min-w-0">
{currentView === "report" ? (
<ReportView isLoading={isLoading} products={state?.products} onExit={exitToProducts} searchQuery={query} report={state?.report} />
) : currentView === "wishlist" ? (
// ...
) : (
// ...
)}
</div>
</div>
)
}
如果在聊天中执行查询，您应该会在聊天界面中看到 CrewAI 购物代理的状态任务执行情况，如下所示。
步骤 5：在前端实现人机交互（HITL）
人机交互（HITL）允许智能体在执行过程中请求人类输入或批准，从而提高人工智能系统的可靠性和可信度。对于需要处理复杂决策或需要人类判断的行动的人工智能应用而言，这种模式至关重要。
您可以在 CopilotKit 文档中了解更多关于“人机交互”的信息 。
要在前端实现人机交互（HITL），您需要使用 CopilotKit 的 useCopilotKitAction 钩子 renderAndWaitForResponse 方法，该方法允许从渲染函数异步返回值，如 frontend/components/shopping-assistant.tsx 文件中所示。
"use client"
import { useCopilotAction } from "@copilotkit/react-core"
import DialogBox from "./tool-response"
export function ShoppingAssistant() {
// ...
useCopilotAction({
name: "list_products",
description: "A list of products that are scraped from the web",
renderAndWaitForResponse: ({ status, respond, args }) => {
// console.log(args, "argsargsargsargs")
return <DialogBox isDisabled={respond == undefined} contentList={args?.products?.map((product: any) => ({ title: product.title, url: product.product_url }))}
onAccept={() => {
debugger
if (respond) {
respond("Accepted")
setState({
...state,
products: args?.products,
buffer_products: args?.buffer_products.slice(5, args?.buffer_products.length),
logs: []
})
let conversations = conversationHistory
conversations.forEach((conversation: any) => {
if (conversation.conversationId == currentChatId) {
conversation.chatName = args?.chat_name
}
})
setConversationHistory(conversations)
setCurrentChatId(currentChatId)
console.log(currentChatId, "currentChatId");
// setConversationHistory((prev: any) => prev.map((conversation: any) => conversation.conversationId === currentChatId ? { ...conversation, chatName: args?.chat_name } : conversation))
// setProducts(args?.products)
}
}}
onReject={() => {
if (respond) {
respond("Rejected")
setState({
...state,
logs: []
})
}
}}
onNeedInfo={() => {
debugger
if (respond) {
respond("Show more products")
setState({
...state,
products: args?.buffer_products?.slice(0, 10),
buffer_products: args?.buffer_products.slice(10, args?.buffer_products.length),
logs: []
})
let conversations = conversationHistory
conversations.forEach((conversation: any) => {
if (conversation.conversationId === currentChatId) {
conversation.chatName = args?.chat_name
}
})
console.log(currentChatId, "currentChatId");
setConversationHistory(conversations)
setCurrentChatId(currentChatId)
// setConversationHistory((prev: any) => prev.map((conversation: any) => conversation.conversationId === currentChatId ? { ...conversation, chatName: args?.chat_name } : conversation))
// setProducts(args?.buffer_products?.slice(0, 10))
}
}} />
}
})
return (
<div className="flex h-screen bg-[#FAFCFA] overflow-hidden">
// ...
<div className="flex-1 flex flex-col min-w-0">
{currentView === "report" ? (
<ReportView isLoading={isLoading} products={state?.products} onExit={exitToProducts} searchQuery={query} report={state?.report} />
) : currentView === "wishlist" ? (
// ...
) : (
// ...
)}
</div>
</div>
)
}
当代理通过工具/操作名称触发前端操作，以在执行过程中请求人工输入或反馈时，最终用户会看到一个选择提示（显示在聊天界面中）。然后，用户可以通过按下聊天界面中的按钮进行选择，如下图所示。
步骤 6：在前端流式传输 AG-UI + CrewAI 代理响应
要在前端流式传输 CrewAI 购物代理的响应或结果，请将代理的状态字段值传递给前端组件，如文件中所示 frontend/components/shopping-assistant.tsx 。
"use client"
import { useCoAgent } from "@copilotkit/react-core"
export function ShoppingAssistant() {
// ...
const { state, setState, start, run } = useCoAgent({
name: "shopping_agent_crewai",
initialState: conversationHistory.length > 0 ? { ...conversationHistory[0]?.state, show_results: (conversationHistory[0]?.state?.products?.length > 0 ? true : false) } : {
products: [],
favorites: typeof window !== 'undefined' && window.localStorage.getItem("wishlist") ? JSON.parse(window.localStorage.getItem("wishlist") || "[]") : [],
wishlist: typeof window !== 'undefined' && window.localStorage.getItem("wishlist") ? JSON.parse(window.localStorage.getItem("wishlist") || "[]") : [],
buffer_products: [],
logs: [] as ToolLog[],
report: null,
show_results: false,
canvas_logs: {
title: "",
subtitle: ""
}
}
})
return (
<div className="flex h-screen bg-[#FAFCFA] overflow-hidden">
// ...
<div className="flex-1 flex flex-col min-w-0">
{currentView === "report" ? (
<ReportView isLoading={isLoading} products={state?.products} onExit={exitToProducts} searchQuery={query} report={state?.report} />
) : currentView === "wishlist" ? (
<WishlistView
clearAllWishlist={() => {
debugger
setState({
...state,
favorites: []
})
setConversationHistory((prev: any) => prev.map((conversation: any) => conversation.conversationId === currentChatId ? { ...conversation, state: { ...conversation.state, favorites: [] } } : conversation))
if (typeof window !== 'undefined') {
window.localStorage.setItem("wishlist", JSON.stringify([]))
}
}}
products={state?.favorites}
onExit={exitToProducts}
onToggleWishlist={toggleWishlist}
onDeleteProduct={deleteProduct}
/>
) : (
<Canvas
canvasLogs={state?.canvas_logs}
start={run}
show_results={state?.show_results}
report={state?.report}
products={state?.products}
isLoading={isLoading && !state?.show_results}
query={query}
wishlistLength={state?.favorites?.length}
wishlist={state?.favorites}
onToggleWishlist={toggleWishlist}
onDeleteProduct={deleteProduct}
onGoToWishlist={goToWishlist}
onGoToReport={goToReport}
/>
)}
</div>
</div>
)
}
如果您向客服人员提出询问并批准其反馈请求，您应该会在用户界面中看到客服人员的回复或结果，如下所示。
结论
在本指南中，我们逐步介绍了如何使用 CrewAI 构建全栈购物助手，然后使用 CopilotKit 为代理添加前端。
虽然我们已经探索了一些功能，但我们仅仅触及了 CopilotKit 无数用例的冰山一角，从构建交互式 AI 聊天机器人到构建代理解决方案——本质上，CopilotKit 可以让您在几分钟内为您的产品添加大量有用的 AI 功能。
希望本指南能帮助您更轻松地将人工智能驱动的副驾驶功能集成到您现有的应用程序中。
在Twitter上关注 CopilotKit 并打个招呼，如果你想开发一些很酷的东西，请加入 Discord 社区。
文章来源：https://dev.to/copilotkit/building-a-full-stack-ai-shopping-assistant-with-crewai-and-tavily-4366