fastapi-ai-chat-demo/├── main.py                 # 主应用文件├── config.py              # 配置管理├── start_server.py        # 启动脚本├── requirements.txt       # 依赖包列表├── .env.example          # 环境变量模板├── ai_providers/         # AI提供商模块│   ├── __init__.py│   ├── base.py          # 基础类定义│   ├── factory.py       # 提供商工厂│   ├── openai_provider.py│   ├── deepseek_provider.py│   ├── doubao_provider.py│   ├── kimi_provider.py│   ├── qianwen_provider.py│   └── openai_compatible_provider.py├── static/              # 静态文件│   ├── index.html      # 聊天界面│   ├── css/│   └── js/├── docs/               # 项目文档└── logs/              # 日志文件目录

@app.post("/chat/start")async def start_chat(user_id: str):    session_id = generate_session_id()    return {"session_id": session_id, "welcome_message": "你好！我是你的AI助手"}

@app.get("/chat/stream")async def chat_stream(user_id: str, session_id: str, message: str, role: str = "assistant"):    return StreamingResponse(generate_streaming_response(user_id, session_id, message, role))

@app.get("/chat/history")async def get_chat_history(user_id: str, session_id: str):    history = await get_conversation_history(user_id, session_id)    return {"messages": history, "total": len(history)}

@app.delete("/chat/history/{session_id}")async def clear_conversation_history(session_id: str, user_id: str):    redis_client.delete(get_conversation_key(user_id, session_id))    return {"message": "对话历史已清除"}

@app.get("/roles")async def get_roles():    return {"roles": AI_ROLES, "default_role": "assistant"}

<divclass="chat-container">    <!-- 1. 头部：显示标题和角色选择 -->    <divclass="chat-header">        <h1>🤖 AI智能助手</h1>        <selectid="roleSelect">            <optionvalue="assistant">💬 智能助手</option>            <optionvalue="teacher">👨‍🏫 AI老师</option>            <optionvalue="programmer">👨‍💻 编程专家</option>        </select>    </div>    <!-- 2. 消息区域：显示对话内容 -->    <divclass="messages-container"id="messagesContainer">        <!-- 消息会动态添加到这里 -->    </div>    <!-- 3. 输入区域：用户输入消息 -->    <divclass="input-container">        <inputtype="text"id="messageInput"placeholder="输入你的消息...">        <buttononclick="sendMessage()">📤 发送</button>    </div>    <!-- 4. 工具栏：常用功能按钮 -->    <divclass="toolbar">        <buttononclick="clearHistory()">🗑️ 清除历史</button>        <buttononclick="newChat()">🆕 新对话</button>    </div></div>

