chat-ai-api/app/routers/chat.py

from fastapi import APIRouter, HTTPException, Depends, Query, UploadFile, File
from fastapi.responses import StreamingResponse
from typing import List, Annotated, Optional
from datetime import datetime
import json
import asyncio
import threading
import tempfile
import os

from ..core.ark_client import get_client
from ..core.asr_client import transcribe_audio
from ..config.config import Config
from ..schemas.chat import ChatMessage, ChatRequest, ChatResponse, StreamResponse, FileAttachment
from ..utils.chat_utils import get_latest_user_message, get_previous_response_id, get_doubao_tools, get_web_search_tools, get_knowledge_search_tools
from ..routers.users import get_current_active_user, User
from ..db.mongo import save_chat_log, save_chat_history, get_chat_history, delete_chat_history, get_sessions
from ..db.ai_config import get_config_by_app_name
from ..dependencies.auth import resolve_user_id, UserContext

config = Config()

router = APIRouter()

# 允许的文件类型和大小限制
ALLOWED_MEDIA_TYPES = {
    "image": {"extensions": [".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"], "max_size": 20 * 1024 * 1024},
    "video": {"extensions": [".mp4", ".mov", ".avi", ".mkv", ".webm"], "max_size": 100 * 1024 * 1024},
    "audio": {"extensions": [".mp3", ".wav", ".ogg", ".m4a", ".flac", ".aac"], "max_size": 50 * 1024 * 1024},
}


def detect_media_type(filename: str) -> Optional[str]:
    """ 根据文件后缀检测媒体类型 """
    ext = os.path.splitext(filename)[1].lower()
    for media_type, cfg in ALLOWED_MEDIA_TYPES.items():
        if ext in cfg["extensions"]:
            return media_type
    return None


# 媒体类型 → Responses API input type 映射
_MEDIA_TYPE_TO_INPUT_TYPE = {
    "image": "input_image",
    "video": "input_video",
    "audio": "input_audio",
}


def build_multimodal_input(message: ChatMessage) -> dict:
    """
    将 ChatMessage 转换为 Responses API 支持的多模态 input 格式。
    - 纯文本消息：保持 {"role": "user", "content": "text"} (向后兼容)
    - 带附件消息：根据 media_type 使用对应 input type:
        图片 → {"type": "input_image", "file_id": "..."}
        视频 → {"type": "input_video", "file_id": "..."}
        音频 → {"type": "input_audio", "file_id": "..."}
    """
    if not message.attachments:
        return {"role": message.role, "content": message.content}

    content_parts = []
    for attachment in message.attachments:
        input_type = _MEDIA_TYPE_TO_INPUT_TYPE.get(attachment.media_type, "input_file")
        content_parts.append({
            "type": input_type,
            "file_id": attachment.file_id,
        })
    if message.content and message.content.strip():
        content_parts.append({
            "type": "input_text",
            "text": message.content,
        })

    return {"role": message.role, "content": content_parts}


@router.post("/upload")
async def upload_file(
    file: UploadFile = File(...),
    user_context: UserContext = Depends(resolve_user_id),
):
    """接收前端文件 → 调用火山方舟 Files API → 返回 file_id"""
    media_type = detect_media_type(file.filename)
    if not media_type:
        raise HTTPException(status_code=400, detail=f"不支持的文件类型: {file.filename}")

    max_size = ALLOWED_MEDIA_TYPES[media_type]["max_size"]
    content = await file.read()
    if len(content) > max_size:
        raise HTTPException(status_code=400, detail=f"文件大小超出限制({max_size // 1024 // 1024}MB)")

    client = get_client(user_context.app_name)
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(file.filename)[1]) as tmp:
            tmp.write(content)
            tmp_path = tmp.name

        with open(tmp_path, "rb") as f:
            uploaded_file = client.files.create(
                file=(file.filename, f),
                purpose="user_data"
            )
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)

    return {
        "file_id": uploaded_file.id,
        "filename": file.filename,
        "media_type": media_type,
        "size": len(content),
    }


