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

"""
IM Bot 真实 LLM 调用。

原实现位于 channels.py（通用频道管理模块），该模块已移除。
为保持 IM Bot（微信机器人等）正常生成回复，将 _call_real_llm 迁移至此独立模块。

行为（对齐 AI 聊天）：
- Issue 7：图像/视频/语音意图优先走平台媒体模型（与 AI 聊天同一套 Auto 路由），
  不再对"制作图片"这类请求只回文本。
- Issue 8：默认 Auto 模型；每轮先自动发现本账号已安装且平台已部署的技能，
  多个适用时由模型自行判断选 1 个最优，通过 function calling 执行。
- 2026-07-15：修复 trust_env=False + proxy 处理，避免容器坏代理导致调用失败；
  媒体生成前先经 _optimize_media_prompt 精炼提示词，提升 AGNES 生成质量。
"""
import asyncio
import json
import re as _re
from typing import Optional
from urllib.parse import urlparse as _urlparse

import httpx
from sqlalchemy.orm import Session

from ..capability_registry import resolve_auto_model
from ..models import AIModel, ModelProvider


def _proxy_for_url(url: str) -> str | None:
    """按目标域名选择是否走代理（与 chat.py 同逻辑）。

    对 AGNES 等国内域名返回 None（直连），平台自有域名直连，其余走智能路由代理。
    """
    import os
    host = (_urlparse(url).hostname or "").lower()
    if "agnes-ai.com" in host or "agnes-ai.space" in host:
        return None
    if "qihuyun.com" in host:
        return None
    return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890"


def _run_async(coro):
    """在同步上下文中运行协程，兼容「无事件循环」与「已有事件循环」两种场景。

    IM 个人微信在独立线程中调用（无事件循环）→ asyncio.run；
    IM 多平台 Webhook 可能在 FastAPI 协程内调用（已有事件循环）→ 丢到线程池执行。
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None
    if loop is not None and loop.is_running():
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor(1) as ex:
            return ex.submit(lambda: asyncio.run(coro)).result()
    return asyncio.run(coro)


def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im", user_id: str = None) -> str:
    """调用七虎AI 生成回复，供 IM Bot 复用。支持媒体意图路由与已安装技能自动发现。"""
    try:
        from app.api.chat import (
            _generate_media,
            detect_media_intent,
            _discover_installed_skills,
            _build_system_prompt,
            _execute_skill_tool,
        )

        # ===== 1) 媒体意图路由（Issue 7）=====
        intent = detect_media_intent(prompt)
        if intent:
            ai_model = resolve_auto_model(intent, db)
            if ai_model:
                try:
                    result = _run_async(_generate_media(db, ai_model, prompt, user_id))
                except Exception as e:
                    result = f"（媒体生成失败：{e}）"
                if result:
                    # IM 不渲染 HTML，把 <img>/<video> 还原为纯文本链接返回
                    s = result.strip()
                    if s.startswith("<img") or s.startswith("<video"):
                        m = _re.search(r'src="([^"]+)"', s)
                        url = m.group(1) if m else ""
                        label = "图片" if s.startswith("<img") else "视频"
                        return f"已为你生成{label}：{url}" if url else "（生成完成，但未能提取媒体地址）"
                    # 媒体生成失败：_generate_media 已返回友好中文（如限频/队列繁忙），
                    # 直接透传，绝不把原始提示词/标签/JSON 透传给用户
                    if s.startswith("⚠️") or s.startswith("（媒体"):
                        return s
                    return "（媒体生成未能返回有效内容，请重试）"
                    return result

        # ===== 2) 文本回复 + 技能自动发现（Issue 8）=====
        ai_model = resolve_auto_model("text", db)
        if not ai_model:
            return "（未配置默认文本模型，无法生成回复）"
        provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first()
        api_key = ai_model.api_key or (provider.api_key if provider else None)
        api_url = ai_model.endpoint
        if not api_url and provider and provider.api_base_url:
            api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"
        if not (api_key and api_url):
            return "（模型未配置 API Key 或地址，无法生成回复）"

        # 自动发现本账号已安装且平台已部署的技能
        skills = []
        if user_id:
            try:
                skills = _discover_installed_skills(db, user_id)[:15]
            except Exception:
                skills = []

        system_prompt, function_definitions = _build_system_prompt(
            skills=skills,
            agents=[],
            kb_context="",
            user_content=prompt,
            mcp=[],
            auto_discovered=bool(skills),
        )
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ]

        max_rounds = 3
        ai_content = ""
        for _ in range(max_rounds):
            request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
            if function_definitions and skills:
                request_body["tools"] = [
                    {"type": "function", "function": fd} for fd in function_definitions
                ]
                request_body["tool_choice"] = "auto"
            with httpx.Client(
                timeout=ai_model.timeout or 120,
                proxy=_proxy_for_url(api_url),
                trust_env=False,
            ) as client:
                resp = client.post(
                    api_url,
                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                    json=request_body,
                )
                if resp.status_code != 200:
                    return f"（AI 模型调用失败 HTTP {resp.status_code}）"
                data = resp.json()
                message = data.get("choices", [{}])[0].get("message", {})
                tool_calls = message.get("tool_calls", [])
                if tool_calls and skills:
                    messages.append({
                        "role": "assistant",
                        "content": message.get("content") or "",
                        "tool_calls": tool_calls,
                    })
                    for tc in tool_calls:
                        func_name = tc.get("function", {}).get("name", "")
                        try:
                            func_args = json.loads(tc.get("function", {}).get("arguments", "{}"))
                        except Exception:
                            func_args = {"input": tc.get("function", {}).get("arguments", "")}
                        tool_result = _execute_skill_tool(db, func_name, func_args, skills, prompt)
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tc.get("id", "call_1"),
                            "content": tool_result,
                        })
                    continue
                ai_content = message.get("content", "")
                break
        return ai_content or "（AI 回复为空）"
    except Exception as e:
        return f"（AI 回复生成失败：{e}）"