async function startNewChat() {    // 调用后端API创建新会话    const response = await fetch('/api/chat/start', { method: 'POST' });    const data = await response.json();    currentSessionId = data.session_id;    // 显示欢迎消息    addMessage('assistant', '你好！我是你的AI助手，有什么可以帮助你的吗？');}

async function sendMessage() {    const message = document.getElementById('messageInput').value;    // 显示用户消息    addMessage('user', message);    // 使用EventSource接收流式响应    const eventSource = new EventSource(`/api/chat/stream?session_id=${currentSessionId}&message=${message}`);    eventSource.onmessage = function(event) {        const data = JSON.parse(event.data);        if (data.content) {            // 实时显示AI回复            updateAIMessage(data.content);        }    };}

function addMessage(role, content) {    const messageDiv = document.createElement('div');    messageDiv.className = `message ${role}`;    // 用户消息显示在右边，AI消息显示在左边    const icon = role === 'user' ? '👤' : '🤖';    messageDiv.innerHTML = `${icon}${content}`;    document.getElementById('messagesContainer').appendChild(messageDiv);    // 自动滚动到最新消息    messageDiv.scrollIntoView({ behavior: 'smooth' });}

async function clearHistory() {    if (confirm('确定要清除所有对话历史吗？')) {        await fetch(`/api/chat/history/${currentSessionId}`, { method: 'DELETE' });        document.getElementById('messagesContainer').innerHTML = '';        addMessage('system', '对话历史已清除');    }}

class ChatMessage(BaseModel):    role: str        # 谁说的话："user"(用户) 或 "assistant"(AI)    content: str     # 说了什么：具体的对话内容    timestamp: float # 什么时候说的：消息时间戳

def generate_session_id() -> str:    return str(uuid.uuid4())

def save_message(user_id: str, session_id: str, message: ChatMessage):    conversation_key = get_conversation_key(user_id, session_id)    redis_client.lpush(conversation_key, json.dumps(message_data))    redis_client.ltrim(conversation_key, 0, 19)  # 只保留最近20条消息

def get_conversation_history(user_id: str, session_id: str):    conversation_key = get_conversation_key(user_id, session_id)    messages = redis_client.lrange(conversation_key, 0, -1)    return [json.loads(msg) for msg in messages]

user_msg = ChatMessage(role="user", content=user_message)save_message(session_id, user_msg)

history = get_conversation_history(session_id, limit=10)

messages = [    {"role": "system", "content": AI_ROLES[role]},  # AI角色设定    *history,  # 历史对话    {"role": "user", "content": user_message}  # 当前问题]

response = client.chat.completions.create(    model="gpt-4o",    messages=messages,    stream=True  # 关键：启用流式响应)

for chunk in response:    if chunk.choices[0].delta.content:        content = chunk.choices[0].delta.content        yield f"data: {json.dumps({'content': content})}\n\n"

1、打开页面 GET /static/index.html（main.py:398 根路径重定向）2、getUserId()从 localStorage取或生成 user_id（demo_user_xxx，index.html:1845）3、启动时 loadProviders()调 GET /providers（index.html:1864），拿可选提供商/模型填充下拉框4、会话：前端本地 currentSessionId（无后端建会话接口，前端自行维护/复用）

const message = messageInput.value.trim();   // 读取并去空格                                       ...                                                                                           messageInput.value = '';      // 发送后清空输入框                              

1、sendMessage() 构建请求体（index.html:2390）：{ user_id, session_id, message, provider, model, image_data, image_type }2、fetch POST /chat/stream，Content-Type: application/json（index.html:2401）

1、读取 ChatRequest，role被硬编码为 "assistant"（request.role被忽略）2、校验角色、直接返回 StreamingResponse(media_type="text/event-stream")（SSE）3、实际工作在生成器 generate_streaming_response中执行

1、存用户消息：把用户消息封装 AIMessage→ save_message_to_redis（main.py:177）  - Redis 可用：LPUSH conversation:{user_id}:{session_id}，设 7 天过期，并更新 user_sessionshash   - Redis 不可用：写入进程内 MEMORY_STORAGE2、取历史：get_conversation_history（main.py:234）LRANGE后 reverse（Redis 倒序存）3、裁剪：只取最近 MAX_HISTORY_MESSAGES（默认20）条4、构造消息：每轮转换回 AIMessage（带 image_data/image_type）5、调模型：ai_manager.generate_streaming_response(messages, provider, model, system_prompt)

1、选定提供商：前端传的 provider（存在则用），否则默认提供商2、model非空时放进 kwargs 透传3、调对应实例的 generate_streaming_response

1、format_messages（:165）：系统提示放最前；带图的 user 消息转成多模态数组：[{type:"image_url", image_url:{url:"data:image/jpeg;base64,..."}}, {type:"text", text:"..."}]2、_build_request_params（:216）：model（优先 kwargs → 配置）、max_tokens、temperature、extra_body={"enable_thinking": True}、stream=True3、client.chat.completions.create(**params)返回流迭代器4、逐 chunk 输出（:147）：   - 有 delta.reasoning_content → data: {"type":"reasoning","content":...}   - 有 delta.content → data: {"type":"content","content":...}

1、每个 chunk：累计 full_response；解析 JSON，只把 type=='content' 累进 content_only_response（深度思考不进历史）2、yield chunk原样透传前端3、流结束后：content_only_response封装 assistant ChatMessage → save_message_to_redis入库4、最后 yield data: {"type":"end","session_id":...}（main.py:387）

1、reader.read()逐块解码，按 data: 前缀解析 JSON2、type==='reasoning'→ 渲染进思考区（.reasoning-content，折叠展示，实时 markdown）3、type==='content' → 渲染正文消息（renderMarkdownContent）4、收到 type==='end' → 恢复输入框、更新会话列表

1、后端生成器内：抛错则 yield data: {"type":"error",...}（main.py:392）2、provider 层：SDK 异常被 provider 内 catch，yield 的是裸错误字符串（无 data: 前缀，openai_compatible_provider.py:163），前端按 data:过滤会直接忽略——错误对用户不可见（配合日志里 403 过期 key 的现象，前端表现为"没回复"）

index.html sendMessage  → POST /chat/stream (main.py:408)    → generate_streaming_response (main.py:307)      → save_message_to_redis (用户消息)      [main.py:177]      → get_conversation_history               [main.py:234]      → MultiProviderManager.generate_streaming_response (factory.py:332)        → OpenAICompatibleProvider.generate_streaming_response (openai_compatible_provider.py:117)          → OpenAI SDK chat.completions.create(stream=True)      → 逐 chunk 解析 + 透传 SSE      → save_message_to_redis (assistant 回复，仅 content)   [main.py:384]  → 前端 processStream 按类型渲染

pip install -r requirements.txt

cp .env.example .env

# OpenAI配置OPENAI_API_KEY=your_openai_api_keyOPENAI_BASE_URL=https://api.openai.com/v1# DeepSeek配置DEEPSEEK_API_KEY=your_deepseek_api_key# 豆包配置DOUBAO_API_KEY=your_doubao_api_key# Kimi配置KIMI_API_KEY=your_kimi_api_key# 通义千问配置QIANWEN_API_KEY=your_qianwen_api_key# Redis配置（可选）REDIS_HOST=localhostREDIS_PORT=6379REDIS_PASSWORD=REDIS_DB=0# 应用配置DEFAULT_AI_PROVIDER=openaiDEBUG=trueLOG_LEVEL=INFO

# 方式1：直接启动python main.py# 方式2：使用启动脚本python start_server.py# 方式3：使用uvicornuvicorn main:app --host 0.0.0.0 --port 8000 --reload