qihuaiflow/backend/python/app/api/chat.py

from fastapi import APIRouter, Depends, HTTPException, Query, Response
from fastapi.responses import StreamingResponse

from sqlalchemy.orm import Session

from sqlalchemy import and_, desc, asc, func

from pydantic import BaseModel, Field

from typing import Optional

import uuid

from datetime import datetime, timezone, timedelta

import re

import json

import asyncio

import math

import httpx

import os

import logging

from ..database import get_db, SessionLocal

from ..models import (







    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,







    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,







    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,







    MCPServer,







    gen_uuid







)

from ..auth import get_current_user

from ..platform_knowledge import get_platform_knowledge

from ..mcp_client import list_mcp_tools, call_mcp_tool
from ..core.claw_adapters import SseAdapter

from ..middleware.log_service import LogService

from ..capability_registry import resolve_auto_model

from app.api.chat_agents import (_call_coze_agent, _call_dify_agent, _call_external_agent, _call_n8n_agent, _normalize_dify_chat_url)

from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context, _build_chat_history, _maybe_compress_history)

from app.api.chat_knowledge import (_build_knowledge_graph)

from app.api.chat_mcp import (_list_enabled_mcp_servers, _resolve_active_mcp_servers, _resolve_market_mcp)

from app.api.chat_media import (_agnes_post_json, _friendly_media_err, _generate_media, _generate_media_task, _optimize_media_prompt, _write_media_token_log, detect_ambiguous_creative_intent, detect_media_intent)

from app.api.chat_skill import (_bm25_search, _discover_installed_skills, _execute_skill_tool, _parse_skill_content, _resolve_market_agents, _resolve_market_skills)

from app.api.chat_utils import (CST_TZ, CreateConversationRequest, SendMessageRequest, UpdateConversationRequest, _HK_PROXY, _extract_text_from_json, _html_escape, _outbound_proxy, _proxy_for_url, cst_now, logger)



"""







增强版 Chat API — Phase 2.5







包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入,







      多轮上下文保持, BM25 KB检索, Function Calling循环增强







"""



from app.api.chat_service import _send_message_core
from app.core.llm_router import set_route_ctx



router = APIRouter()



_MEDIA_RELAY_ALLOWED = ("agnes-ai.com", "agnes-ai.space", "qihuyun.com")





# Media relay client (trust_env=False to bypass proxy for image fetches)
_media_client = None
def _get_media_client():
    global _media_client
    if _media_client is None:
        _media_client = httpx.Client(timeout=30.0, follow_redirects=True, trust_env=False)
    return _media_client


@router.get("/media/relay")







async def media_relay(url: str = Query(...)):







    """中转 AGNES 媒体(图片/视频)到平台域名，供微信/iLink 内联渲染。















    仅允许白名单域名；否则 400。返回的媒体流保持原 content-type。







    """

    import urllib.parse as _up
    import httpx
    from fastapi.responses import StreamingResponse

    host = (_up.urlparse(url).hostname or "").lower()
    # 白名单域名检查
    allowed = ("siliconflow.cn", "s3.siliconflow.cn", "aliyuncs.com", "tencentcos.cn",
               "qpic.cn", "alicdn.com", "api.openai.com", "myqcloud.com")
    if not any(a in host for a in allowed):
        raise HTTPException(status_code=400, detail="unsupported media host")

    try:
        client = _get_media_client()
        resp = client.get(url, timeout=30.0, follow_redirects=True)
        if resp.status_code != 200:
            raise HTTPException(status_code=502, detail=f"upstream returned {resp.status_code}")
        ct = resp.headers.get("content-type", "application/octet-stream")
        cl = resp.headers.get("content-length")
        headers = {}
        if ct:
            headers["content-type"] = ct
        if cl:
            headers["content-length"] = cl
        headers["cache-control"] = "public, max-age=3600"
        return StreamingResponse(
            resp.iter_bytes(chunk_size=65536),
            status_code=200,
            headers=headers,
            media_type=ct
        )
    except httpx.TimeoutException:
        raise HTTPException(status_code=504, detail="upstream timeout")
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"relay error: {str(e)[:200]}")








    import urllib.parse as _up







    host = (_up.urlparse(url).hostname or "").lower()







    if not any(allowed in host for allowed in _MEDIA_RELAY_ALLOWED):







        raise HTTPException(status_code=400, detail="unsupported media host")







    try:







        async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client:







            r = await client.get(url)







            if r.status_code != 200:







                raise HTTPException(status_code=r.status_code, detail="upstream error")







            content_type = r.headers.get("content-type", "application/octet-stream")







            data = await r.aread()







        return Response(content=data, media_type=content_type)







    except HTTPException:







        raise







    except Exception as e:







        raise HTTPException(status_code=502, detail=f"relay failed: {e}")



