基于OpenAI Realtime API构建实时AI智能体：从原理到实践

1. 项目概述：实时AI智能体的新范式

最近在AI应用开发圈里，一个名为openai-realtime-agents的项目悄然走红。这并非一个全新的独立产品，而是OpenAI官方在GitHub上开源的一套示例代码库。但千万别小看它，这个项目精准地指向了当前AI应用开发最前沿、也最令人兴奋的一个方向：实时、多模态、具备自主行动能力的智能体系统。简单来说，它提供了一个基于OpenAI Realtime API构建智能体的“官方最佳实践”脚手架。

如果你还在用传统的“一问一答”模式调用GPT，或者觉得构建一个能听、能说、能看、能思考、能执行复杂任务的AI助手门槛太高，那么这个项目就是你绝佳的起点。它把OpenAI最新的Realtime API、GPT-4o模型、以及函数调用（Function Calling）等能力，封装成了一个结构清晰、易于扩展的框架。开发者可以基于此，快速搭建出能处理音频流、视频流，并能在对话中自主调用工具完成任务的智能体。

这个项目的核心价值在于“降本增效”。它解决了几个关键痛点：一是实时流式交互的复杂性，二是多模态信息（语音、文本、视觉）的统一处理，三是智能体决策与工具执行的循环逻辑。通过研究这个官方示例，我们能最直接地理解OpenAI对于下一代AI应用形态的设想，并快速将这种能力应用到客服机器人、虚拟助手、智能导购、教育陪练等无数场景中。接下来，我们就深入拆解这个项目的设计精髓与实现细节。

2. 核心架构与设计哲学解析

2.1 为何是“Realtime” + “Agents”？

要理解这个项目，必须先厘清两个核心概念：“Realtime”和“Agents”。在OpenAI的语境下，Realtime API与传统Chat Completions API的最大区别在于通信模式。传统API是“请求-响应”式，你发送一整段文本，模型处理完后返回一整段结果。而Realtime API建立了持久的WebSocket连接，支持双向、低延迟的流式通信。这意味着智能体可以像真人一样，听到你说话的开头就开始思考，并逐步生成回复，实现真正的“实时对话”体验，这对于语音交互场景至关重要。

而Agents（智能体）在这里特指具备“思考-行动”循环的AI系统。它不仅仅是语言模型，更是一个自主的决策者。智能体接收来自用户的多模态输入（语音转文字、图像描述），通过模型“思考”（生成推理内容），决定是否需要调用外部工具（如查询数据库、执行计算、控制设备），执行工具，并根据工具返回的结果组织最终的回复输出。openai-realtime-agents项目完美地将这两者结合，构建了一个既能实时交互，又能自主使用工具的智能体基础框架。

2.2 项目整体架构拆解

打开项目仓库，其目录结构就体现了清晰的分层设计思想。虽然具体文件可能迭代，但其核心模块通常包括：

1. 核心会话管理器：这是大脑中枢，负责维护与OpenAI Realtime API的WebSocket连接，管理会话状态，并作为所有事件（用户输入、工具调用结果、模型回复）的调度中心。
2. 输入/输出处理器：负责处理多模态数据。对于输入，它需要整合语音流（通过SDK或前端采集）并转换为音频事件，或处理直接输入的文本。对于输出，它需要将模型返回的文本内容，通过TTS（文本转语音）合成音频流播放给用户，或渲染为文本信息。
3. 工具系统：这是智能体能力的扩展库。项目会定义一套工具的标准接口（例如，一个工具包含名称、描述、参数JSON Schema）。智能体在思考时，如果判断需要调用工具，就会生成一个工具调用请求。会话管理器捕获这个请求，在本地执行对应的工具函数，并将执行结果返回给模型，让模型基于结果生成最终回复。
4. 事件循环与状态机：这是智能体流畅运行的关键。整个交互过程被建模为一系列事件的顺序处理，例如conversation.item.created（新消息项）、response.text.delta（模型回复文本流）、response.function_call_arguments.delta（工具调用参数流）等。项目需要稳健地处理这些事件流，并管理好智能体的状态（如是否正在等待工具执行结果）。

这种架构的优势在于高内聚、低耦合。工具系统可以独立扩展，添加新的能力只需定义新的工具函数；输入输出层可以替换，比如从网页语音交互切换到电话线路接口；而核心的智能体逻辑保持不变。