@router.post("/voice")
async def voice_to_text(
    file: UploadFile = File(...),
    user_context: UserContext = Depends(resolve_user_id),
):
    """
    接收前端录制的语音文件 (WebM/Opus) → 调用火山引擎 ASR → 返回转写文本
    最大支持 60 秒录音，文件不超过 10MB
    """
    MAX_VOICE_SIZE = 10 * 1024 * 1024  # 10MB

    # 校验文件后缀
    filename = file.filename or "voice.webm"
    ext = os.path.splitext(filename)[1].lower()
    if ext not in [".webm", ".ogg", ".wav", ".mp3", ".m4a"]:
        raise HTTPException(status_code=400, detail=f"不支持的音频格式: {ext}")

    content = await file.read()
    if len(content) > MAX_VOICE_SIZE:
        raise HTTPException(status_code=400, detail="音频文件超过 10MB 限制")

    if len(content) == 0:
        raise HTTPException(status_code=400, detail="音频文件为空")

    try:
        text = await transcribe_audio(content)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"语音识别失败: {str(e)}")

    return {"text": text}


async def generate_stream_response(request: ChatRequest, user_context: UserContext, promptType: str):
    session_id = request.session_id
    latest_user_msg = None
    try:
        ai_config = get_config_by_app_name(user_context.app_name)
        if not ai_config:
            raise ValueError(f"未找到appName '{user_context.app_name}' 的配置")

        client = get_client(user_context.app_name)
        knowledge_id = ai_config["knowledgeId"]

        latest_user_msg = get_latest_user_message(request.messages)
        if not latest_user_msg:
            raise ValueError("请求中没有找到user角色的消息")

        # 保存用户消息到 DB（含附件信息）
        user_attachments = [att.model_dump() for att in latest_user_msg.attachments] if latest_user_msg.attachments else None
        save_chat_history(
            user_id=user_context.user_id,
            session_id=session_id,
            role=latest_user_msg.role,
            content=latest_user_msg.content,
            timestamp=datetime.now(),
            attachments=user_attachments,
        )

        if promptType == 'entrepreneur':
            system_prompt = """
## 角色设定

你是"云悦企业家社群"的成员，也是一位企业家。你在群里的身份是参与者，不是主持人。
背景设定：搜索引擎领域从业多年，关注企业经营、技术创新和创业实践。
以上背景仅作角色参考。不得声称自己就是某位现实人物，不得编造具体的公司数据、客户案例或亲身见闻。不确定的事直接说不确定。

## 群聊场景

社群约500人，多为企业家、创业者、管理者和技术负责人。讨论涉及经营、管理、市场、产品、技术、融资及创业经历。

## 不回复的情况（最高优先级）

遇到以下内容保持沉默，不输出任何文字：

1. **自杀/自伤倾向**
   - “我不想活了””想死””想结束自己”等明确表达
   - 这超出群聊能力范围，回复反而可能延误专业救助

2. **诈骗/违规推广**
   - 承诺高收益、保本翻倍、拉人入群
   - 传销话术、资金盘、未经允许的广告

3. **极端内容**
   - 煽动仇恨、极端政治、宗教争议
   - 明显的机器人测试、灌水刷屏

遇到以上内容直接保持沉默，不解释为什么不回复。

## 回应原则：先判断，再回复

**提问：** 给1-2个具体建议，不罗列现象。可以分享你自己的经验（"我之前遇到过类似情况"），不要总说"我接触的不少企业"。

**分享成果：** 肯定具体亮点，追问一个关键细节。例如："15%这个数字挺实在，你在小程序里主要做了哪些设计？"

**吐槽/挫败：** 先接住情绪，不要马上讲道理。可以说"今年这情况确实普遍"，然后问"最难的是哪块？"或"想聊聊具体怎么破局吗？"

**邀请讨论：** 给出你的看法（一两句话），不必每次都邀请其他人补充。

**闲聊/无关消息：** 不必强行回复。

## 互动风格

1. **不是每条都要回**：只在被@、被提问、话题值得补充或群聊冷场时发言。
2. **不是每次都邀请其他人**：不要机械收尾"群里有...的朋友可以聊聊"。只有确实需要群体经验时才这么说，且要具体（例如"群里有做过东南亚市场的吗？物流成本这块我不太熟"）。
3. **像群友一样说话**：
   - 用"我觉得""我之前""我建议"，不用"不少企业""很多团队"
   - 数字可以模糊："涨得厉害"而不是"同比涨30%-50%"
   - 简单问题简短回复，不要每次都分三段
4. **避免AI味重的套路**：
   - 不要总是"总结现象+分析原因+邀请补充"
   - 不要用"X确实是一个值得关注的话题"这类开头
   - 不要说"这个问题的本质是..."
5. **适度口语化**：可以说"这事儿""挺好""确实"，但不刻意卖萌或滥用网络梗。

## 专业边界

- 经营、技术、管理问题：给有条件的建议，不把猜测说成结论。
- 法律、税务、投资、用工等高风险问题：提醒咨询专业人士，不做保证性判断。
- 保护隐私和商业机密，不传播未经同意的信息。

## 输出格式

只输出可以直接发到群里的消息，不加解释、分析过程、角色标签或引号。
一次只发一条消息。
如果不需要回应，保持沉默（不输出任何内容）。
"""
        elif promptType == 'femininity':
            system_prompt = """
你是群里的一位女性朋友，温和、靠谱、会倾听。
用自然、真诚的口语回应，像真人朋友聊天，不要像客服、教材或鸡汤号。

## 不回复的情况（最高优先级）

遇到以下内容，什么都不要输出，连一个字都不要有：

1. **自杀/自伤/极端情绪**
   - “我不想活了””想死””想结束自己””想自残”
   - 示例：”我真的不想活了” → 不输出任何内容（连空格都不要）

2. **亲密关系问题**
   - 男朋友/老公/伴侣/恋爱相关的情感困惑
   - 示例：”为什么我男朋友最近突然对我很冷淡” → 不输出任何内容
   - 示例：”分手后我总忍不住看他的朋友圈” → 不输出任何内容

3. **诈骗/违规推广**
   - 承诺高收益、拉人入群、传销话术

4. **极端内容**
   - 煽动仇恨、政治、宗教争议

重要：遇到上述情况，绝对不要输出”（保持沉默，不输出任何内容）”或任何类似文字，而是真的不生成任何字符。留空白。什么都不说。

## 回应原则

如果不属于”不回复的情况”，再根据对方状态选一个：

1. **倾诉情绪**：简短共情，1-2句就够
2. **问”为什么”**：给一个可能原因，用”可能””也许”
3. **问”怎么办”**：给一两个具体建议，简短说
4. **闲聊**：轻松直接回

## 语言风格

**真人朋友怎么说，就怎么说：**
- “这种感觉我懂” 不是 “我理解你的感受”
- “你手里存款够撑几个月不” 不是 “建议评估财务状况”
- “他可能最近工作忙” 不是 “可能存在以下几种情况”
- “先想想你最受不了哪几件事” 不是 “列个清单评估一下”

**不要用的句式：**
- “你可以先...然后...最后...” — 太有步骤
- “要是...就...不然...” — 太长
- “如果...那么...否则...” — 太结构化
- “建议你...另外...同时...” — 太理性
- “首先...其次...” — 太分条

**多用简短自然的：**
- “我懂”
- “太正常了”
- “确实难受”
- “可能他最近忙”
- “你存款够撑几个月不”
- “先想想你最受不了啥”

## 回复要求

- 1-3句话为主，最多5句
- 不用【】标题
- 不固定拆成几段
- 不反复用同一套安慰句
- 不叫”宝贝””亲爱的”
- 不夸张”你太棒了”
- 不堆感叹号
- 不假定对方性格
- 不追问细节
- 不命令式
- 不制造性别对立

## 专业边界

- 不远程诊断心理疾病
- 严重情况（持续失眠/抑郁/惊恐）建议找专业人士
- 涉及症状/用药提醒咨询医生
- 不编造案例
- 不确定的事说不确定

## 输出格式

如果属于”不回复的情况”：什么都不输出，完全空白。
如果不属于”不回复的情况”：只输出可以直接发到群里的话。
"""
        else:
            system_prompt = """
           你是云悦公司开发的智能助手。你的核心行为准则如下：

            ## 一、身份与基本行为规范
             你具备以下能力：
               - 可接收和读取各类文档（PDF、Excel、PPT、Word 等），并执行总结、分析、翻译、润色等任务；
               - 可读取图片/照片、网址、抖音链接的内容；
               - 可根据用户提供的文本描述生成或绘制图片；
               - 可搜索各类信息（含图片和视频）以满足用户需求。

            ## 二、工具使用总原则
            1. 优先使用「知识库」检索信息，只有当知识库的信息不足以支撑回答时，才能使用联网搜索；如果知识库信息足够，则不联网。
            2. 对于以下问题，优先参考「知识库」中的信息进行回复：
               - 云悦产品相关问题（如：XX宝）；
               - 企业信息相关问题（如：云悦）；
               - 创始人或负责人相关问题（如：陈沛）。
            3. 当用户提问涉及企业、企业产品、企业负责人、人物信息等内容时，应先尝试通过知识库检索；若知识库无法提供足够信息，再判断为当前信息不足并启用联网搜索。
            4. 若知识库无结果或结果不足，不需要向用户说明“知识库未命中”或“正在联网搜索”，直接继续完成检索与回答。
            5. 不得为了形式完整而强行联网；若知识库已足够回答，则直接基于已有信息作答。

            ## 三、联网搜索触发规则
            仅在以下情况下，才允许调用联网搜索：
            1. 知识库信息不足以支撑回答；
            2. 问题具有明显时效性，例如近3年的数据、最新动态、近期人事变动、当前价格、最新产品信息等；
            3. 问题属于你的知识盲区，且知识库也未覆盖，例如特定企业薪资、实时工商状态、近期新闻事件等；
            4. 用户问题需要依赖最新公开信息，而当前已有信息无法确保准确性。

            若不满足以上条件，则不联网。

            ## 四、搜索与信息验证规则
            当必须联网搜索时，应遵循以下原则：

            1. 搜索范围
               - 默认获取 top10 搜索结果作为候选信息；
               - 优先关注与用户问题强相关的信息。

            2. 来源可信度判断
               - 优先采用高可信来源的信息，例如：
                 - 官方网站、官方公告、官方公众号；
                 - 权威媒体；
                 - 行业机构、公开财报、监管披露、学术或专业数据库。
               - 对来源不明、营销导向强、内容农场、明显搬运或缺乏佐证的信息，应降低权重或直接舍弃。

            3. 信息真实性验证
               - 对关键事实进行交叉验证，尤其是：
                 - 企业名称、产品名称；
                 - 职位、负责人身份；
                 - 时间、金额、价格、融资、营收等关键数据；
                 - 产品能力、发布时间、合作关系等。
               - 重点检查：
                 - 时间是否一致；
                 - 表述是否存在逻辑冲突；
                 - 是否有多个独立来源支持；
                 - 是否存在明显异常或夸张描述。
               - 如果信息可能不实，则直接排除，不用于回答。

            4. 信息整合
               - 优先采用高质量、可交叉验证的信息形成答案；
               - 若多个可信来源一致，可提高回答确定性；
               - 若信息存在冲突，应仅保留相对稳妥、可验证的部分，避免武断下结论；
               - 若搜索结果整体质量较低、无法形成可靠结论，则视为“未搜索到可靠信息”。

            5. 搜索失败处理
               - 若联网搜索后仍无可靠信息，不编造、不猜测；
               - 应直接告诉用户目前无法找到可靠信息。

            ## 五、回答规则

            ### 1. 内容层面
            - 优先回答用户的核心问题，内容应准确、直接、完整；
            - 在不偏离主问题的前提下，可适度补充必要背景，帮助用户理解；
            - 对复杂概念可使用简洁例子或类比辅助说明；
            - 若问题范围较广或需求不明确，先给出简要概述，再覆盖关键点；
            - 大多数情况下不需要提供过多延伸内容，围绕用户主需回答即可；
            - 若信息不足或搜索结果不可靠，应明确说明无法确认，不得编造。

            ### 2. 来源呈现规则
            - 可以内部参考知识库和搜索结果进行作答；
            - 但对用户输出时，**不得暴露参考资料的存在**；
            - 不得出现类似：
              - “根据参考资料”
              - “根据知识库”
              - “根据检索结果”
              - “我查到”
              - “搜索显示”
              等表述；
            - 不需要展示引用链接、角标引用、参考文献列表。

            ### 3. 时效性表达
            - 对企业、产品、负责人、人事变动、价格、营收、融资等容易变化的信息，应自然标注时间范围；
            - 推荐表达方式：
              - “截至2025年3月，……”
              - “从目前公开信息来看，……”
              - “根据2024年下半年的公开信息，……”
            - 时效性表达应自然融入回答，不要生硬罗列。

            ### 4. 格式层面
            通常情况下，对知识问答类问题使用清晰、结构化表达，确保用户轻松理解和使用：
            - 优先使用自然分段；
            - 需要表达顺序关系时，使用有序列表（1. 2. 3.）；
            - 需要表达并列关系时，使用无序列表；
            - 可适度使用加粗突出标题和关键信息；
            - 非必要不使用复杂嵌套列表；
            - 对创作、数理逻辑、阅读理解等任务，按惯常方式回答；
            - 若用户明确指定回复风格，优先满足用户需求。

            ## 六、特殊场景处理
            1. 如果知识库已有云悦、XX宝、陈沛相关信息，优先使用知识库内容，不主动联网补充。
            2. 如果知识库对上述主题信息不足，再进行联网搜索，并仅吸收可信、可验证的信息。
            3. 对敏感、隐私、争议信息保持谨慎，尤其是个人资产、未经证实的履历、传闻、八卦、负面指控等；若缺乏可靠依据，应拒绝采纳或明确表示无法确认。
            4. 若用户提问本身不清晰，可先简短追问澄清；但若已有足够上下文，也可先给出当前可确定的答案。

            ## 七、禁止事项
            1. 不得在知识库信息足够时擅自联网；
            2. 不得把低可信、未验证、可能不实的信息写入答案；
            3. 不得编造事实、时间、数据、人物关系或产品能力；
            4. 不得向用户暴露知识库、检索、搜索策略、来源筛选过程或内部判断过程；
            5. 不得输出”思考过程””搜索关键词””为什么需要搜索”等内部推理内容；
            6. 不得使用”根据参考资料/根据知识库/根据搜索结果”等表述。

            ## 八、最终目标
            在保证回答自然、清晰、易懂的前提下：
            - 优先使用知识库；
            - 仅在必要时联网；
            - 对联网结果进行真实性与可信度验证；
            - 用结构化语言给出准确、稳妥、不过度暴露内部过程的回答。

           """
        system_prompt = {"role": "system", "content": [{"type": "input_text", "text": system_prompt}]}

        api_messages = [system_prompt, build_multimodal_input(latest_user_msg)]
        tools = get_web_search_tools()
        if knowledge_id:
            tools += get_knowledge_search_tools(knowledge_id)
        previous_response_id = get_previous_response_id(user_context.user_id, session_id)

        stream = client.responses.create(
            model=config.MODEL_NAME,
            input=api_messages,
            tools=tools,
            stream=True,
            previous_response_id=previous_response_id,
        )

        accumulated_content = ""
        accumulated_thinking = ""
        accumulated_searching = ""
        response_id = None
        input_tokens = output_tokens = total_tokens = None

        # 将同步阻塞的 stream 迭代放入子线程，通过 Queue 传递给异步生成器
        # 避免阻塞事件循环，保证每个 chunk 到达时立即 yield 推送给前端
        loop = asyncio.get_event_loop()
        queue: asyncio.Queue = asyncio.Queue()

        def _iterate_stream():
            try:
                for chunk in stream:
                    loop.call_soon_threadsafe(queue.put_nowait, chunk)
            except Exception as e:
                loop.call_soon_threadsafe(queue.put_nowait, e)
            finally:
                loop.call_soon_threadsafe(queue.put_nowait, None)  # 结束哨兵

        threading.Thread(target=_iterate_stream, daemon=True).start()

        print("=== 边想边搜启动 ===")
        while True:
            chunk = await queue.get()
            if chunk is None:
                break
            if isinstance(chunk, Exception):
                raise chunk

            chunk_type = getattr(chunk, 'type', '')

            # ① 处理AI思考过程
            if chunk_type == 'response.reasoning_summary_text.delta':
                delta_text = getattr(chunk, 'delta', '')
                if delta_text:
                    accumulated_thinking += delta_text
                    yield f"data: {StreamResponse(content=delta_text, finished=False, model=config.MODEL_NAME, timestamp=datetime.now(), type='thinking').model_dump_json()}\n\n"

            # ② 处理搜索状态
            elif 'web_search_call' in chunk_type:
                if 'in_progress' in chunk_type:
                    _now_str = datetime.now().strftime("%H:%M:%S")
                    msg = f'开始搜索 [{_now_str}]'
                    accumulated_searching += msg + "\n"
                    yield f"data: {StreamResponse(content=msg, finished=False, model=config.MODEL_NAME, timestamp=datetime.now(), type='searching').model_dump_json()}\n\n"
                elif 'completed' in chunk_type:
                    _now_str = datetime.now().strftime("%H:%M:%S")
                    msg = f'搜索完成 [{_now_str}]'
                    accumulated_searching += msg + "\n"
                    yield f"data: {StreamResponse(content=msg, finished=False, model=config.MODEL_NAME, timestamp=datetime.now(), type='searching').model_dump_json()}\n\n"

            # ③ 处理搜索关键词
            elif (chunk_type == 'response.output_item.done'
                  and hasattr(chunk, 'item')
                  and str(getattr(chunk.item, 'id', '')).startswith('ws_')):
                if hasattr(chunk.item, 'action') and hasattr(chunk.item.action, 'query'):
                    query = chunk.item.action.query
                    msg = f'搜索关键词: {query}'
                    accumulated_searching += msg + "\n"
                    yield f"data: {StreamResponse(content=msg, finished=False, model=config.MODEL_NAME, timestamp=datetime.now(), type='searching').model_dump_json()}\n\n"

            # ④ 处理最终回答文本（实时推送给前端）
            elif chunk_type == 'response.output_text.delta':
                delta_text = getattr(chunk, 'delta', '')
                if delta_text:
                    accumulated_content += delta_text
                    yield f"data: {StreamResponse(content=delta_text, finished=False, model=config.MODEL_NAME, timestamp=datetime.now()).model_dump_json()}\n\n"

            # ⑤ 处理响应完成事件
            elif chunk_type == 'response.completed':
                response_obj = getattr(chunk, 'response', None)
                if response_obj and hasattr(response_obj, 'id'):
                    response_id = response_obj.id
                usage_obj = getattr(response_obj, 'usage', None) if response_obj else None
                if usage_obj:
                    input_tokens = getattr(usage_obj, 'input_tokens', None)
                    output_tokens = getattr(usage_obj, 'output_tokens', None)
                    total_tokens = getattr(usage_obj, 'total_tokens', None)
                save_chat_log(
                    user_id=user_context.user_id,
                    question=latest_user_msg.content,
                    stream_mode=True,
                    raw_response=repr(response_obj),
                    status="success",
                )

        print(f"\n\n=== 边想边搜完成 [{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] ===")

        if accumulated_content:
            # 保存助手消息到 DB（含 thinking / searching）
            save_chat_history(
                user_id=user_context.user_id,
                session_id=session_id,
                role="assistant",
                content=accumulated_content,
                timestamp=datetime.now(),
                response_id=response_id,
                thinking=accumulated_thinking or None,
                searching=accumulated_searching or None,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                total_tokens=total_tokens,
                app_name=user_context.app_name,
            )
            yield f"data: {StreamResponse(content='', finished=True, model=config.MODEL_NAME, timestamp=datetime.now()).model_dump_json()}\n\n"

    except Exception as e:
        error_response = {
            "error": str(e),
            "finished": True,
            "timestamp": datetime.now().isoformat()
        }
        save_chat_log(
            user_id=user_context.user_id,
            question=latest_user_msg.content if latest_user_msg else "",
            stream_mode=True,
            status="error",
            error=str(e),
        )
        yield f"data: {json.dumps(error_response)}\n\n"



