fix: 修复AI聊天/IM多项问题(媒体直连+技能自动发现+agent/mcp上下文)

- Issue4/7: 图像/视频生成对AGNES国内域名强制直连(trust_env=False),修复Server disconnected(代理172.18.0.1:7890返回502)

- Issue8: 修复_parse_skill_content缺失def导致的NameError;新增_discover_installed_skills默认技能自动发现

- Issue2/3: system prompt注入agent资料与mcp描述/endpoint

- Issue7: IM Bot制作图片走Auto路由媒体模型
This commit is contained in:
QihuAI Design committed 2026-07-15 12:23:56 +08:00
1 parent bc2899ed09
commit 996d5413da
2 files changed
+232 -29

No files matched your search

+116 -15
View File
@@ -125,13 +125,30 @@ def _html_escape(s: str) -> str:
def _outbound_proxy() -> str: def _outbound_proxy() -> str:
"""出网代理地址。容器内直连 apihub.agnes-ai.com 会被对端重置 """出网代理地址(兜底)。默认经 172.18.0.1:7890 智能路由代理;
(httpx RemoteProtocolError: Server disconnected without sending a response), 运行时若未注入代理环境变量则回退到该地址。"""
必须经 172.18.0.1:7890 智能路由代理(其对国内 AGNES 域名走直连)。"""
import os import os
return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890" return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890"
def _proxy_for_url(url: str) -> str | None:
"""按目标域名选择是否走代理。
实测(2026-07-15):apihub.agnes-ai.com / platform-outputs.agnes-ai.space 为国内域名,
容器内直连稳定返回 200;而经智能路由代理 172.18.0.1:7890 当前返回 502,导致
httpx 报 "Server disconnected without sending a response"。
因此对 AGNES 等国内域名返回 None(直连),其余域名才走 _outbound_proxy()。
注意:httpx 默认 trust_env=True 会忽略 proxy=None 而改用环境变量 HTTP_PROXY,
故调用方必须同时传 trust_env=False 才能真正直连。
"""
from urllib.parse import urlparse
host = (urlparse(url).hostname or "").lower()
if "agnes-ai.com" in host or "agnes-ai.space" in host:
return None
return _outbound_proxy()
async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str: async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str:
"""图像/视频生成模型统一入口:按 capabilities 调用 AGNES 等媒体生成 API。 """图像/视频生成模型统一入口:按 capabilities 调用 AGNES 等媒体生成 API。
返回可直接在 AI 聊天中渲染的富文本(含 <img>/<video> 标签)。""" 返回可直接在 AI 聊天中渲染的富文本(含 <img>/<video> 标签)。"""
@@ -151,14 +168,16 @@ async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str
api_url = f"{provider.api_base_url.rstrip('/')}/v1/images/generations" if provider else "" api_url = f"{provider.api_base_url.rstrip('/')}/v1/images/generations" if provider else ""
if not api_url: if not api_url:
return "⚠️ 该图像模型未配置 Endpoint。" return "⚠️ 该图像模型未配置 Endpoint。"
# 注意:extra_body 是 OpenAI SDK 专属参数,不能直接作为原始 JSON 字段下发,
# AGNES 收到未知顶层字段会直接断开连接(表现为 "Server disconnected without
# sending a response")。保持与 OpenAI 兼容的极简结构即可。
payload = { payload = {
"model": ai_model.name, "model": ai_model.name,
"prompt": prompt, "prompt": prompt,
"size": "1024x768", "size": "1024x768",
"extra_body": {"response_format": "url"},
} }
try: try:
async with httpx.AsyncClient(timeout=timeout, proxy=_outbound_proxy()) as client: async with httpx.AsyncClient(timeout=timeout, proxy=_proxy_for_url(api_url), trust_env=False) as client:
resp = await client.post(api_url, headers=headers, json=payload) resp = await client.post(api_url, headers=headers, json=payload)
if resp.status_code != 200: if resp.status_code != 200:
return f"⚠️ 图像生成失败 (HTTP {resp.status_code}): {resp.text[:200]}" return f"⚠️ 图像生成失败 (HTTP {resp.status_code}): {resp.text[:200]}"
@@ -194,7 +213,7 @@ async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str
"frame_rate": 24, "frame_rate": 24,
} }
try: try:
async with httpx.AsyncClient(timeout=timeout, proxy=_outbound_proxy()) as client: async with httpx.AsyncClient(timeout=timeout, proxy=_proxy_for_url(api_url), trust_env=False) as client:
resp = await client.post(api_url, headers=headers, json=payload) resp = await client.post(api_url, headers=headers, json=payload)
if resp.status_code != 200: if resp.status_code != 200:
return f"⚠️ 视频生成失败 (HTTP {resp.status_code}): {resp.text[:200]}" return f"⚠️ 视频生成失败 (HTTP {resp.status_code}): {resp.text[:200]}"
@@ -450,6 +469,35 @@ def _resolve_market_mcp(db: Session, mcp_ids: list, user_id: str) -> list:
}) })
return results return results
def _discover_installed_skills(db: Session, user_id: str, limit: int = 15) -> list:
"""
Issue 8: 默认技能自动发现。
检索「本账号已安装(app_installations) 且 平台已部署发布(MarketApp.status='published')」的
skill 型应用,返回可供 function calling 使用的技能定义列表(最多 limit 个)。
仅当用户未显式选择技能时才启用;多个技能适用时由模型自行判断选 1 个最优。
"""
if not user_id:
return []
installed = db.query(AppInstallation).filter(
AppInstallation.user_id == user_id
).all()
app_ids = [i.app_id for i in installed]
if not app_ids:
return []
apps = db.query(MarketApp).filter(
MarketApp.id.in_(app_ids),
MarketApp.type == "skill",
MarketApp.status == "published",
).all()
ids = [a.id for a in apps]
if not ids:
return []
return _resolve_market_skills(db, ids, user_id)[:limit]
def _parse_skill_content(content: str) -> dict:
""" """
解析SKILL.md格式内容, 提取结构化信息。 解析SKILL.md格式内容, 提取结构化信息。
SKILL.md格式: SKILL.md格式:
@@ -727,6 +775,7 @@ def _build_system_prompt(
kb_context: str, kb_context: str,
user_content: str, user_content: str,
mcp: list = None, mcp: list = None,
auto_discovered: bool = False,
) -> tuple: ) -> tuple:
""" """
动态组装system prompt + function definitions。 动态组装system prompt + function definitions。
@@ -747,21 +796,41 @@ def _build_system_prompt(
function_definitions = [] function_definitions = []
# Stage B.1: Agent角色注入 # Stage B.1: Agent角色注入(Issue 2:让用户询问 agent 能力时,AI 能“看到”该 agent 资料)
if agents: if agents:
for agent in agents: for agent in agents:
name = agent.get("name", "")
role = agent.get("role_prompt", "") role = agent.get("role_prompt", "")
instructions = agent.get("instructions", "") instructions = agent.get("instructions", "")
desc = agent.get("description", "")
block = []
if name:
block.append(
f"用户已选择智能体「{name}」。当用户询问该智能体的功能、能力、用法或相关问题时,"
f"请基于以下资料准确作答,不要说“不知道这个智能体”。"
)
if role: if role:
system_parts.append(f"当前角色设定: {role}") block.append(f"该智能体的角色设定: {role}")
if desc:
block.append(f"该智能体的简介: {desc}")
if instructions: if instructions:
system_parts.append(f"角色行为指引: {instructions}") block.append(f"该智能体的行为指引: {instructions}")
if block:
system_parts.append("\n".join(block))
# Stage A.2: Skill工具注入 # Stage A.2: Skill工具注入
if skills: if skills:
skill_descs = ", ".join([f"{s['name']}({s.get('category', 'general')})" for s in skills]) skill_descs = ", ".join([f"{s['name']}({s.get('category', 'general')})" for s in skills])
system_parts.append(f"可用技能工具: {skill_descs}") system_parts.append(f"可用技能工具: {skill_descs}")
system_parts.append("重要规则: 当用户请求涉及上述技能工具的处理范围时, 你必须优先通过function calling调用对应工具, 而不是直接回复文字。调用后将获得工具执行结果, 基于结果生成最终回复。") if auto_discovered:
# Issue 8:每轮默认先寻找本账号已安装且平台已部署的技能,多个适用时 AI 自行判断选 1 个最优
system_parts.append(
"重要规则: 以上技能为本账号已安装且平台已部署的技能。每轮对话请先判断用户请求是否可由其中某个技能处理;"
"若多个技能都适用,请自行判断并选择最贴切的 1 个优先调用,不要一次性全部调用。"
"调用对应工具后将获得执行结果,基于结果生成最终回复。若这些技能均不适用,则直接基于你的知识回答。"
)
else:
system_parts.append("重要规则: 当用户请求涉及上述技能工具的处理范围时, 你必须优先通过function calling调用对应工具, 而不是直接回复文字。调用后将获得工具执行结果, 基于结果生成最终回复。")
# 为每个skill构建function definition # 为每个skill构建function definition
for s in skills: for s in skills:
@@ -812,11 +881,26 @@ def _build_system_prompt(
if kb_context: if kb_context:
system_parts.append(f"参考以下知识库内容回答用户问题:\n{kb_context}") system_parts.append(f"参考以下知识库内容回答用户问题:\n{kb_context}")
# MCP 应用注入 # MCP 应用注入(Issue 3:让用户询问 MCP 能力时,AI 能“看到”该 MCP 的资料与地址)
if mcp: if mcp:
mcp_descs = ", ".join([f"{s['name']}({s.get('transport', 'stdio')})" for s in mcp]) mcp_lines = []
system_parts.append(f"已连接的 MCP 应用(外部工具服务): {mcp_descs}") for m in mcp:
system_parts.append("当用户请求需要调用外部系统/工具(如查询数据库、调用API、操作第三方服务)时,可参考上述 MCP 应用提供的能力;若平台已接入对应 MCP 工具,优先通过工具调用完成。") name = m.get("name", "")
desc = m.get("description", "")
transport = m.get("transport", "stdio")
endpoint = m.get("endpoint", "")
line = f"用户已选择 MCP 应用「{name}」(传输方式: {transport})"
if desc:
line += f",功能简介: {desc}"
if endpoint:
line += f",服务地址: {endpoint}"
mcp_lines.append(line)
system_parts.append("\n".join(mcp_lines))
system_parts.append(
"用户已选择上述 MCP 应用。当请求需要借助外部工具/系统(如查询数据、调用 API、操作第三方服务)时,"
"请优先参考这些 MCP 应用的能力来回答;若平台已接入对应工具,优先通过工具调用完成;"
"不要说“不了解该 MCP”。"
)
system_prompt = "\n".join(system_parts) system_prompt = "\n".join(system_parts)
return system_prompt, function_definitions return system_prompt, function_definitions
@@ -1281,6 +1365,20 @@ async def send_message(
agents = _resolve_market_agents(db, effective_agent_ids, user_id) agents = _resolve_market_agents(db, effective_agent_ids, user_id)
mcps = _resolve_market_mcp(db, effective_mcp_ids, user_id) mcps = _resolve_market_mcp(db, effective_mcp_ids, user_id)
# Issue 8: 默认技能自动发现(仅当用户未显式选择技能时启用)
# 每轮先寻找本账号已安装且平台已部署的技能,多个适用时由模型自行判断选 1 个最优。
used_auto_skills = False
if not effective_skill_ids:
try:
discovered = _discover_installed_skills(db, user_id)
except Exception as e:
logger.warning(f"自动发现已安装技能失败: {e}")
discovered = []
if discovered:
skills = discovered
used_auto_skills = True
effective_use_tools = True
# Stage B.1: Agent可能自带skill_ids和kb_ids # Stage B.1: Agent可能自带skill_ids和kb_ids
agent_extra_skill_ids = [] agent_extra_skill_ids = []
agent_extra_kb_ids = [] agent_extra_kb_ids = []
@@ -1312,6 +1410,7 @@ async def send_message(
kb_context=kb_context, kb_context=kb_context,
user_content=req.content, user_content=req.content,
mcp=mcps, mcp=mcps,
auto_discovered=used_auto_skills,
) )
# 构建消息列表 # 构建消息列表
@@ -1325,7 +1424,9 @@ async def send_message(
if h.role == "user" or h.role == "assistant": if h.role == "user" or h.role == "assistant":
messages.append({"role": h.role, "content": h.content}) messages.append({"role": h.role, "content": h.content})
messages.append({"role": "user", "content": req.content}) # Issue 1: 必须把附件内容(attachment_block)一并发给模型,
# 否则用户在对话窗上传文件后模型“看不到”文件内容。
messages.append({"role": "user", "content": req.content + attachment_block})
# Stage B.3: Function Calling循环(最多5轮) # Stage B.3: Function Calling循环(最多5轮)
max_tool_rounds = 5 max_tool_rounds = 5
+116 -14
View File
@@ -3,7 +3,15 @@ IM Bot 真实 LLM 调用。
原实现位于 channels.py(通用频道管理模块),该模块已移除。 原实现位于 channels.py(通用频道管理模块),该模块已移除。
为保持 IM Bot(微信机器人等)正常生成回复,将 _call_real_llm 迁移至此独立模块。 为保持 IM Bot(微信机器人等)正常生成回复,将 _call_real_llm 迁移至此独立模块。
行为(对齐 AI 聊天):
- Issue 7:图像/视频/语音意图优先走平台媒体模型(与 AI 聊天同一套 Auto 路由),
不再对“制作图片”这类请求只回文本。
- Issue 8:默认 Auto 模型;每轮先自动发现本账号已安装且平台已部署的技能,
多个适用时由模型自行判断选 1 个最优,通过 function calling 执行。
""" """
import asyncio
import json
from typing import Optional from typing import Optional
import httpx import httpx
@@ -13,9 +21,54 @@ from ..capability_registry import resolve_auto_model
from ..models import AIModel, ModelProvider from ..models import AIModel, ModelProvider
def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im") -> str: def _run_async(coro):
"""调用七虎AI 默认文本模型生成回复,供 IM Bot 复用。""" """在同步上下文中运行协程,兼容「无事件循环」与「已有事件循环」两种场景。
IM 个人微信在独立线程中调用(无事件循环)→ asyncio.run;
IM 多平台 Webhook 可能在 FastAPI 协程内调用(已有事件循环)→ 丢到线程池执行。
"""
try: 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> 还原为纯文本链接返回
if result.startswith("<img") or result.startswith("<video"):
import re as _re
m = _re.search(r'src="([^"]+)"', result)
url = m.group(1) if m else ""
label = "图片" if result.startswith("<img") else "视频"
return f"已为你生成{label}:{url}" if url else result
return result
# ===== 2) 文本回复 + 技能自动发现(Issue 8)=====
ai_model = resolve_auto_model("text", db) ai_model = resolve_auto_model("text", db)
if not ai_model: if not ai_model:
return "(未配置默认文本模型,无法生成回复)" return "(未配置默认文本模型,无法生成回复)"
@@ -26,20 +79,69 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im
api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"
if not (api_key and api_url): if not (api_key and api_url):
return "(模型未配置 API Key 或地址,无法生成回复)" 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 = [ messages = [
{"role": "system", "content": "你是七虎AI 助手,负责在即时通讯中回复用户消息。"}, {"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}, {"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: max_rounds = 3
resp = client.post( ai_content = ""
api_url, for _ in range(max_rounds):
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
json=request_body, if function_definitions and skills:
) request_body["tools"] = [
if resp.status_code != 200: {"type": "function", "function": fd} for fd in function_definitions
return f"(AI 模型调用失败 HTTP {resp.status_code})" ]
data = resp.json() request_body["tool_choice"] = "auto"
return data.get("choices", [{}])[0].get("message", {}).get("content", "") 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()
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: except Exception as e:
return f"(AI 回复生成失败:{e})" return f"(AI 回复生成失败:{e})"