注意：官方示例为了清晰展示原理，可能会将部分逻辑写在一起。在实际生产级开发中，我们通常会将工具注册、会话管理、业务逻辑进一步分离，以提高可维护性。

3. 关键技术点深度剖析

3.1 Realtime API的事件驱动模型

这是项目中最具技术挑战性的部分之一。与简单的HTTP调用不同，WebSocket连接上流动的是一系列结构化的事件。你需要监听这些事件，并做出正确响应。主要的事件类型包括：

• 会话事件：如session.created，表示连接建立。
• 输入事件：如input_audio_buffer.committed，表示一段用户音频已准备就绪，可发送给模型。或者input_audio_buffer.speech_started，用于检测用户何时开始说话，实现VAD（语音活动检测）。
• 项目事件：如conversation.item.created，表示对话中新增了一个条目（可能是用户消息或助理回复）。
• 响应事件：这是核心。response.text.delta是模型返回的文本流，你需要实时拼接并显示（或缓存起来用于后续TTS）。response.function_call_arguments.delta更关键，它流式地传输工具调用的参数，你需要实时解析（通常是JSON片段），直到事件完成，才能得到一个完整的、可执行的工具调用请求。
• 工具调用事件：如response.function_call_arguments.done，表示工具参数流传输完毕，此时应执行本地工具。

处理这些事件的关键是状态管理。例如，当模型开始流式输出response.function_call_arguments.delta时，智能体应进入“等待工具参数”状态，并开始累积参数片段。一旦收到done事件，就立即解析完整参数，调用工具，然后将工具执行结果作为新的输入项发送回会话。整个过程中，要避免事件处理阻塞主线程，通常需要异步编程模型（如Python的asyncio）。

3.2 工具调用（Function Calling）的实现与优化

工具调用是智能体“动手能力”的体现。在openai-realtime-agents中，工具的定义遵循OpenAI的函数调用规范。你需要为每个工具提供一个详细的描述和参数模式，模型会根据这个描述来决定何时以及如何调用它。

一个典型的工具定义示例（概念性代码）:

def get_weather(location: str, unit: str = “celsius”): “”“获取指定城市的当前天气情况。 Args: location: 城市名，例如“北京”、“San Francisco”。 unit: 温度单位，“celsius” 或 “fahrenheit”。 ”“” # 模拟或真实调用天气API return f”{location}的天气是晴朗，温度25{‘°C’ if unit == ‘celsius’ else ‘°F’}。” # 向会话注册的工具列表 tools = [ { “type”: “function”, “function”: { “name”: “get_weather”, “description”: “获取某个城市的当前天气。”, “parameters”: { “type”: “object”, “properties”: { “location”: {“type”: “string”, “description”: “城市名称”}, “unit”: {“type”: “string”, “enum”: [“celsius”, “fahrenheit”], “description”: “温度单位”} }, “required”: [“location”] } } } ]

实操心得：工具描述的“艺术”工具的描述和参数描述至关重要，它直接决定了模型调用工具的准确率。描述要清晰、无歧义、并说明适用场景。例如，“获取天气”比“查询气象信息”更好；“城市名称”比“地点”更明确。有时，你甚至需要在描述中加入一些调用示例的暗示，比如“用于回答用户关于气温、湿度、是否下雨等问题”。

另一个关键优化点是错误处理。工具执行可能会失败（如网络超时、参数无效）。项目需要设计机制，将工具执行错误信息友好地反馈给模型，让模型能够向用户解释或尝试其他方案。例如，捕获异常后，生成一个如{"error": "天气服务暂时不可用，请稍后再试。"}的结果返回给会话，模型通常会据此生成得体的用户回复。

3.3 多模态输入的处理流程

项目支持语音作为主要输入方式，这涉及音频流处理管线：

1. 采集：通过浏览器getUserMedia或系统的音频接口采集原始PCM音频数据。
2. 缓冲与提交：音频数据以小块（例如每100ms）缓冲。当检测到用户停止说话（VAD）或达到最大缓冲时长时，将这段音频缓冲区“提交”（commit）。在Realtime API中，这对应着生成一个input_audio_buffer.append事件，然后input_audio_buffer.commit。
3. 转录与理解：提交的音频数据通过Realtime API发送到云端，由 Whisper 或类似模型进行实时语音识别（STT），转换为文本。这段文本作为新的对话条目加入上下文，并由GPT-4o等模型进行理解。
4. 视觉信息整合：如果项目扩展了视觉能力（GPT-4o支持图像输入），那么处理流程还需包含图像编码。前端需要将摄像头捕获的图像帧转换为base64编码，并通过input_image等事件类型发送。模型会同时理解文本和图像内容，实现“看图说话”或基于视觉信息的决策。