@router.post("/chat/{promptType}", response_model=ChatResponse)
async def chat(
    promptType: str,
    request: ChatRequest,
    user_context: UserContext = Depends(resolve_user_id),
):
    try:
        if request.stream:
            return StreamingResponse(
                generate_stream_response(request, user_context, promptType),
                media_type="text/plain",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                    "Content-Type": "text/event-stream",
                }
            )
        # 以下是非流格式的输出， 采用了豆包助手工具，目前项目中没有使用到
        # 以下是非流式输出处理
        session_id = request.session_id
        latest_user_msg = get_latest_user_message(request.messages)
        if not latest_user_msg:
            raise ValueError("请求中没有找到user角色的消息")

        ai_config = get_config_by_app_name(user_context.app_name)
        if not ai_config:
            raise ValueError(f"未找到appName '{user_context.app_name}' 的配置")

        client = get_client(user_context.app_name)

        user_attachments = [att.model_dump() for att in latest_user_msg.attachments] if latest_user_msg.attachments else None
        save_chat_history(
            user_id=user_context.user_id,
            session_id=session_id,
            role=latest_user_msg.role,
            content=latest_user_msg.content,
            timestamp=datetime.now(),
            attachments=user_attachments,
        )

        api_messages = [build_multimodal_input(latest_user_msg)]
        tools = get_doubao_tools()
        previous_response_id = get_previous_response_id(user_context.user_id, session_id)

        response = client.responses.create(
            model=config.MODEL_NAME,
            input=api_messages,
            tools=tools,
            stream=False,
            store=True,
            previous_response_id=previous_response_id,
        )

        save_chat_log(
            user_id=user_context.user_id,
            question=latest_user_msg.content,
            stream_mode=False,
            raw_response=repr(response),
            status="success",
        )

        if not (response.output and len(response.output) > 0):
            raise HTTPException(status_code=500, detail="AI模型返回了空响应")

        message_content = ""
        for item in response.output:
            if hasattr(item, 'type') and item.type == 'doubao_app_call':
                if hasattr(item, 'blocks') and item.blocks:
                    for block in item.blocks:
                        if hasattr(block, 'type') and block.type == 'output_text' and hasattr(block, 'text'):
                            message_content += block.text
            elif hasattr(item, 'type') and item.type == 'message':
                if hasattr(item, 'content'):
                    if isinstance(item.content, list):
                        for content_item in item.content:
                            if hasattr(content_item, 'text'):
                                message_content += content_item.text
                    else:
                        message_content += str(item.content)

        if not message_content:
            raise HTTPException(status_code=500, detail="无法从AI响应中提取文本内容")

        now = datetime.now()
        save_chat_history(
            user_id=user_context.user_id,
            session_id=session_id,
            role="assistant",
            content=message_content,
            timestamp=now,
            response_id=response.id,
        )

        assistant_message = ChatMessage(
            role="assistant",
            content=message_content,
            timestamp=now,
            response_id=response.id,
        )
        return ChatResponse(
            message=assistant_message,
            model=response.model,
            usage=response.usage.model_dump() if response.usage else None,
            response_id=response.id,
        )

    except HTTPException:
        raise
    except Exception as e:
        error_message = f"处理聊天请求时发生错误: {str(e)}"
        save_chat_log(
            user_id=user_context.user_id,
            question=request.messages[-1].content if request.messages else "",
            stream_mode=request.stream,
            status="error",
            error=error_message,
        )
        raise HTTPException(status_code=500, detail=error_message)


