46 lines
2.0 KiB
Python
46 lines
2.0 KiB
Python
"""
|
||
IM Bot 真实 LLM 调用。
|
||
|
||
原实现位于 channels.py（通用频道管理模块），该模块已移除。
|
||
为保持 IM Bot（微信机器人等）正常生成回复，将 _call_real_llm 迁移至此独立模块。
|
||
"""
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..capability_registry import resolve_auto_model
|
||
from ..models import AIModel, ModelProvider
|
||
|
||
|
||
def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im") -> str:
|
||
"""调用七虎AI 默认文本模型生成回复，供 IM Bot 复用。"""
|
||
try:
|
||
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 或地址，无法生成回复）"
|
||
messages = [
|
||
{"role": "system", "content": "你是七虎AI 助手，负责在即时通讯中回复用户消息。"},
|
||
{"role": "user", "content": prompt},
|
||
]
|
||
request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
|
||
with httpx.Client(timeout=ai_model.timeout or 120) 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()
|
||
return data.get("choices", [{}])[0].get("message", {}).get("content", "")
|
||
except Exception as e:
|
||
return f"（AI 回复生成失败：{e}）"
|