注意事项：延迟与用户体验实时音频处理的挑战在于延迟。从用户停止说话，到音频上传、转录、模型推理、生成回复、TTS合成、播放，这个链条的延迟必须控制在可接受范围内（通常希望小于1-2秒）。优化方法包括：使用更快的模型（如GPT-4o-turbo）、在客户端进行轻量级VAD以减少无效音频上传、以及使用流式TTS让智能体可以一边生成文本一边开始播报开头部分。

4. 从零开始构建你的第一个实时智能体

4.1 环境准备与依赖安装

假设我们使用Python作为后端开发语言，前端使用简单的HTML/JavaScript进行语音交互。首先，确保你拥有OpenAI账户并已生成API密钥，且该密钥有权限访问Realtime API（通常需要单独申请或加入等待列表）。

后端环境 (Python):

# 创建项目目录并进入 mkdir my-realtime-agent cd my-realtime-agent # 创建虚拟环境（推荐） python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install openai httpx websockets python-dotenv # 可能还需要安装异步框架，如 fastapi 和 uvicorn 用于提供Web服务 pip install fastapi uvicorn

创建一个.env文件来安全存储你的API密钥：

OPENAI_API_KEY=sk-your-secret-key-here

前端环境：一个简单的index.html文件即可，利用现代浏览器的Web Audio API和WebSocket能力。我们还需要一个后端服务来中转前端和OpenAI Realtime API之间的通信（因为前端直接调用OpenAI API存在暴露密钥的风险）。

4.2 核心后端服务搭建

后端的主要职责是：1) 为前端提供WebSocket连接点；2) 管理与OpenAI Realtime API的会话；3) 处理工具调用。

步骤1：创建OpenAI会话管理器我们创建一个realtime_session.py文件，封装与OpenAI的交互：

import asyncio import json from typing import Dict, Any, Optional import openai from openai import AsyncOpenAI import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RealtimeSession: def __init__(self, api_key: str, tools: list): self.client = AsyncOpenAI(api_key=api_key) self.session = None self.tools = tools # 用于存储工具执行函数映射 self.tool_functions = { “get_weather”: self._execute_get_weather, “calculate”: self._execute_calculate, } async def create_session(self): “”“创建并连接到一个新的Realtime API会话”“” try: # 创建会话，指定模型和工具 self.session = await self.client.beta.realtime.sessions.create( model=“gpt-4o-realtime-preview-2024-12-17”, # 使用最新的实时模型 voice=“alloy”, # 选择一种语音 tools=self.tools, modalities=[“text”, “audio”], # 启用文本和音频模态 ) logger.info(f”Session created with ID: {self.session.id}”) # 这里通常需要建立WebSocket连接，但OpenAI SDK可能封装了细节 # 实际中可能需要使用 client.beta.realtime.connect(session_id=…) return self.session except Exception as e: logger.error(f”Failed to create session: {e}”) raise async def _execute_get_weather(self, arguments: Dict[str, Any]) -> str: “”“模拟执行获取天气工具”“” location = arguments.get(“location”, “未知城市”) unit = arguments.get(“unit”, “celsius”) # 这里应调用真实的天气API，此处模拟 logger.info(f”Executing get_weather for {location} in {unit}”) return json.dumps({“weather”: “sunny”, “temperature”: 25, “unit”: unit, “location”: location}) async def _execute_calculate(self, arguments: Dict[str, Any]) -> str: “”“执行简单计算工具”“” expression = arguments.get(“expression”) try: # 警告：在生产环境中直接eval是危险的，此处仅作演示 result = eval(expression, {“__builtins__”: {}}, {}) return json.dumps({“result”: result}) except Exception as e: return json.dumps({“error”: f”计算失败: {e}”}) async def process_tool_call(self, tool_call_id: str, function_name: str, arguments: str): “”“处理来自模型的工具调用请求”“” if function_name not in self.tool_functions: return json.dumps({“error”: f”未知工具: {function_name}”}) try: args_dict = json.loads(arguments) func = self.tool_functions[function_name] result = await func(args_dict) # 将工具执行结果发送回会话 # 注意：此处需要根据实际的Realtime API事件发送机制来操作 # 概念上类似于：session.send({“type”: “function_call_output”, “output”: result, …}) logger.info(f”Tool {function_name} executed, result: {result}”) return result except json.JSONDecodeError: return json.dumps({“error”: “工具参数格式无效”}) except Exception as e: logger.error(f”Tool execution error: {e}”) return json.dumps({“error”: str(e)})