@router.get("/conversations")







def list_conversations(







    status: Optional[str] = Query(None),







    search: Optional[str] = Query(None),







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):







    query = db.query(Conversation).filter(Conversation.user_id == current_user.id)















    if status:







        query = query.filter(Conversation.status == status)







    else:







        query = query.filter(Conversation.status != "archived")















    if search:







        query = query.filter(Conversation.title.ilike(f"%{search}%"))















    query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at))















    conversations = query.all()







    result = []







    for conv in conversations:







        result.append({







            "id": conv.id,







            "title": conv.title,







            "model_id": conv.model_id,







            "status": conv.status,







            "pinned": conv.pinned,







            "knowledge_graph_count": conv.knowledge_graph_count or 0,







            "source": getattr(conv, "source", None),







            "created_at": str(conv.created_at) if conv.created_at else None,







            "updated_at": str(conv.updated_at) if conv.updated_at else None,







        })















    return {"code": 0, "data": result, "message": "ok"}



@router.post("/conversations")







def create_conversation(







    req: CreateConversationRequest,







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):







    conv = Conversation(







        id=gen_uuid(),







        user_id=current_user.id,







        title=req.title,







        model_id=req.model_id,







        status="active",







        pinned=False,







        knowledge_graph_count=0,







        created_at=cst_now(),







        updated_at=cst_now(),







    )







    db.add(conv)







    db.commit()







    db.refresh(conv)















    return {







        "code": 0,







        "data": {







            "id": conv.id,







            "title": conv.title,







            "model_id": conv.model_id,







            "status": conv.status,







            "pinned": conv.pinned,







            "knowledge_graph_count": conv.knowledge_graph_count,







            "created_at": str(conv.created_at) if conv.created_at else None,







            "updated_at": str(conv.updated_at) if conv.updated_at else None,







        },







        "message": "创建成功",







    }



@router.put("/conversations/{conv_id}")







def update_conversation(







    conv_id: str,







    req: UpdateConversationRequest,







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):







    conv = db.query(Conversation).filter(







        and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)







    ).first()







    if not conv:







        raise HTTPException(status_code=404, detail="对话不存在")















    if req.title is not None:







        conv.title = req.title







    if req.status is not None:







        conv.status = req.status







    if req.pinned is not None:







        conv.pinned = req.pinned







    # Stage B.2: 保存agent/skill/kb选择到对话上下文







    # (存储在第一条system消息的knowledge_ids/agent_ids/skill_ids字段中)







    conv.updated_at = cst_now()















    db.commit()







    db.refresh(conv)















    return {"code": 0, "data": {







        "id": conv.id, "title": conv.title, "status": conv.status, "pinned": conv.pinned,







        "updated_at": str(conv.updated_at),







    }, "message": "更新成功"}



@router.delete("/conversations/{conv_id}")







def delete_conversation(







    conv_id: str,







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):







    conv = db.query(Conversation).filter(







        and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)







    ).first()







    if not conv:







        raise HTTPException(status_code=404, detail="对话不存在")















    conv.status = "archived"







    conv.updated_at = cst_now()







    db.commit()















    return {"code": 0, "message": "删除成功"}



@router.get("/conversations/{conv_id}/messages")







