OpenAI ChatKit 演示：宠物食品助手
A complete demonstration of OpenAI ChatKit featuring **text chat**, **voice assistant**, and **vision capabilities** for image analysis. This demo creates a comprehensive AI agent that can communicate through multiple modalities with a self-hosted ChatKit server and optional MCP server integration.🌟（星星） 多模态AI代理能力
这个演示展示了一个 完整的AI代理 具备三大强大功能：
🎤 语音助手
- 实时语音对话 使用OpenAI实时API
- 自然语音识别 以及响应生成
- 麦克风选择 以及音频设备管理
- 现场文字记录 在对话中展示
💬 文本聊天界面
- 传统的聊天界面 具有现代用户界面
- 富文本对话 配备宠物食品助手
- 文件和图片上传 支持
- 持久的对话历史
👁️ 视觉与图像分析
- 宠物照片分析 使用带有视觉功能的GPT-5
- 自动物种检测 （狗/猫身份识别）
- 品种识别 以及年龄估算
- 智能食品推荐 基于视觉分析
- 图像确认工作流程 为了提供准确的建议
🔄 翻译为中文是：循环/旋转（符号本身表示循环或重复的动作，具体翻译可能根据上下文有所变化） 无缝集成
- 统一的代理体验 跨越所有模式
- 一致的性格 以及知识库
- 跨模态上下文 理解
- 渐进增强 从文本→语音→视觉
🏗️ 建筑学
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Frontend │ │ OpenAI ChatKit │ │ MCP Server │
│ OpenAI ChatKit │ │ Server │ │ (Optional) │
│ JS │ │ Locally Hosted │ │ │
│ │ │ │ │ │
│ • Text Chat │◄──►│ • SQLite Store │◄──►│ • OpenSearch │
│ • Voice Panel │ │ • Pet Agent │ │ • Product DB │
│ • Image Upload │ │ • Vision Analysis│ │ • Tools │
│ • React + Vite │ │ • FastAPI │ │ │
└─────────────────┘ └──────────────────┘ └─────────────────┘🎯 这个项目实际做了什么
这个项目展示了如何构建一个 智能多模态宠物食品店助手 这可以通过三种不同的互动方式，帮助客户为他们的狗和猫找到最合适的食品：
📱 表示手机的符号，可翻译为“📱（手机符号）”或根据上下文简化为“手机”。在中文中，通常直接描述其含义为“手机”即可。 文本聊天体验
顾客可以输入他们的问题，比如“患有肾脏问题的老年猫最适合吃什么食物？”，并获得详细且个性化的建议，包括产品对比和营养信息。
🎙️（麦克风图标，常用于表示录音或直播等场景，可译为“麦克风”或根据上下文具体含义翻译） 语音助手体验
顾客可以进行自然语音对话，比如说出“我需要适合肠胃敏感的成年犬食用的食物”，并立即收到语音回复，其中包含产品推荐和解释。
📸（相机图标，常用于表示拍照或照片） 视觉与图像分析经验
顾客可以上传宠物的照片，然后助手会：
- 自动识别 物种（狗/猫）
- 识别这个品种 并估算年龄
- 确认细节 与客户一起
- 推荐合适的食物 基于视觉分析
当这些功能协同工作时，奇迹就发生了——客户可以从文本开始，切换到语音以进行更自然的对话，然后上传照片以进行精确的视觉分析，而这一切都是由同一个智能助手在所有交互中保持上下文连贯性。
*PETS INC演示界面展示了具备语音功能的现代聊天用户界面*
🚀 旅程：这是如何建成的
第一阶段：数据基础
我们首先从各种来源获取分布式、标准化的产品数据，并将其转换为单一的、非规范化结构，以优化其在OpenSearch中的表现。这包括以下几个步骤：
- 从多个来源（品牌、零售商、制造商）收集产品数据
- 规范化不一致的数据格式和命名约定
- 创建一个包含所有相关产品属性的统一模式
- 优化数据结构以实现快速、灵活的搜索
- 使用非规范化的产品文档构建OpenSearch索引
第二阶段：Python 代理
在我们的非规范化数据准备就绪后，我们构建了一个复杂的Python代理（pet_agent.py) 能够智能地浏览产品目录。这个代理足够聪明，能够：
- 理解自然语言查询
- 按物种（狗/猫）、年龄、健康状况等筛选产品。
- 提供个性化推荐
- 处理复杂的搜索场景
- 使用渐进式搜索策略来细化结果
第三阶段：网页界面
接下来，我们构建了一个React前端（ChatKitPanel.tsx它提供了一个美观的聊天界面。客户可以输入问题，并通过简洁现代的网页界面立即获得我们Python代理的回复。
第四阶段：声音革命
当我们将OpenAI的实时API集成进来时，取得了突破（VoicePanel.tsx）。 这使得顾客能够与助手进行自然的语音对话，就像与真正的宠物店员工交谈一样。
第五阶段：后端集成
我们创建了一个FastAPI服务器（session_server.py这个系统作为前端和搜索引擎之间的桥梁，能够无缝处理文本和语音请求。这一统一的后端确保了所有交互模式下的一致性表现。
🎭 用户体验
想象一下，你是一位宠物主人，正在为你的新小狗寻找食物。以下是会发生的情况：
- 你打开应用程序 并看到一个带有文本和语音选项的简洁界面
- 你点击语音按钮 然后说：“嗨，我需要给我的三个月大的小狗买食物。”
- 助手回应道 带着热情的问候：“嗨！我是Aya，您的宠物食品专家。我很乐意帮您为您的小狗找到最合适的食品！”
- 你继续对话 自然而然地说：“他是一只金毛寻回犬，看起来肠胃比较敏感。”
- 助手正在搜索 实时处理数千种产品
- 你会得到建议 附详细说明：“我发现了一些非常适合您肠胃敏感的金毛寻回犬幼犬的选择……”
整个体验就像在和一位非常了解宠物需求的宠物店员工聊天一样，他/她对您的宠物的需求了如指掌。
🧠 技术的魔力
搜索引擎
我们使用OpenSearch（类似于Elasticsearch）来存储和搜索产品数据。每个产品文档包含：
- 基本信息（名称、品牌、价格）
- 宠物的特定信息（种类、年龄、健康状况）
- 营养成分信息
- 客户评价和评分
- 用于语义搜索的向量嵌入
人工智能代理
我们的代理基于OpenAI的GPT-4构建，并采用了一种精细的提示工程方法：
- 它了解宠物营养与健康
- 它能够理解自然语言查询
- 它提供个性化推荐
- 在整个对话过程中，它保持上下文连贯
实时语音
语音组件使用了OpenAI的实时API，该API提供：
- 自然语音识别
- 实时响应生成
- 语音合成
- 无缝对话流程
🎯 为何这很重要
这个项目展示了几个重要概念：
- 基于人工智能的客户服务人工智能如何提供个性化且专业的帮助
- 多模态接口结合文本和语音以提升用户体验
- 实时搜索基于复杂标准的即时产品推荐
- 可扩展架构一个能够处理数千种产品和用户的系统
- 自然语言处理理解客户意图并提供相关回应
🚀 你可以学到什么
通过探索这个项目，你将了解到：
- 如何构建基于人工智能的客户服务代理
- 如何将语音功能集成到网页应用中
- 如何设计高效的搜索系统
- 如何打造个性化的用户体验
- 如何处理实时数据处理
🎉 最终结果
一个功能完备的宠物食品店助手，能够：
- 帮助客户为宠物找到最合适的食品
- 提供个性化推荐
- 处理文本和语音交互
- 扩展至数千种产品和数百万用户
- 提供一种自然、人性化的体验
这不仅仅是一个演示——它是一个构建智能、支持语音的客户服务应用的蓝图，能够彻底改变企业与客户互动的方式。
🛠️ 开发过程
我们克服的挑战
- 模式验证问题OpenAI实时API对工具模式有严格要求。我们不得不精心设计我们的Zod模式，以满足API的期望。
- 语音集成复杂性要集成语音功能，需要理解OpenAI的实时API，处理WebSocket连接，并管理音频流。
- 搜索表现我们拥有数千种产品，因此需要优化我们的OpenSearch查询，以便快速返回结果，同时排除诸如嵌入向量等不必要的数据。
- 跨平台兼容性确保系统能在不同浏览器和设备上正常运行，特别是语音功能。
- 实时数据流管理从语音输入→人工智能处理→搜索引擎→响应生成→语音输出的数据流。
关键的技术决策
- 非规范化数据我们选择将所有产品信息存储在一个单一文档中，而不是使用关系连接来加快搜索速度
- 渐进式搜索代理首先进行广泛搜索，然后根据用户输入逐步添加过滤条件
- 向量嵌入我们使用语义搜索来理解用户意图，即使他们没有使用确切的产品术语
- 实时架构我们构建了一个系统，能够通过同一个后端处理文本和语音请求
吸取的教训
- 从简单开始我们从一个基本的Python代理开始，然后逐步增加了复杂性
- 尽早测试语音集成需要进行广泛的测试以确保用户获得流畅的体验
- 优化性能搜索查询需要仔细优化以保持响应速度
- 扩展计划该架构从一开始就设计为能够应对增长
💼 商业价值与实际应用
这对企业为何至关重要
这个项目展示了人工智能如何改变客户服务：
- 全天候24/7可用顾客可以随时随地获得帮助
- 始终如一的质量每位客户都能获得同等高质量的服务支持
- 可扩展性一个AI代理可以同时处理数千名客户
- 成本效益减少了对大型客户服务团队的需求
- 个性化每次互动都根据客户的特定需求进行定制
实际应用场景
- 电子商务产品推荐、订单协助、退货服务
- 医疗保健症状自查、预约安排、用药指导
- 旅行行程规划、预订协助、旅行建议
- 金融账户管理、投资建议、贷款申请
- 教育课程推荐、学习路径、学术支持
竞争优势
采用人工智能客服技术的公司可以：
- 即时回应客户
- 提供个性化推荐
- 无需人工干预即可处理复杂查询
- 在不增加相应成本的情况下扩展他们的客户服务规模
- 从客户互动中收集有价值的见解
🚀 未来可能性
接下来怎么办？
这个项目为我们打开了许多令人兴奋的可能性的大门：
- 多语言支持扩展助手功能，以帮助不同语言的客户
- 高级分析追踪客户偏好和购买模式
- 与CRM（客户关系管理）系统的集成将客户互动与现有业务系统连接起来
- 手机应用程序为iOS和Android开发原生移动应用
- AR/VR集成让顾客在家就能直观看到产品效果
- 预测性推荐利用机器学习来预测客户可能需要什么
更广阔的图景
这只是个开始。随着人工智能技术的不断发展，我们可以期待：
- 更自然的对话
- 更好地理解上下文
- 增强个性化
- 与其他系统的无缝集成
- 增强的安全性和隐私保护
客户服务的未来已来，且由人工智能驱动。
🎯 这个演示展示了什么
- 文本聊天功能齐全的ChatKit用户界面，支持对话历史、附件和反馈功能
- 语音助手使用OpenAI的实时API进行实时语音对话
- 自托管服务器自定义ChatKit服务器，配备SQLite存储和宠物食品搜索功能
- MCP集成可选模型上下文协议服务器，用于增强功能
🚀 快速入门
先决条件
- Node.js 18岁以上及 npm（Node Package Manager，节点包管理器）
- python 3.12及以上版本 pip（在中文中通常保持原样，不直接翻译，若需解释性翻译，可译为“Python包管理工具”或“Python包安装程序”）
- OpenAI API密钥 使用ChatKit访问
- OpenSearch（开放搜索） 实例（对于MCP服务器）
1. 克隆并设置
git clone
cd mcpben
# Setup Python environment
cd ui
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -r requirements.txt
# Setup Node.js environment
cd chatkit-vite
npm install2. 配置环境
创建 ui/.env.local：
# OpenAI API Configuration
OPENAI_API_KEY=your_openai_api_key_here
# OpenSearch Configuration (for MCP server)
OS_HOST=localhost
OS_PORT=9200
OS_USER=admin
OS_PASS=admin
OS_INDEX=products_pets_v33. 启动服务
1号航站楼 - ChatKit服务器：
cd ui
source .venv/bin/activate
python chatkit_server_simple.py
# Server runs on http://localhost:9000第二航站楼 - MCP服务器（可选）：
python mcp_server.py
# MCP server runs on http://localhost:8000第三航站楼 - 前端：
cd ui/chatkit-vite
npm run dev
# Frontend runs on http://localhost:51734. 访问演示版
打开 http://localhost:5173，你将会看到：
- 文本聊天标签完整的ChatKit界面，包含对话历史记录
- 语音选项卡实时语音助手，支持麦克风选择
📱 前端功能
文本聊天（ChatKitPanel.tsx)
文本界面提供了完整的ChatKit体验：
// Key features implemented:
- Conversation history with SQLite persistence
- Custom headers (user-id, session-id, username)
- Error handling and debugging
- Theme support (light/dark)
- Attachment support
- Feedback system自定义头部配置：
headers: {
"user-id": getOrCreateDeviceId(),
"session-id": "session_123",
"username": "benno",
}语音助手（VoicePanel.tsx）
语音界面提供了实时对话功能：
// Key features implemented:
- Microphone device selection
- Real-time audio transcription
- OpenAI Realtime API integration
- Conversation transcripts
- Voice Activity Detection (VAD)
- Custom agent with tools语音代理配置：
const agent = new RealtimeAgent({
name: agentConfig.name,
instructions: agentConfig.instructions,
tools: agentConfig.tools, // Product search tools
});事件处理：
// Captures user speech
case "conversation.item.input_audio_transcription.completed":
addToTranscript("user", event.transcript);
// Captures AI responses
case "response.output_audio_transcript.done":
addToTranscript("assistant", fullText);🖥️ 自托管的ChatKit服务器
实施（chatkit_server_simple.py）
服务器遵循OpenAI官方的ChatKit文档规范：
核心组件
1. ChatKit 服务器类：
class PetChatKitServer(ChatKitServer):
async def respond(self, thread, input, context):
# Create agent context
agent_context = AgentContext(thread=thread, store=self.store)
# Convert input to agent format
agent_input = await simple_to_agent_input([input])
# Run agent with conversation continuity
result = Runner.run_streamed(
pet_asistant,
agent_input,
context=agent_context,
previous_response_id=previous_response_id
)
# Stream response back to client
async for event in stream_agent_response(agent_context, result):
yield event2. SQLite 存储实现：
class SimpleSQLiteStore(Store[Any]):
# Implements all required ChatKit store methods:
# - load_thread(), save_thread()
# - load_thread_items(), add_thread_item()
# - generate_thread_id(), generate_item_id()3. FastAPI 集成：
@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
# Extract context from headers
user_id = request.headers.get("user-id")
username = request.headers.get("username", "benno")
# Process with ChatKit server
result = await chatkit_server.process(await request.body(), context)
# Return streaming or JSON response
return StreamingResponse(result, media_type="text/event-stream")关键特性
- 对话连贯性用途
previous_response_id为了高效处理对话 - 自定义上下文将用户名和会话信息传递给代理
- 错误处理全面的日志记录和错误恢复
- 跨域资源共享（CORS）支持配置为前端集成
代理集成
服务器集成了一个自定义宠物食品助手代理：
# Agent with OpenSearch integration
from pet_agent import pet_asistant
# Agent provides:
# - Product search with filters
# - Conversation memory
# - Progressive search strategy
# - Personalized responses🔧 MCP 服务器（可选）
实施（mcp_server.py）
MCP服务器通过模型上下文协议提供了增强的功能：
# Key MCP tools implemented:
- search_products: OpenSearch integration
- get_product_details: Detailed product information
- get_recommendations: AI-powered recommendationsOpenSearch 集成：
async def search_products(query: str, species: str, filters: dict):
# Direct OpenSearch queries
# Filtered by species, life stage, price, etc.
# Returns structured product data🗄️ OpenSearch 后端与非规范化数据
产品数据架构
该演示使用了 OpenSearch（开放搜索） 作为后端搜索引擎，配备有 非规范化的产品数据结构 优化为快速、灵活的查询。这种方法消除了对复杂连接的需求，并提供了亚毫秒级的搜索性能。
非规范化产品模式
每个产品文档都包含所有相关信息，且采用单一、扁平化的结构呈现：
{
"_id": "prod_12345",
"title": "Royal Canin Adult Dog Food",
"description": "Complete nutrition for adult dogs...",
"brand": "Royal Canin",
"species": "Dog",
"life_stage": "adult",
"food_type": "dry",
"size": "15kg",
"price": 45.99,
"currency": "USD",
"availability": "in_stock",
"rating": 4.5,
"review_count": 1250,
"ingredients": [
"chicken meal",
"rice",
"corn gluten meal",
"chicken fat"
],
"nutritional_info": {
"protein_min": 22.0,
"fat_min": 12.0,
"fiber_max": 4.0,
"moisture_max": 10.0
},
"special_features": [
"grain_free",
"high_protein",
"digestive_support"
],
"target_conditions": [
"sensitive_stomach",
"weight_management"
],
"searchable_text": "royal canin adult dog food complete nutrition dry kibble chicken meal rice corn gluten meal high protein digestive support sensitive stomach weight management",
"embedding_product": [0.1234, 0.5678, ...], // Vector embeddings for semantic search
"created_at": "2024-01-15T10:30:00Z",
"updated_at": "2024-01-20T14:22:00Z"
}关键的反规范化优势
1. 单次查询性能：
# Instead of multiple JOINs across tables:
# SELECT p.*, b.name, n.protein_min, n.fat_min
# FROM products p
# JOIN brands b ON p.brand_id = b.id
# JOIN nutrition n ON p.id = n.product_id
# WHERE p.species = 'Dog' AND n.protein_min > 20
# OpenSearch single document query:
{
"query": {
"bool": {
"filter": [
{"term": {"species": "Dog"}},
{"range": {"nutritional_info.protein_min": {"gt": 20}}}
]
}
}
}2. 灵活过滤：
# Complex multi-dimensional filters in one query
filters = {
"species": "Dog",
"life_stage": "adult",
"food_type": "dry",
"price_max": 50,
"special_features": "grain_free",
"target_conditions": "sensitive_stomach"
}3. 全文搜索：
# Searchable text field combines multiple attributes
searchable_text = f"{title} {description} {brand} {' '.join(ingredients)} {' '.join(special_features)}"OpenSearch 查询实现
基本产品搜索
async def search_products_opensearch(query: str, species: str, filters: dict):
# Build OpenSearch query with denormalized data
search_body = {
"query": {
"bool": {
"should": [
{
"multi_match": {
"query": query,
"fields": ["title^2", "description", "searchable_text"],
"type": "best_fields",
"fuzziness": "AUTO"
}
},
{
"match": {
"searchable_text": {
"query": query,
"operator": "or"
}
}
}
],
"filter": [
{"term": {"species": species}}
],
"must_not": []
}
},
"size": 3,
"from": 0,
"sort": [{"_score": {"order": "desc"}}],
"_source": {
"excludes": ["embedding_product", "searchable_text", "updated_at", "embeddings"]
}
}高级过滤
# Progressive search strategy with denormalized filters
def build_opensearch_filters(filters: dict):
filter_clauses = []
# Life stage filtering
if "life_stage" in filters:
filter_clauses.append({
"terms": {
"life_stage": [filters["life_stage"], "All"]
}
})
# Price range filtering
if "price_max" in filters:
filter_clauses.append({
"range": {
"price": {"lte": filters["price_max"]}
}
})
# Special features (array field)
if "special_features" in filters:
filter_clauses.append({
"term": {
"special_features": filters["special_features"]
}
})
# Nutritional requirements
if "protein_min" in filters:
filter_clauses.append({
"range": {
"nutritional_info.protein_min": {"gte": filters["protein_min"]}
}
})
return filter_clauses使用嵌入进行语义搜索
# Vector similarity search using denormalized embeddings
def semantic_search(query_embedding: list, species: str):
return {
"query": {
"bool": {
"filter": [{"term": {"species": species}}],
"should": [
{
"knn": {
"field": "embedding_product",
"query_vector": query_embedding,
"k": 10,
"num_candidates": 100
}
}
]
}
}
}数据摄入管道
反规范化过程
def denormalize_product(product_data: dict) -> dict:
"""Convert normalized product data to denormalized OpenSearch document"""
# Flatten nested structures
nutritional_info = {
"protein_min": product_data.get("nutrition", {}).get("protein_min"),
"fat_min": product_data.get("nutrition", {}).get("fat_min"),
"fiber_max": product_data.get("nutrition", {}).get("fiber_max"),
"moisture_max": product_data.get("nutrition", {}).get("moisture_max")
}
# Create searchable text from multiple fields
searchable_text = " ".join([
product_data.get("title", ""),
product_data.get("description", ""),
product_data.get("brand", ""),
" ".join(product_data.get("ingredients", [])),
" ".join(product_data.get("special_features", [])),
" ".join(product_data.get("target_conditions", []))
]).lower()
# Generate vector embeddings
embedding_product = generate_embeddings(searchable_text)
return {
"_id": f"prod_{product_data['id']}",
"title": product_data["title"],
"description": product_data["description"],
"brand": product_data["brand"],
"species": product_data["species"],
"life_stage": product_data["life_stage"],
"food_type": product_data["food_type"],
"size": product_data["size"],
"price": product_data["price"],
"currency": product_data["currency"],
"availability": product_data["availability"],
"rating": product_data["rating"],
"review_count": product_data["review_count"],
"ingredients": product_data["ingredients"],
"nutritional_info": nutritional_info,
"special_features": product_data["special_features"],
"target_conditions": product_data["target_conditions"],
"searchable_text": searchable_text,
"embedding_product": embedding_product,
"created_at": product_data["created_at"],
"updated_at": product_data["updated_at"]
}性能优化
索引配置
{
"mappings": {
"properties": {
"title": {"type": "text", "analyzer": "standard"},
"description": {"type": "text", "analyzer": "standard"},
"searchable_text": {"type": "text", "analyzer": "standard"},
"species": {"type": "keyword"},
"life_stage": {"type": "keyword"},
"food_type": {"type": "keyword"},
"price": {"type": "float"},
"rating": {"type": "float"},
"ingredients": {"type": "keyword"},
"special_features": {"type": "keyword"},
"target_conditions": {"type": "keyword"},
"nutritional_info": {
"type": "object",
"properties": {
"protein_min": {"type": "float"},
"fat_min": {"type": "float"},
"fiber_max": {"type": "float"},
"moisture_max": {"type": "float"}
}
},
"embedding_product": {
"type": "dense_vector",
"dims": 1536,
"index": true,
"similarity": "cosine"
}
}
},
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0,
"refresh_interval": "30s"
}
}查询优化
# Exclude heavy fields from response
"_source": {
"excludes": [
"embedding_product", # Large vector data
"searchable_text", # Redundant with individual fields
"updated_at", # Not needed for display
"embeddings" # Multiple embedding variants
]
}
# Use filters for exact matches (faster than queries)
"filter": [
{"term": {"species": "Dog"}}, # Exact match
{"terms": {"life_stage": ["adult"]}}, # Multiple exact matches
{"range": {"price": {"lte": 50}}} # Range queries
]代理集成优势
去规范化的结构使智能体能够：
- 快速过滤即时物种、生命周期阶段和特征过滤
- 丰富的上下文所有产品信息均可通过一次查询获取
- 语义搜索“为‘为我找类似……的东西’计算向量相似度”
- 渐进式细化添加过滤器，无需额外查询
- 实时更新单文档更新保持一致性
# Agent can make complex queries efficiently
async def agent_search_products(query: str, species: str, user_preferences: dict):
# Single OpenSearch query with all filters
filters = {
"species": species,
"life_stage": user_preferences.get("life_stage"),
"food_type": user_preferences.get("food_type"),
"price_max": user_preferences.get("budget"),
"special_features": user_preferences.get("dietary_needs")
}
# One query returns complete product information
results = await search_products_opensearch(query, species, filters)
return results🎙️ 深入探索语音助手
设置过程
- 麦克风选择用户从可用音频设备中选择
- 瞬时密钥后端生成临时OpenAI API密钥
- 代理创建带有用户名和工具的动态代理
- WebRTC传输实时音频流
- 事件处理捕捉语音和回应
配置
语音检测设置：
turnDetection: {
type: "server_vad",
threshold: 0.5,
prefixPaddingMs: 300,
silenceDurationMs: 700,
createResponse: true,
}音频处理：
const stream = await navigator.mediaDevices.getUserMedia({
audio: {
deviceId: { exact: selectedDeviceId },
echoCancellation: true,
noiseSuppression: true,
autoGainControl: true,
}
});代理工具集成
语音代理包含与文本界面相同的产品搜索工具：
// Product search with voice input
const productSearchTool = tool({
name: "searchProducts",
description: "Search for pet food products...",
parameters: z.object({
query: z.string(),
species: z.enum(['Dog', 'Cat']),
filters: z.string().nullable(),
}),
execute: async ({ query, species, filters }) => {
// Calls backend API with voice-transcribed parameters
}
});🔄 前端-后端集成
文本聊天流程
- 用户输入 → ChatKit React 组件
- 自定义头部 → 包括用户ID、会话ID、用户名
- ChatKit 服务器 → 带有代理上下文的流程
- SQLite 存储 → 保存对话历史
- 流式响应 → 用户界面实时更新
语音聊天流程
- 麦克风访问权限 用户选择音频设备
- 临时密钥 后端生成临时API密钥
- 代理创建 带有用户名的动态代理
- WebRTC 连接 实时音频流
- 工具执行 语音指令触发产品搜索
- 转录本显示 → 用户界面中的对话历史
🛠️ 开发
添加新功能
1. 新的ChatKit工具：
# In pet_agent.py
@tool
async def new_tool(param: str) -> str:
return "Tool result"2. 前端定制化：
// In ChatKitPanel.tsx
const { control } = useChatKit({
// Add new configuration options
customFeature: {
enabled: true,
}
});3. 语音代理增强：
// In VoicePanel.tsx
const agent = new RealtimeAgent({
// Add new tools or modify instructions
tools: [...existingTools, newTool],
});调试
启用调试日志：
// In config.ts
export const FEATURES = {
showDebugLogs: true,
// ... other features
};服务器日志记录：
# Comprehensive logging throughout
logger.info(f"Processing request for thread {thread.id}")
logger.info(f"Agent input: {agent_input}")📊 性能考量
优化策略
- 对话连贯性用途
previous_response_id以避免重新加载全部历史记录 - 流式响应实时更新，不阻塞
- 音频处理带噪声抑制的高效WebRTC
- 数据库带有适当索引的SQLite，以实现快速查询
监测
- 服务器日志全面的请求/响应日志记录
- 前端控制台开发用的调试信息
- 错误处理在故障时优雅降级
🔐 安全
API密钥管理
- 瞬时密钥语音会话的临时密钥
- 环境变量安全存储API密钥
- CORS 配置限制在前端领域
数据隐私
- 本地存储用于对话历史的SQLite数据库
- 无外部日志记录敏感数据留在本地
- 用户控制明确的数据删除功能
🚀 部署
生产方面的考虑因素
- 环境变量安全的配置管理
- 数据库考虑在生产环境中使用 PostgreSQL
- 缩放具有负载均衡的多个服务器实例
- 监测应用程序性能监控
- 安全HTTPS、API速率限制、输入验证
Docker 部署
# Example Dockerfile for server
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "chatkit_server_simple.py"]📚 了解更多
🤝 贡献
这个演示是构建生产级ChatKit应用程序的基础。请随意：
- 添加新的代理工具和功能
- 通过增加更多功能来优化用户界面
- 实现更复杂的对话流程
- 添加与其他数据源的集成
📄 许可证
此项目仅用于演示目的。请确保您遵守OpenAI的使用政策和服务条款。