ChatGPT算命模板：从零构建高可用AI对话系统的技术实践

1次阅读

没有评论

共计 2160 个字符，预计需要花费 6 分钟才能阅读完成。

在开发 AI 对话系统时，尤其是算命这类需要多轮交互的场景，开发者常遇到以下问题：

意图识别不准 ：用户输入 ” 我今年运势如何 ” 和 ” 看看我的财运 ” 本质是同类意图，但传统规则引擎需要分别处理
上下文丢失 ：当用户追问 ” 那健康方面呢？” 时，系统无法关联前文提到的用户星座 / 生日信息
响应超时 ：直接调用 API 可能导致移动端用户在弱网环境下等待超时

我们的解决方案采用三层架构：

接入层 ：处理 HTTP 请求、参数校验和限流
对话引擎层 ：核心包含状态机管理、意图分类和上下文缓存
数据持久层 ：使用 Redis 存储对话状态，支持集群部署

关键设计决策：

每个对话 session 维护独立的状态机
用户输入先经过意图分类器预处理
最近 3 轮对话自动缓存为 prompt 上下文

from typing import Dict, Optional
import redis
from pydantic import BaseModel

class DialogueState(BaseModel):
    session_id: str
    current_step: str = 'init'
    user_attributes: Dict[str, str] = {}
    context_messages: list[str] = []

class StateManager:
    def __init__(self, redis_conn: redis.Redis):
        self.redis = redis_conn

    async def get_state(self, session_id: str) -> Optional[DialogueState]:
        """从 Redis 获取对话状态"""
        try:
            state_data = await self.redis.get(f'dialogue:{session_id}')
            return DialogueState.parse_raw(state_data) if state_data else None
        except Exception as e:
            print(f"State read error: {e}")
            return None

    async def save_state(self, state: DialogueState, ttl: int = 3600):
        """持久化状态并设置 TTL"""
        try:
            await self.redis.setex(f'dialogue:{session_id}', 
                ttl,
                state.json())
        except Exception as e:
            print(f"State save error: {e}")

import re
from typing import List

class ContentFilter:
    def __init__(self, blocklist_path: str):
        with open(blocklist_path) as f:
            self.block_patterns = [re.compile(line.strip(), re.IGNORECASE) 
                for line in f if line.strip()]

    def sanitize(self, text: str) -> str:
        """替换敏感词为 * 号"""
        for pattern in self.block_patterns:
            text = pattern.sub('*'*8, text)
        return text

使用 aiohttp 处理并发请求，避免阻塞事件循环：

import aiohttp

async def query_gpt(session: aiohttp.ClientSession, prompt: str) -> str:
    """异步调用 GPT API"""
    try:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            json={"messages": [{"role": "user", "content": prompt}]},
            timeout=aiohttp.ClientTimeout(total=10)
        ) as resp:
            if resp.status == 200:
                data = await resp.json()
                return data['choices'][0]['message']['content']
    except asyncio.TimeoutError:
        return "系统正在思考，请稍后再试"