@router.get("/history")
async def get_user_history(
    current_user: Annotated[User, Depends(get_current_active_user)],
    sessionId: str = Query(..., description="会话ID"),
) -> List[ChatMessage]:
    docs = get_chat_history(current_user.userId, sessionId)
    return [
        ChatMessage(
            role=doc["role"],
            content=doc["content"],
            attachments=[FileAttachment(**att) for att in doc["attachments"]] if doc.get("attachments") else None,
            timestamp=doc.get("timestamp"),
            response_id=doc.get("response_id"),
            thinking=doc.get("thinking"),
            searching=doc.get("searching"),
        )
        for doc in docs
    ]


@router.delete("/history")
async def clear_user_history(
    current_user: Annotated[User, Depends(get_current_active_user)],
    sessionId: str = Query(..., description="会话ID"),
):
    deleted_count = delete_chat_history(current_user.userId, sessionId)
    if deleted_count > 0:
        return {"message": "聊天历史已清空", "user": current_user.userId, "deleted_messages": deleted_count, "timestamp": datetime.now()}
    return {"message": "用户没有聊天历史", "user": current_user.userId, "deleted_messages": 0, "timestamp": datetime.now()}


@router.get("/sessions")
async def get_user_sessions(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return get_sessions(current_user.userId)


@router.get("/health")
async def health_check():
    return {"status": "healthy", "timestamp": datetime.now(), "version": "1.0.0", "model": config.MODEL_NAME}


router.tags = ["聊天服务"]
router.responses = {
    401: {"description": "未授权 - 需要有效的JWT令牌"},
    429: {"description": "请求过多 - 配额已用完"},
    500: {"description": "服务器内部错误"}
}