步骤2：创建FastAPI应用桥接前后端创建main.py作为应用入口：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles import asyncio import json import os from dotenv import load_dotenv from realtime_session import RealtimeSession load_dotenv() app = FastAPI() # 定义工具列表（与session中的映射对应） TOOLS = [ { “type”: “function”, “function”: { “name”: “get_weather”, “description”: “获取指定城市的当前天气和温度。用户询问天气时使用。”, “parameters”: { “type”: “object”, “properties”: { “location”: {“type”: “string”, “description”: “城市或地区名称，例如‘北京’、‘纽约’。”}, “unit”: {“type”: “string”, “enum”: [“celsius”, “fahrenheit”], “description”: “温度单位，摄氏度或华氏度。”} }, “required”: [“location”] } } }, { “type”: “function”, “function”: { “name”: “calculate”, “description”: “执行一个简单的数学计算。例如‘计算3加5乘2’。”, “parameters”: { “type”: “object”, “properties”: { “expression”: {“type”: “string”, “description”: “数学表达式，例如‘3+5*2’、‘sqrt(16)’。”} }, “required”: [“expression”] } } } ] class ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] = [] # 为每个连接维护一个会话实例（简单示例，生产环境需更复杂管理） self.sessions: dict = {} async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) # 为每个新连接创建独立的Realtime会话 session = RealtimeSession(api_key=os.getenv(“OPENAI_API_KEY”), tools=TOOLS) await session.create_session() self.sessions[websocket] = session logger.info(f”New client connected. Total: {len(self.active_connections)}”) def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) self.sessions.pop(websocket, None) logger.info(f”Client disconnected. Total: {len(self.active_connections)}”) async def receive_audio(self, websocket: WebSocket, audio_data: bytes): “”“接收来自前端的音频数据，并转发给OpenAI会话”“” session = self.sessions.get(websocket) if session: # 此处需要将音频数据包装成Realtime API要求的格式并发送 # 例如：await session.send_audio(audio_data) # 由于OpenAI SDK的Realtime接口可能仍在演进，此处为概念性代码 pass async def handle_realtime_event(self, websocket: WebSocket, event: dict): “”“处理从OpenAI会话传回的事件（如文本回复、工具调用）”“” event_type = event.get(“type”) if event_type == “response.text.delta”: # 将模型返回的文本流推送给前端 text_delta = event.get(“delta”, “”) await websocket.send_json({“type”: “text_delta”, “data”: text_delta}) elif event_type == “response.function_call_arguments.delta”: # 累积工具调用参数 pass elif event_type == “response.function_call_arguments.done”: # 工具参数接收完毕，执行工具 tool_call = event.get(“function_call”) if tool_call: session = self.sessions.get(websocket) result = await session.process_tool_call( tool_call_id=tool_call.get(“id”), function_name=tool_call.get(“name”), arguments=tool_call.get(“arguments”) ) # 将工具执行结果发送回OpenAI会话 # await session.submit_tool_output(…) manager = ConnectionManager() @app.websocket(“/ws”) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: # 接收前端消息 data = await websocket.receive_json() msg_type = data.get(“type”) if msg_type == “audio_data”: await manager.receive_audio(websocket, data.get(“data”)) elif msg_type == “text_input”: # 处理文本输入 pass except WebSocketDisconnect: manager.disconnect(websocket) @app.get(“/”) async def get(): with open(“static/index.html”, “r”) as f: html_content = f.read() return HTMLResponse(content=html_content) # 挂载静态文件目录 app.mount(“/static”, StaticFiles(directory=“static”), name=“static”)

4.3 前端交互界面实现

在static目录下创建index.html和app.js。前端主要负责音频采集、播放，以及与后端WebSocket通信。