def get_messages(







    conv_id: str,







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):







    conv = db.query(Conversation).filter(







        and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)







    ).first()







    if not conv:







        raise HTTPException(status_code=404, detail="对话不存在")















    messages = (







        db.query(Message)







        .filter(Message.conversation_id == conv_id)







        .order_by(asc(Message.created_at), asc(Message.id))







        .all()







    )















    result = []







    for msg in messages:







        result.append({







            "id": msg.id,







            "conversation_id": msg.conversation_id,







            "role": msg.role,







            "content": msg.content,







            "token_used": msg.token_used or 0,







            "model_id": msg.model_id,







            "used_skills": msg.used_skills or [],

            "used_mcp": msg.used_mcp or [],







            "created_at": str(msg.created_at) if msg.created_at else None,







        })















    return {"code": 0, "data": result, "message": "ok"}



@router.post("/conversations/{conv_id}/messages")







async def send_message(







    conv_id: str,







    req: SendMessageRequest,







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):

    from app.core.observability import span

    # #876 双 Claw 维度：入口最早注入，确保外层 span 即带 tenant_id + claw_type 标签
    set_route_ctx(
        tenant_id=str(getattr(current_user, "tenant_id", "") or ""),
        claw_type="web",
    )

    async with span("chat.send_message", conv_id=conv_id, user_id=getattr(current_user, "id", None)):

        return await _send_message_core(conv_id, req, db, current_user)


@router.post("/conversations/{conv_id}/messages/stream")
async def send_message_stream(
    conv_id: str,
    req: SendMessageRequest,
    current_user: User = Depends(get_current_user),
):
    """SSE 流式聊天端点(#873)。

    与 /messages 共用 EngineCore(_send_message_core)，仅通过 stream=True +
    on_event=SseAdapter 实时吐出增量(delta/reasoning/step)，末尾推 done 终帧
    (含 message_id 等)。SseAdapter 保证线格式与改造前逐字节一致。
    """
    db = SessionLocal()
    try:
        # #876 双 Claw 维度：入口最早注入。runner() 经 asyncio.create_task 派生子协程，
        # contextvars 自动随上下文传播，使 _send_message_core 内所有 span 均带 tenant_id + claw_type。
        set_route_ctx(
            tenant_id=str(getattr(current_user, "tenant_id", "") or ""),
            claw_type="web",
        )

        adapter = SseAdapter(
            trace_id=f"sse-{conv_id}",
            tenant_id=str(getattr(current_user, "tenant_id", "") or ""),
        )

        async def runner():
            try:
                await _send_message_core(
                    conv_id, req, db, current_user,
                    stream=True, on_event=adapter,
                )
            except Exception as exc:
                logger.warning(f"[sse] chat stream error: {exc}", exc_info=True)
                await adapter.emit_error(str(exc))
            finally:
                await adapter.close()

        task = asyncio.create_task(runner())

        async def event_gen():
            try:
                async for frame in adapter.frames():
                    yield frame
            finally:
                await task
                db.close()

        return StreamingResponse(event_gen(), media_type="text/event-stream")
    except Exception:
        db.close()
        raise

@router.post("/conversations/{conv_id}/clear")







def clear_conversation(







    conv_id: str,







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):







    conv = db.query(Conversation).filter(







        and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)







    ).first()







    if not conv:







        raise HTTPException(status_code=404, detail="对话不存在")















    db.query(Message).filter(







        Message.conversation_id == conv_id







    ).delete()















    conv.knowledge_graph_count = 0







    conv.updated_at = cst_now()







    db.commit()















    return {"code": 0, "message": "对话上下文已清空"}



@router.post("/conversations/{conv_id}/to-knowledge-graph")







