MiniMax模型与Raven框架集成实战：构建智能对话系统

最近在开发智能对话应用时很多开发者都面临一个共同挑战如何将强大的大语言模型与实际业务场景深度结合而不仅仅是简单的问答交互。MiniMax 作为国内领先的AI公司其模型在中文理解和生成方面表现出色而 Raven 智能体框架则为构建复杂AI应用提供了系统化解决方案。本文将完整演示如何将 MiniMax 模型集成到 Raven 框架中打造真正可用的智能对话系统。1. MiniMax 与 Raven 框架核心概念解析1.1 MiniMax 模型能力与特点MiniMax 是一家专注于通用人工智能研发的科技公司其大语言模型在中文自然语言处理方面具有显著优势。与国外模型相比MiniMax 对中文语境、文化背景和表达习惯有更好的理解这在中文对话场景中尤为重要。MiniMax 模型提供多种规格从轻量级到超大参数模型满足不同场景的性能需求。其 API 接口设计遵循 RESTful 规范支持流式响应、多轮对话记忆、角色设定等高级功能为构建复杂对话系统提供了坚实基础。1.2 Raven 智能体框架架构设计Raven 是一个专为构建AI智能体应用而设计的开源框架它采用模块化架构将复杂的AI应用拆分为可组合的组件。框架核心包含以下层次对话管理层负责维护对话状态、上下文记忆和会话流程控制技能插件层提供可扩展的技能模块如知识检索、工具调用、外部API集成模型抽象层统一不同AI模型的调用接口实现模型的无缝切换业务逻辑层将AI能力与具体业务需求相结合实现端到端的解决方案Raven 的设计理念是配置即代码通过声明式配置定义智能体行为大大降低了AI应用的开发门槛。1.3 集成方案的价值与适用场景将 MiniMax 模型集成到 Raven 框架中可以发挥两者的协同效应。MiniMax 提供强大的语言理解和生成能力Raven 则负责复杂的对话逻辑和业务集成。这种组合特别适合以下场景智能客服系统需要理解用户复杂诉求并给出准确回复教育辅导应用要求模型具备知识推理和多轮对话能力企业知识助手结合内部知识库提供专业咨询服务创意写作工具利用模型的创造性生成各种类型的内容2. 环境准备与依赖配置2.1 系统环境要求在开始集成之前需要确保开发环境满足以下要求操作系统Windows 10/11, macOS 10.15, 或 Ubuntu 18.04 等主流操作系统Python 版本3.8 或更高版本推荐 3.9内存要求至少 8GB RAM复杂应用建议 16GB网络连接稳定的互联网连接用于调用 MiniMax API可以通过以下命令检查 Python 环境python --version pip --version2.2 创建项目结构与虚拟环境建议为每个项目创建独立的虚拟环境避免依赖冲突# 创建项目目录 mkdir minimax-raven-integration cd minimax-raven-integration # 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Windows venv\Scripts\activate # Linux/macOS source venv/bin/activate项目目录结构设计如下minimax-raven-integration/ ├── src/ │ ├── agents/ # 智能体定义 │ ├── skills/ # 技能模块 │ ├── models/ # 模型接口封装 │ ├── config/ # 配置文件 │ └── utils/ # 工具函数 ├── tests/ # 测试代码 ├── requirements.txt # 依赖列表 └── README.md # 项目说明2.3 安装核心依赖包创建requirements.txt文件包含以下依赖raven-framework0.5.0 requests2.28.0 pydantic1.10.0 python-dotenv0.19.0 httpx0.23.0 asyncio3.4.3 loguru0.7.0安装依赖pip install -r requirements.txt如果 Raven 框架尚未发布到 PyPI可以从 GitHub 仓库直接安装pip install githttps://github.com/raven-framework/raven.git3. MiniMax API 配置与认证3.1 获取 MiniMax API 密钥要使用 MiniMax 模型首先需要注册账号并获取 API 密钥访问 MiniMax 官方网站完成开发者注册进入控制台创建新的应用项目在项目设置中生成 API Key记录密钥并妥善保管避免泄露3.2 配置环境变量为了保护敏感信息建议使用环境变量管理配置。创建.env文件# 创建环境变量文件 touch .env在.env文件中配置 MiniMax API 信息MINIMAX_API_KEYyour_actual_api_key_here MINIMAX_API_BASEhttps://api.minimax.chat MINIMAX_GROUP_IDyour_group_id MINIMAX_MODELabab5.5-chat在代码中读取环境变量# src/config/settings.py import os from dotenv import load_dotenv load_dotenv() class MiniMaxConfig: API_KEY os.getenv(MINIMAX_API_KEY) API_BASE os.getenv(MINIMAX_API_BASE, https://api.minimax.chat) GROUP_ID os.getenv(MINIMAX_GROUP_ID) MODEL os.getenv(MINIMAX_MODEL, abab5.5-chat) classmethod def validate(cls): 验证配置完整性 if not cls.API_KEY: raise ValueError(MINIMAX_API_KEY 未设置) if not cls.GROUP_ID: raise ValueError(MINIMAX_GROUP_ID 未设置)3.3 实现 API 客户端封装创建 MiniMax API 的封装类统一处理请求和响应# src/models/minimax_client.py import httpx import json from typing import Dict, List, Optional from src.config.settings import MiniMaxConfig class MiniMaxClient: def __init__(self): self.api_key MiniMaxConfig.API_KEY self.base_url MiniMaxConfig.API_BASE self.group_id MiniMaxConfig.GROUP_ID self.model MiniMaxConfig.MODEL self.client httpx.AsyncClient(timeout30.0) async def chat_completion(self, messages: List[Dict], temperature: float 0.7, max_tokens: int 2048) - Dict: 调用 MiniMax 聊天补全接口 url f{self.base_url}/v1/text/chatcompletion headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: self.model, messages: messages, temperature: temperature, max_tokens: max_tokens, group_id: self.group_id } try: response await self.client.post(url, jsonpayload, headersheaders) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: raise Exception(fAPI 请求失败: {e.response.status_code} - {e.response.text}) except Exception as e: raise Exception(f请求异常: {str(e)}) async def close(self): 关闭 HTTP 客户端 await self.client.aclose()4. Raven 框架基础配置4.1 初始化 Raven 应用Raven 框架采用应用实例管理模式首先需要创建应用实例# src/app.py from raven import RavenApplication from raven.agents import AgentManager from raven.memory import ConversationMemory def create_app(): 创建 Raven 应用实例 app RavenApplication( nameminimax-chat-agent, version1.0.0, description基于 MiniMax 的智能对话助手 ) # 配置对话记忆 memory ConversationMemory( max_history_length10, # 保留最近10轮对话 memory_typevolatile # 内存类型可选持久化存储 ) # 初始化智能体管理器 agent_manager AgentManager(memorymemory) app.agent_manager agent_manager return app # 创建全局应用实例 app create_app()4.2 定义基础智能体类创建自定义智能体类集成 MiniMax 模型能力# src/agents/minimax_agent.py from raven.agents.base import BaseAgent from raven.memory import MemoryRecord from typing import Dict, List, Any from src.models.minimax_client import MiniMaxClient class MiniMaxAgent(BaseAgent): def __init__(self, name: str minimax_assistant, description: str 基于 MiniMax 的智能助手, **kwargs): super().__init__(namename, descriptiondescription, **kwargs) self.client MiniMaxClient() self.temperature kwargs.get(temperature, 0.7) self.max_tokens kwargs.get(max_tokens, 2048) async def process_message(self, message: str, context: Dict[str, Any] None) - Dict[str, Any]: 处理用户消息的核心方法 # 构建对话历史 messages await self._build_messages(message, context) # 调用 MiniMax API response await self.client.chat_completion( messagesmessages, temperatureself.temperature, max_tokensself.max_tokens ) # 解析响应 result self._parse_response(response) # 更新对话记忆 await self._update_memory(message, result[response]) return result async def _build_messages(self, message: str, context: Dict) - List[Dict]: 构建符合 MiniMax API 要求的消息格式 messages [] # 添加系统提示如果有 if context and system_prompt in context: messages.append({ role: system, content: context[system_prompt] }) # 添加对话历史 history await self.memory.get_recent_history(limit5) for record in history: messages.append({ role: user if record.role user else assistant, content: record.content }) # 添加当前消息 messages.append({ role: user, content: message }) return messages def _parse_response(self, response: Dict) - Dict[str, Any]: 解析 MiniMax API 响应 try: choices response.get(choices, []) if not choices: raise ValueError(API 响应中未找到有效回复) first_choice choices[0] return { response: first_choice.get(message, {}).get(content, ), usage: response.get(usage, {}), finish_reason: first_choice.get(finish_reason, ), raw_response: response } except Exception as e: return { response: f处理响应时出现错误: {str(e)}, usage: {}, finish_reason: error, raw_response: response } async def _update_memory(self, user_message: str, assistant_response: str): 更新对话记忆 user_record MemoryRecord(roleuser, contentuser_message) assistant_record MemoryRecord(roleassistant, contentassistant_response) await self.memory.add_record(user_record) await self.memory.add_record(assistant_record)5. 完整集成实战示例5.1 创建对话管理服务实现一个完整的对话服务管理智能体生命周期和会话状态# src/services/chat_service.py from typing import Dict, Any, Optional from src.agents.minimax_agent import MiniMaxAgent from src.app import app class ChatService: def __init__(self): self.agents {} # 用户会话到智能体的映射 self.default_agent_config { temperature: 0.7, max_tokens: 2048, system_prompt: 你是一个有帮助的AI助手请用中文友好地回答用户问题。 } async def get_agent_for_session(self, session_id: str) - MiniMaxAgent: 获取或创建会话对应的智能体 if session_id not in self.agents: agent MiniMaxAgent( namefagent_{session_id}, **self.default_agent_config ) # 注册到 Raven 应用 await app.agent_manager.register_agent(agent) self.agents[session_id] agent return self.agents[session_id] async def process_chat(self, session_id: str, message: str, context: Optional[Dict[str, Any]] None) - Dict[str, Any]: 处理聊天消息 try: agent await self.get_agent_for_session(session_id) # 合并上下文信息 chat_context self.default_agent_config.copy() if context: chat_context.update(context) result await agent.process_message(message, chat_context) return { success: True, response: result[response], usage: result[usage], session_id: session_id } except Exception as e: return { success: False, error: str(e), session_id: session_id } async def clear_session(self, session_id: str): 清空会话历史 if session_id in self.agents: agent self.agents[session_id] await agent.memory.clear() del self.agents[session_id]5.2 实现 Web API 接口创建 RESTful API 接口提供 HTTP 服务# src/api/chat_api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from src.services.chat_service import ChatService import uuid app FastAPI(titleMiniMax Raven Chat API) chat_service ChatService() class ChatRequest(BaseModel): message: str session_id: Optional[str] None temperature: Optional[float] 0.7 system_prompt: Optional[str] None class ChatResponse(BaseModel): success: bool response: Optional[str] None session_id: str error: Optional[str] None usage: Optional[dict] None app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): 聊天接口 try: # 生成或使用现有会话ID session_id request.session_id or str(uuid.uuid4()) # 构建上下文 context {} if request.system_prompt: context[system_prompt] request.system_prompt if request.temperature: context[temperature] request.temperature # 处理消息 result await chat_service.process_chat( session_idsession_id, messagerequest.message, contextcontext ) return ChatResponse(**result) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.delete(/sessions/{session_id}) async def clear_session(session_id: str): 清空会话接口 try: await chat_service.clear_session(session_id) return {success: True, message: 会话已清空} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health_check(): 健康检查接口 return {status: healthy, service: minimax-raven-chat}5.3 创建启动脚本编写应用启动脚本整合所有组件# main.py import uvicorn from src.config.settings import MiniMaxConfig from src.api.chat_api import app def main(): 主启动函数 # 验证配置 try: MiniMaxConfig.validate() print(✓ 配置验证通过) except ValueError as e: print(f✗ 配置错误: {e}) return # 启动服务 print( 启动 MiniMax Raven 聊天服务...) uvicorn.run( app, host0.0.0.0, port8000, log_levelinfo ) if __name__ __main__: main()6. 高级功能与定制化开发6.1 实现流式响应支持对于需要实时响应的场景实现流式输出功能# src/models/streaming_client.py import json import httpx from src.models.minimax_client import MiniMaxClient class StreamingMiniMaxClient(MiniMaxClient): async def stream_chat_completion(self, messages: List[Dict], temperature: float 0.7, max_tokens: int 2048): 流式聊天补全接口 url f{self.base_url}/v1/text/chatcompletion/stream headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, Accept: text/event-stream } payload { model: self.model, messages: messages, temperature: temperature, max_tokens: max_tokens, group_id: self.group_id, stream: True } async with httpx.AsyncClient(timeout30.0) as client: async with client.stream(POST, url, jsonpayload, headersheaders) as response: async for line in response.aiter_lines(): if line.startswith(data: ): data line[6:] # 移除 data: 前缀 if data.strip() [DONE]: break try: event_data json.loads(data) yield event_data except json.JSONDecodeError: continue6.2 添加技能插件系统扩展 Raven 框架支持自定义技能插件# src/skills/base_skill.py from abc import ABC, abstractmethod from typing import Dict, Any class BaseSkill(ABC): 技能基类 def __init__(self, name: str, description: str): self.name name self.description description abstractmethod async def execute(self, parameters: Dict[str, Any]) - Dict[str, Any]: 执行技能 pass abstractmethod def get_schema(self) - Dict[str, Any]: 获取技能参数模式 pass # 示例天气查询技能 class WeatherSkill(BaseSkill): def __init__(self): super().__init__(weather, 查询城市天气信息) async def execute(self, parameters: Dict[str, Any]) - Dict[str, Any]: city parameters.get(city, ) # 这里可以集成真实天气API return { success: True, result: f{city}的天气是晴朗25℃, source: weather_skill } def get_schema(self) - Dict[str, Any]: return { type: object, properties: { city: { type: string, description: 城市名称 } }, required: [city] }6.3 实现智能体技能路由创建技能路由机制让智能体能够自动选择和执行合适的技能# src/agents/skilled_agent.py from src.agents.minimax_agent import MiniMaxAgent from src.skills.base_skill import BaseSkill from typing import Dict, Any, List class SkilledMiniMaxAgent(MiniMaxAgent): def __init__(self, skills: List[BaseSkill] None, **kwargs): super().__init__(**kwargs) self.skills skills or [] self.skill_registry {skill.name: skill for skill in self.skills} async def process_message(self, message: str, context: Dict[str, Any] None) - Dict[str, Any]: # 先检查是否需要使用技能 skill_result await self._try_use_skill(message) if skill_result[should_use_skill]: return skill_result # 否则使用普通对话 return await super().process_message(message, context) async def _try_use_skill(self, message: str) - Dict[str, Any]: 尝试使用技能处理消息 # 这里可以集成意图识别逻辑 # 简化示例通过关键词匹配 for skill_name, skill in self.skill_registry.items(): if skill_name in message.lower(): try: # 提取参数简化处理 parameters self._extract_parameters(message, skill) result await skill.execute(parameters) return { should_use_skill: True, response: result[result], skill_used: skill_name, raw_result: result } except Exception as e: return { should_use_skill: False, error: f技能执行失败: {str(e)} } return {should_use_skill: False}7. 测试与验证方案7.1 单元测试编写为核心组件编写单元测试确保代码质量# tests/test_minimax_client.py import pytest import asyncio from src.models.minimax_client import MiniMaxClient from src.config.settings import MiniMaxConfig pytest.fixture def minimax_client(): return MiniMaxClient() pytest.mark.asyncio async def test_chat_completion(minimax_client): 测试聊天补全功能 messages [ {role: user, content: 你好请简单介绍一下你自己} ] try: response await minimax_client.chat_completion(messages) assert choices in response assert len(response[choices]) 0 assert message in response[choices][0] except Exception as e: # 如果 API 不可用标记为跳过而不是失败 pytest.skip(fMiniMax API 不可用: {str(e)}) pytest.mark.asyncio async def test_invalid_api_key(): 测试无效 API 密钥处理 original_key MiniMaxConfig.API_KEY MiniMaxConfig.API_KEY invalid_key client MiniMaxClient() messages [{role: user, content: test}] with pytest.raises(Exception): await client.chat_completion(messages) # 恢复原始密钥 MiniMaxConfig.API_KEY original_key7.2 集成测试方案创建端到端的集成测试验证整个系统功能# tests/integration/test_chat_flow.py import pytest import asyncio from src.services.chat_service import ChatService pytest.fixture def chat_service(): return ChatService() pytest.mark.asyncio async def test_chat_flow(chat_service): 测试完整聊天流程 session_id test_session_123 # 第一轮对话 result1 await chat_service.process_chat( session_idsession_id, message你好我是测试用户 ) assert result1[success] True assert len(result1[response]) 0 # 第二轮对话测试上下文记忆 result2 await chat_service.process_chat( session_idsession_id, message我刚才说了什么 ) assert result2[success] True # 这里可以验证模型是否记得上下文 # 清理测试会话 await chat_service.clear_session(session_id)7.3 性能测试与优化编写性能测试脚本评估系统响应时间和资源消耗# tests/performance/test_performance.py import asyncio import time import statistics from src.services.chat_service import ChatService async def performance_test(): 性能测试函数 chat_service ChatService() session_id perf_test_session test_messages [ 你好, 今天天气怎么样, 请写一首短诗, 解释一下机器学习, 谢谢再见 ] latencies [] for i, message in enumerate(test_messages): start_time time.time() result await chat_service.process_chat( session_idsession_id, messagemessage ) end_time time.time() latency end_time - start_time latencies.append(latency) print(f消息 {i1}: 延迟 {latency:.2f}秒 - 成功: {result[success]}) # 统计结果 avg_latency statistics.mean(latencies) max_latency max(latencies) min_latency min(latencies) print(f\n性能统计:) print(f平均延迟: {avg_latency:.2f}秒) print(f最大延迟: {max_latency:.2f}秒) print(f最小延迟: {min_latency:.2f}秒) # 清理 await chat_service.clear_session(session_id) if __name__ __main__: asyncio.run(performance_test())8. 部署与生产环境配置8.1 Docker 容器化部署创建 Dockerfile实现应用容器化# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装 Python 依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY src/ ./src/ COPY main.py . # 创建非 root 用户 RUN useradd --create-home --shell /bin/bash app USER app # 暴露端口 EXPOSE 8000 # 启动命令 CMD [python, main.py]创建 docker-compose.yml 用于多服务部署# docker-compose.yml version: 3.8 services: minimax-chat: build: . ports: - 8000:8000 environment: - MINIMAX_API_KEY${MINIMAX_API_KEY} - MINIMAX_GROUP_ID${MINIMAX_GROUP_ID} volumes: - ./logs:/app/logs restart: unless-stopped # 可以添加 Redis 用于会话持久化 redis: image: redis:7-alpine ports: - 6379:6379 volumes: - redis_data:/data restart: unless-stopped volumes: redis_data:8.2 生产环境配置优化创建生产环境专用配置# src/config/production.py import os from src.config.settings import MiniMaxConfig class ProductionConfig: 生产环境配置 # API 配置 API_TIMEOUT 30 MAX_RETRIES 3 RETRY_DELAY 1 # 性能配置 MAX_CONCURRENT_REQUESTS 100 REQUEST_QUEUE_SIZE 1000 # 安全配置 CORS_ORIGINS os.getenv(CORS_ORIGINS, ).split(,) RATE_LIMIT_REQUESTS 100 # 每分钟请求限制 RATE_LIMIT_WINDOW 60 # 日志配置 LOG_LEVEL INFO LOG_FILE /app/logs/application.log classmethod def apply(cls): 应用生产环境配置 # 可以在这里进行生产环境特定的初始化 pass8.3 监控与日志配置配置完整的监控和日志系统# src/utils/logging.py import logging import sys from logging.handlers import RotatingFileHandler import json def setup_logging(log_levelINFO, log_fileNone): 配置日志系统 logger logging.getLogger() logger.setLevel(getattr(logging, log_level.upper())) # 日志格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) # 控制台处理器 console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件处理器如果指定了日志文件 if log_file: file_handler RotatingFileHandler( log_file, maxBytes10*1024*1024, # 10MB backupCount5 ) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger class APIMetrics: API 指标收集 def __init__(self): self.request_count 0 self.error_count 0 self.total_latency 0 def record_request(self, latency: float, success: bool): 记录请求指标 self.request_count 1 self.total_latency latency if not success: self.error_count 1 def get_metrics(self) - dict: 获取当前指标 avg_latency (self.total_latency / self.request_count if self.request_count 0 else 0) error_rate (self.error_count / self.request_count if self.request_count 0 else 0) return { total_requests: self.request_count, error_count: self.error_count, error_rate: error_rate, average_latency: avg_latency }9. 常见问题与解决方案9.1 API 调用问题排查问题现象可能原因解决方案API Error: 401 UnauthorizedAPI 密钥无效或过期检查 MINIMAX_API_KEY 配置重新生成密钥API Error: 402 Insufficient Balance账户余额不足充值 MiniMax 账户余额API Error: 400 Bad Request请求参数错误检查 messages 格式和参数合法性Connection Timeout网络连接问题检查网络连接增加超时时间Rate Limit Exceeded请求频率超限实现请求限流添加重试机制9.2 性能优化建议内存优化使用连接池管理 HTTP 连接合理设置对话历史长度避免内存泄漏定期清理闲置会话性能优化实现响应缓存减少重复 API 调用使用异步编程模式提高并发处理能力对长文本实现分块处理可靠性提升添加断路器模式防止级联失败实现优雅降级在 API 不可用时提供基础服务添加健康检查和自动恢复机制9.3 安全最佳实践API 密钥安全永远不要将 API 密钥提交到代码仓库使用环境变量或密钥管理服务定期轮换 API 密钥输入验证对所有用户输入进行验证和清理实现请求参数白名单验证防止提示词注入攻击访问控制实现基于角色的访问控制添加 API 速率限制记录完整的审计日志10. 扩展与进阶应用10.1 多模型路由策略实现智能模型路由根据场景选择最合适的模型# src/services/model_router.py from typing import Dict, Any from enum import Enum class ModelType(Enum): MINIMAX minimax # 可以扩展其他模型 # OPENAI openai # CLAUDE claude class ModelRouter: 模型路由服务 def __init__(self): self.models { ModelType.MINIMAX: { client: None, # 延迟初始化 cost_per_token: 0.001, # 示例价格 max_tokens: 4096, strengths: [中文理解, 创造性写作] } } async def select_model(self, message: str, context: Dict[str, Any]) - ModelType: 根据消息和上下文选择合适模型 # 简化策略目前只支持 MiniMax # 可以扩展为基于内容类型、复杂度、成本等因素的智能路由 return ModelType.MINIMAX async def route_request(self, message: str, context: Dict[str, Any]): 路由请求到合适模型 model_type await self.select_model(message, context) model_info self.models[model_type] # 这里可以添加模型特定的预处理逻辑 result await self._call_model(model_type, message, context) return { model_used: model_type.value, result: result, cost_estimate: self._estimate_cost(result, model_info) }10.2 实现对话状态管理增强对话状态管理支持复杂多轮对话# src/services/dialog_manager.py from typing import Dict, Any, List from enum import Enum class DialogState(Enum): INITIAL initial ACTIVE active COMPLETED completed ERROR error class DialogManager: 对话状态管理器 def __init__(self): self.dialogs {} # 会话ID到对话状态的映射 async def create_dialog(self, session_id: str, goal: str None) - str: 创建新对话 dialog_id fdialog_{session_id}_{len(self.dialogs) 1} self.dialogs[dialog_id] { session_id: session_id, goal: goal, state: DialogState.INITIAL, steps: [], created_at: datetime.now(), updated_at: datetime.now() } return dialog_id async def update_d