index.html核心部分:

<!DOCTYPE html> <html> <head> <title>实时AI智能体演示</title> </head> <body> <h1>实时语音智能体</h1> <button id=”startBtn”>开始对话</button> <button id=”stopBtn” disabled>停止</button> <div id=”status”>状态：未连接</div> <div id=”transcript”></div> <div id=”agentResponse”></div> <script src=”app.js”></script> </body> </html>

app.js核心逻辑:

let mediaRecorder; let audioChunks = []; let socket; const startBtn = document.getElementById(‘startBtn’); const stopBtn = document.getElementById(‘stopBtn’); const statusDiv = document.getElementById(‘status’); const transcriptDiv = document.getElementById(‘transcript’); const responseDiv = document.getElementById(‘agentResponse’); startBtn.onclick = async () => { statusDiv.textContent = ‘正在获取麦克风权限…’; try { const stream = await navigator.mediaDevices.getUserMedia({ audio: true }); // 配置音频格式，例如16kHz单声道，适合语音识别 const options = { mimeType: ‘audio/webm; codecs=opus’ }; mediaRecorder = new MediaRecorder(stream, options); mediaRecorder.ondataavailable = (event) => { if (event.data.size > 0) { audioChunks.push(event.data); // 可以将数据转换为ArrayBuffer并发送给后端 // 为了简化，这里我们仅在停止时发送。实际应实时发送小片段。 } }; mediaRecorder.onstop = async () => { const audioBlob = new Blob(audioChunks, { type: ‘audio/webm’ }); audioChunks = []; const arrayBuffer = await audioBlob.arrayBuffer(); // 通过WebSocket发送音频数据 if (socket && socket.readyState === WebSocket.OPEN) { socket.send(JSON.stringify({ type: ‘audio_data’, data: Array.from(new Uint8Array(arrayBuffer)) // 转换为数组传输 })); } }; // 连接WebSocket const wsProtocol = window.location.protocol === ‘https:’ ? ‘wss:’ : ‘ws:’; const wsUrl = `${wsProtocol}//${window.location.host}/ws`; socket = new WebSocket(wsUrl); socket.onopen = () => { statusDiv.textContent = ‘已连接，可以开始说话’; startBtn.disabled = true; stopBtn.disabled = false; mediaRecorder.start(1000); // 每1秒生成一个数据块 }; socket.onmessage = (event) => { const msg = JSON.parse(event.data); if (msg.type === ‘text_delta’) { // 流式显示智能体的回复 responseDiv.innerHTML += msg.data; } else if (msg.type === ‘transcript’) { // 显示用户语音的实时转写 transcriptDiv.textContent = msg.data; } }; socket.onclose = () => { statusDiv.textContent = ‘连接已断开’; startBtn.disabled = false; stopBtn.disabled = true; if (mediaRecorder.state === ‘recording’) { mediaRecorder.stop(); } }; } catch (err) { console.error(‘无法访问麦克风:’, err); statusDiv.textContent = ‘麦克风访问失败’; } }; stopBtn.onclick = () => { if (mediaRecorder && mediaRecorder.state === ‘recording’) { mediaRecorder.stop(); statusDiv.textContent = ‘已停止录音’; stopBtn.disabled = true; startBtn.disabled = false; } if (socket) { socket.close(); } };

4.4 运行与测试

1. 将前端文件放入static文件夹。
2. 在项目根目录运行后端服务：

   uvicorn main:app --reload --host 0.0.0.0 --port 8000

3. 打开浏览器，访问http://localhost:8000。
4. 点击“开始对话”，授权麦克风，然后说话。你的语音会被录制、发送到后端、转发给OpenAI Realtime API，模型的文本回复会流式显示在页面上，同时（如果配置了TTS）还会播放语音回复。你可以尝试问：“上海今天的天气怎么样？” 智能体会调用get_weather工具并给出回答。

5. 生产环境部署与优化指南

5.1 架构扩展与高可用考虑

上述示例是单进程、内存存储会话的简单实现，不适合生产。生产环境需要考虑：

• 会话状态持久化：将会话信息（连接状态、对话历史、工具调用上下文）存储到Redis或数据库中，以支持服务重启和水平扩展。
• 连接管理与心跳：实现WebSocket连接的心跳检测和自动重连机制，处理网络不稳定的情况。
• 水平扩展：使用消息队列（如RabbitMQ、Kafka）解耦音频接收、AI推理、工具执行等环节。可以部署多个无状态的工作节点来处理OpenAI API调用和工具执行，通过负载均衡器分发WebSocket连接。
• 安全性：对API密钥进行严格管理（使用密钥管理服务），对用户输入进行必要的清洗和过滤，防止注入攻击。WebSocket连接应使用WSS（加密），并考虑添加身份验证（如JWT Token）。

5.2 性能优化关键点

1. 音频处理优化：

   • 前端VAD：在浏览器端使用WebAssembly版本的VAD库（如silero-vad），只在检测到人声时才发送音频，节省带宽和后台处理资源。
   • 音频压缩：在发送前对音频进行适当的编码压缩（如Opus），但需平衡压缩率和延迟。
   • 分块大小：调整音频提交的分块大小。太小会增加请求开销，太大会增加延迟。通常100-300ms是一个平衡点。

2. 模型与提示词优化：

   • 模型选择：根据场景在效果、速度和成本间权衡。gpt-4o-realtime效果最好但成本高，gpt-4o-mini-realtime可能更经济。
   • 系统提示词：精心设计系统提示词（System Prompt），明确智能体的角色、能力边界和回复风格。这对于控制工具调用频率和回复质量至关重要。例如，可以加入“除非用户明确要求，否则不要频繁使用工具”、“用简洁的口语化中文回复”等指令。
   • 上下文管理：实时对话可能很长，需要策略性地修剪或总结历史上下文，以保持在模型的令牌限制内，同时保留重要信息。

3. 工具系统优化：

   • 工具缓存：对于耗时或调用频繁的工具（如查询天气），可以引入缓存机制，避免重复调用。
   • 异步执行：工具调用（尤其是涉及网络IO的）应全部异步化，避免阻塞事件循环。
   • 超时与重试：为工具调用设置合理的超时时间和重试策略。

5.3 监控、日志与调试

• 全链路日志：记录关键事件的日志，包括WebSocket连接/断开、音频接收/发送、工具调用请求/结果、模型回复内容（注意脱敏）。使用结构化日志（JSON格式）便于检索和分析。
• 关键指标监控：
  • 端到端延迟：从用户停止说话到听到回复首字的延迟。
  • 工具调用成功率与耗时。
  • API调用错误率与令牌消耗。
  • 并发会话数。
• 调试工具：开发一个管理界面，可以查看活跃会话、实时对话流、工具调用历史，甚至能手动注入事件进行调试。

6. 常见问题排查与实战技巧

在实际开发和运维中，你肯定会遇到各种问题。下面是一些典型问题及其排查思路：

实战技巧分享：

1. “思考过程”可视化：在开发调试阶段，可以让模型输出其“链式思考”（Chain-of-Thought）。虽然Realtime API的响应事件流中可能不直接包含思考过程，但你可以在系统提示词中要求模型在最终回复前，先输出一段以特定标记（如【思考】）开头的内容。前端可以解析并显示这部分内容，帮助你理解模型的决策逻辑。
2. 优雅处理打断：在实时对话中，用户可能会打断智能体的发言。你需要实现打断检测。一种方法是，当检测到用户开始说话（input_audio_buffer.speech_started）时，立即向会话发送一个response.cancel事件，取消模型正在进行的回复生成。
3. 上下文切换与长期记忆：对于多轮对话，简单的上下文窗口可能不够。可以考虑引入向量数据库（如Pinecone、Chroma），将每轮对话的核心信息向量化存储。当新对话开始时，先进行语义检索，将与当前话题最相关的历史信息作为“记忆”注入上下文，实现长期、连贯的对话。
4. 成本控制：实时API和GPT-4o的调用成本不菲。务必实施用量监控和限流。可以为每个用户设置对话时长或令牌数上限。对于工具调用，也要做好限制，防止恶意用户诱导智能体无限循环调用高成本的外部API。

构建一个稳定、高效、智能的实时AI智能体是一个持续迭代的过程。openai-realtime-agents项目提供了坚实的地基和清晰的设计蓝图。理解其架构，掌握事件流、工具调用和多模态处理这些核心模块，再结合具体的业务场景进行深度定制和优化，你就能打造出真正具有实用价值和惊艳体验的下一代AI应用。