def convert_to_knowledge_graph(







    conv_id: str,







    db: Session = Depends(get_db),







    current_user: User = Depends(get_current_user),







):







    """将对话内容转换为知识图谱（幂等：同一对话仅生成一次知识库）"""







    conv = db.query(Conversation).filter(







        and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)







    ).first()







    if not conv:







        raise HTTPException(status_code=404, detail="对话不存在")















    messages = (







        db.query(Message)







        .filter(Message.conversation_id == conv_id)







        .order_by(asc(Message.created_at), asc(Message.id))







        .all()







    )















    if not messages:







        raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱")















    # 幂等：若该对话已生成过知识图谱且知识库仍存在，则直接刷新返回







    if conv.knowledge_graph_base_id:







        existing_kb = db.query(KnowledgeBase).filter(







            KnowledgeBase.id == conv.knowledge_graph_base_id







        ).first()







        if existing_kb:







            graph_data = _build_knowledge_graph(db, existing_kb)







            conv.knowledge_graph_count = 1







            conv.updated_at = cst_now()







            db.commit()







            return {







                "code": 0,







                "data": {







                    "knowledge_base_id": existing_kb.id,







                    "knowledge_base_name": existing_kb.name,







                    "graph": graph_data,







                    "knowledge_graph_count": 1,







                    "already_exists": True,







                },







                "message": "知识图谱已存在, 已刷新展示",







            }















    kb_name = f"对话图谱-{conv.title}"







    kb = KnowledgeBase(







        id=gen_uuid(),







        user_id=current_user.id,







        name=kb_name,







        description=f"从对话「{conv.title}」自动生成的知识图谱",







        directory_path="/",







        document_count=0,







        created_at=cst_now(),







        updated_at=cst_now(),







    )







    db.add(kb)







    db.flush()















    doc_count = 0







    for idx, msg in enumerate(messages):







        role_label = "用户" if msg.role == "user" else "AI"







        preview = re.sub(r'[\\n\\r\\t]+', ' ', msg.content[:40]).strip()







        filename = f"msg_{idx+1:03d}_{role_label}_{preview}.txt"















        doc = KnowledgeDocument(







            id=gen_uuid(),







            knowledge_base_id=kb.id,







            filename=filename,







            file_type="txt",







            file_size=len(msg.content.encode('utf-8')),







            content=msg.content,







            status="completed",







            chunk_count=max(1, len(msg.content) // 500 + 1),







            created_at=cst_now(),







            updated_at=cst_now(),







        )







        db.add(doc)







        doc_count += 1















    kb.document_count = doc_count







    db.flush()















    # 构建知识图谱节点和边







    graph_data = _build_knowledge_graph(db, kb)







    conv.knowledge_graph_base_id = kb.id







    conv.knowledge_graph_count = 1







    conv.updated_at = cst_now()















    db.commit()















    LogService.log_operation(







        user_id=current_user.id, username=current_user.username, action="convert_to_graph",







        target="conversation", target_id=conv_id,







        detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name}",







    )















    return {







        "code": 0,







        "data": {







            "knowledge_base_id": kb.id,







            "knowledge_base_name": kb.name,







            "graph": graph_data,







            "knowledge_graph_count": 1,







            "already_exists": False,







        },







        "message": f"知识图谱构建完成",







    }



@router.get("/models")







def get_models(db: Session = Depends(get_db)):







    models = db.query(AIModel).filter(AIModel.is_enabled == True).order_by(AIModel.is_default.desc(), AIModel.name.asc()).all()







    result = []







    for m in models:







        provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first()







        result.append({







            "id": m.id,







            "name": m.name,







            "display_name": getattr(m, "display_name", None) or m.name,







            "provider_id": m.provider_id,







            "provider_name": provider.name if provider else "",







            "capabilities": m.capabilities or [],







            "status": m.status,







            "price_per_million_tokens": float(m.price_per_million_tokens or 0),







            "is_enabled": m.is_enabled,







            "is_default": bool(m.is_default) if hasattr(m, 'is_default') else False,







            "total_calls": m.total_calls or 0,







            "total_tokens": m.total_tokens or 0,







            "created_at": str(m.created_at) if m.created_at else None,







        })







    result.sort(key=lambda x: (0 if x["is_default"] else 1, x["name"]))







    return {"code": 0, "data": result, "message": "ok"}