chore: normalize line endings to LF via .gitattributes

Root-cause fix for mixed CRLF/LF in the repo (no .gitattributes previously).
Adds '* text=auto eol=lf' and explicit binary rules so all future checkouts
are LF. Renormalizes all tracked text files (no logical content change).
Backup tags: image qihuaiflow:backup-lineend-20260725-121632, git backup-lineend-pre-20260725-121632.
This commit is contained in:
QihuAI Design committed 2026-07-25 12:16:39 +08:00
1 parent 2a7bb875f9
commit 771c8456b8
39 files changed
+34810 -34777

No files matched your search

+179 -179
View File
@@ -1,179 +1,179 @@
"""
AnySearch 联网检索客户端
======================
调用 AnySearch 网页/领域搜索 API (https://api.anysearch.com/v1/search),
为 AI 聊天注入实时联网检索结果,亦可被后端端点 / Agent 工具复用。
约束:平台容器存在智能路由代理,对外部 API 偶发重置;本客户端使用
trust_env=False 直连 (等价于 NO_PROXY=api.anysearch.com),规避代理重置。
失败一律返回空/降级结果,保证聊天链路不中断。
"""
import json
import logging
from typing import List, Dict, Optional
logger = logging.getLogger("aiflow.anysearch")
DEFAULT_API_URL = "https://api.anysearch.com/v1/search"
DEFAULT_ZONE = "cn"
REQUEST_TIMEOUT = 8.0
async def anysearch_search(
api_key: str,
api_url: str = DEFAULT_API_URL,
query: str = "",
top_k: int = 5,
zone: str = DEFAULT_ZONE,
freshness: Optional[str] = None,
timeout: float = REQUEST_TIMEOUT,
) -> List[Dict]:
"""调用 AnySearch 搜索,返回结构化结果列表。
每个结果:
{
"content": snippet 摘要文本,
"score": 0.9..0.5 递减,
"source": "web_anysearch",
"metadata": {"url", "title", "source_domain", "zone"}
}
失败时返回 [] (不抛异常)。
"""
if not api_key:
return []
try:
import httpx
except ImportError:
logger.warning("httpx not available for AnySearch")
return []
payload = {
"query": query,
"zone": zone,
"top_k": max(1, min(int(top_k), 20)),
}
if freshness:
payload["freshness"] = freshness
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
}
try:
# trust_env=False -> 忽略容器内 HTTP(S)_PROXY,直连 api.anysearch.com
async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client:
resp = await client.post(api_url, json=payload, headers=headers)
if resp.status_code != 200:
logger.warning(f"AnySearch HTTP {resp.status_code}: {resp.text[:200]}")
return []
data = resp.json()
if data.get("code") != 0:
logger.warning(f"AnySearch biz error: {data.get('message')}")
return []
results = (data.get("data") or {}).get("results") or []
hits = []
for idx, r in enumerate(results[:top_k]):
title = r.get("title", "") or ""
url = r.get("url", "") or ""
snippet = r.get("snippet") or r.get("content") or ""
if not snippet and not title:
continue
try:
from urllib.parse import urlparse
domain = urlparse(url).netloc
except Exception:
domain = ""
hits.append({
"content": snippet,
"score": max(0.4, 0.9 - idx * 0.08),
"source": "web_anysearch",
"metadata": {
"url": url,
"title": title,
"source_domain": domain,
"zone": zone,
},
})
return hits
except Exception as e:
logger.warning(f"AnySearch search failed: {e}")
return []
async def anysearch_health(api_key: str, api_url: str = DEFAULT_API_URL, timeout: float = REQUEST_TIMEOUT) -> Dict:
"""连通性探测:返回 {"status": "connected"/"disconnected", "detail": ...}。"""
if not api_key:
return {"status": "disconnected", "detail": "未配置 API Key", "connected": False, "usable": False}
try:
import httpx
async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client:
resp = await client.post(
api_url,
json={"query": "ping", "zone": DEFAULT_ZONE, "top_k": 1},
headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"},
)
if resp.status_code == 200:
return {"status": "connected", "detail": "AnySearch 连接正常", "connected": True, "usable": True}
return {"status": "error", "detail": f"HTTP {resp.status_code}", "connected": False, "usable": False}
except Exception as e:
return {"status": "error", "detail": str(e)[:200], "connected": False, "usable": False}
def load_anysearch_config(db_session) -> Dict:
"""从 site_settings 读取全局 AnySearch 配置(失败返回空字典)。"""
if db_session is None:
return {}
try:
from app.models import SiteSetting
row = db_session.query(SiteSetting).filter(SiteSetting.key == "anysearch_config").first()
if not row or not row.value:
return {}
cfg = json.loads(row.value)
if not isinstance(cfg, dict):
return {}
return cfg
except Exception as e:
logger.warning(f"load_anysearch_config failed: {e}")
return {}
def save_anysearch_config(
db_session,
api_key: str,
is_enabled: bool = True,
api_url: str = DEFAULT_API_URL,
zone: str = DEFAULT_ZONE,
freshness: Optional[str] = None,
top_k: int = 5,
) -> bool:
"""保存/更新全局 AnySearch 配置到 site_settings。"""
if db_session is None:
return False
try:
from app.models import SiteSetting
from datetime import datetime
cfg = {
"api_key": api_key.strip(),
"is_enabled": bool(is_enabled),
"api_url": api_url.strip() or DEFAULT_API_URL,
"zone": zone or DEFAULT_ZONE,
"freshness": freshness,
"top_k": max(1, min(int(top_k), 20)),
}
row = db_session.query(SiteSetting).filter(SiteSetting.key == "anysearch_config").first()
if row:
row.value = json.dumps(cfg, ensure_ascii=False)
row.updated_at = datetime.now()
else:
from app.models import gen_uuid
row = SiteSetting(id=gen_uuid(), key="anysearch_config", value=json.dumps(cfg, ensure_ascii=False))
db_session.add(row)
db_session.commit()
return True
except Exception as e:
logger.warning(f"save_anysearch_config failed: {e}")
try:
db_session.rollback()
except Exception:
pass
return False
"""
AnySearch 联网检索客户端
======================
调用 AnySearch 网页/领域搜索 API (https://api.anysearch.com/v1/search),
为 AI 聊天注入实时联网检索结果,亦可被后端端点 / Agent 工具复用。
约束:平台容器存在智能路由代理,对外部 API 偶发重置;本客户端使用
trust_env=False 直连 (等价于 NO_PROXY=api.anysearch.com),规避代理重置。
失败一律返回空/降级结果,保证聊天链路不中断。
"""
import json
import logging
from typing import List, Dict, Optional
logger = logging.getLogger("aiflow.anysearch")
DEFAULT_API_URL = "https://api.anysearch.com/v1/search"
DEFAULT_ZONE = "cn"
REQUEST_TIMEOUT = 8.0
async def anysearch_search(
api_key: str,
api_url: str = DEFAULT_API_URL,
query: str = "",
top_k: int = 5,
zone: str = DEFAULT_ZONE,
freshness: Optional[str] = None,
timeout: float = REQUEST_TIMEOUT,
) -> List[Dict]:
"""调用 AnySearch 搜索,返回结构化结果列表。
每个结果:
{
"content": snippet 摘要文本,
"score": 0.9..0.5 递减,
"source": "web_anysearch",
"metadata": {"url", "title", "source_domain", "zone"}
}
失败时返回 [] (不抛异常)。
"""
if not api_key:
return []
try:
import httpx
except ImportError:
logger.warning("httpx not available for AnySearch")
return []
payload = {
"query": query,
"zone": zone,
"top_k": max(1, min(int(top_k), 20)),
}
if freshness:
payload["freshness"] = freshness
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
}
try:
# trust_env=False -> 忽略容器内 HTTP(S)_PROXY,直连 api.anysearch.com
async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client:
resp = await client.post(api_url, json=payload, headers=headers)
if resp.status_code != 200:
logger.warning(f"AnySearch HTTP {resp.status_code}: {resp.text[:200]}")
return []
data = resp.json()
if data.get("code") != 0:
logger.warning(f"AnySearch biz error: {data.get('message')}")
return []
results = (data.get("data") or {}).get("results") or []
hits = []
for idx, r in enumerate(results[:top_k]):
title = r.get("title", "") or ""
url = r.get("url", "") or ""
snippet = r.get("snippet") or r.get("content") or ""
if not snippet and not title:
continue
try:
from urllib.parse import urlparse
domain = urlparse(url).netloc
except Exception:
domain = ""
hits.append({
"content": snippet,
"score": max(0.4, 0.9 - idx * 0.08),
"source": "web_anysearch",
"metadata": {
"url": url,
"title": title,
"source_domain": domain,
"zone": zone,
},
})
return hits
except Exception as e:
logger.warning(f"AnySearch search failed: {e}")
return []
async def anysearch_health(api_key: str, api_url: str = DEFAULT_API_URL, timeout: float = REQUEST_TIMEOUT) -> Dict:
"""连通性探测:返回 {"status": "connected"/"disconnected", "detail": ...}。"""
if not api_key:
return {"status": "disconnected", "detail": "未配置 API Key", "connected": False, "usable": False}
try:
import httpx
async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client:
resp = await client.post(
api_url,
json={"query": "ping", "zone": DEFAULT_ZONE, "top_k": 1},
headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"},
)
if resp.status_code == 200:
return {"status": "connected", "detail": "AnySearch 连接正常", "connected": True, "usable": True}
return {"status": "error", "detail": f"HTTP {resp.status_code}", "connected": False, "usable": False}
except Exception as e:
return {"status": "error", "detail": str(e)[:200], "connected": False, "usable": False}
def load_anysearch_config(db_session) -> Dict:
"""从 site_settings 读取全局 AnySearch 配置(失败返回空字典)。"""
if db_session is None:
return {}
try:
from app.models import SiteSetting
row = db_session.query(SiteSetting).filter(SiteSetting.key == "anysearch_config").first()
if not row or not row.value:
return {}
cfg = json.loads(row.value)
if not isinstance(cfg, dict):
return {}
return cfg
except Exception as e:
logger.warning(f"load_anysearch_config failed: {e}")
return {}
def save_anysearch_config(
db_session,
api_key: str,
is_enabled: bool = True,
api_url: str = DEFAULT_API_URL,
zone: str = DEFAULT_ZONE,
freshness: Optional[str] = None,
top_k: int = 5,
) -> bool:
"""保存/更新全局 AnySearch 配置到 site_settings。"""
if db_session is None:
return False
try:
from app.models import SiteSetting
from datetime import datetime
cfg = {
"api_key": api_key.strip(),
"is_enabled": bool(is_enabled),
"api_url": api_url.strip() or DEFAULT_API_URL,
"zone": zone or DEFAULT_ZONE,
"freshness": freshness,
"top_k": max(1, min(int(top_k), 20)),
}
row = db_session.query(SiteSetting).filter(SiteSetting.key == "anysearch_config").first()
if row:
row.value = json.dumps(cfg, ensure_ascii=False)
row.updated_at = datetime.now()
else:
from app.models import gen_uuid
row = SiteSetting(id=gen_uuid(), key="anysearch_config", value=json.dumps(cfg, ensure_ascii=False))
db_session.add(row)
db_session.commit()
return True
except Exception as e:
logger.warning(f"save_anysearch_config failed: {e}")
try:
db_session.rollback()
except Exception:
pass
return False
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+319 -319
View File
@@ -1,319 +1,319 @@
# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
from fastapi import APIRouter, Depends, HTTPException, Query, Response
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 ..middleware.log_service import LogService
from ..capability_registry import resolve_auto_model
from app.api.chat_utils import (cst_now)
def _build_knowledge_graph(db: Session, kb: KnowledgeBase) -> dict:
"""从知识库文档构建知识图谱节点和边,返回 graph_data。"""
documents = db.query(KnowledgeDocument).filter(
and_(
KnowledgeDocument.knowledge_base_id == kb.id,
KnowledgeDocument.status == "completed",
)
).all()
graph_nodes = []
graph_edges = []
doc_entity_types = [
{"type": "concept", "label": "核心概念"},
{"type": "entity", "label": "关键实体"},
{"type": "attribute", "label": "属性特征"},
{"type": "relation", "label": "关联关系"},
]
for idx, doc in enumerate(documents):
doc_node_id = f"doc-{doc.id[:8]}"
graph_nodes.append({
"id": doc_node_id, "label": doc.filename, "type": "document", "size": 18, "group": "document",
})
filename_base = doc.filename.rsplit(".", 1)[0]
words = filename_base.replace("_", " ").replace("-", " ").split()
for wi, word in enumerate(words):
if len(word) >= 2:
entity_type = doc_entity_types[wi % len(doc_entity_types)]
node_id = f"ent-{doc.id[:8]}-{wi}"
graph_nodes.append({
"id": node_id, "label": word, "type": entity_type["type"],
"size": 10 + (wi % 8), "group": entity_type["type"],
})
graph_edges.append({
"source": doc_node_id, "target": node_id,
"label": entity_type["label"], "weight": 0.5 + (0.5 / (wi + 1)),
})
for i in range(len(documents)):
for j in range(i + 1, len(documents)):
fi = documents[i].filename.lower().rsplit(".", 1)[0]
fj = documents[j].filename.lower().rsplit(".", 1)[0]
words_i = set(fi.replace("_", " ").replace("-", " ").split())
words_j = set(fj.replace("_", " ").replace("-", " ").split())
common = words_i & words_j
if common and len(common) > 0:
graph_edges.append({
"source": f"doc-{documents[i].id[:8]}",
"target": f"doc-{documents[j].id[:8]}",
"label": f"共有关键词: {', '.join(list(common)[:3])}",
"weight": min(0.9, 0.3 * len(common)),
})
graph_data = {
"nodes": graph_nodes,
"edges": graph_edges,
"stats": {
"total_nodes": len(graph_nodes), "total_edges": len(graph_edges),
"document_count": len(documents),
},
}
kb.knowledge_graph_count = len(graph_nodes)
kb.updated_at = cst_now()
return graph_data
# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
from fastapi import APIRouter, Depends, HTTPException, Query, Response
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 ..middleware.log_service import LogService
from ..capability_registry import resolve_auto_model
from app.api.chat_utils import (cst_now)
def _build_knowledge_graph(db: Session, kb: KnowledgeBase) -> dict:
"""从知识库文档构建知识图谱节点和边,返回 graph_data。"""
documents = db.query(KnowledgeDocument).filter(
and_(
KnowledgeDocument.knowledge_base_id == kb.id,
KnowledgeDocument.status == "completed",
)
).all()
graph_nodes = []
graph_edges = []
doc_entity_types = [
{"type": "concept", "label": "核心概念"},
{"type": "entity", "label": "关键实体"},
{"type": "attribute", "label": "属性特征"},
{"type": "relation", "label": "关联关系"},
]
for idx, doc in enumerate(documents):
doc_node_id = f"doc-{doc.id[:8]}"
graph_nodes.append({
"id": doc_node_id, "label": doc.filename, "type": "document", "size": 18, "group": "document",
})
filename_base = doc.filename.rsplit(".", 1)[0]
words = filename_base.replace("_", " ").replace("-", " ").split()
for wi, word in enumerate(words):
if len(word) >= 2:
entity_type = doc_entity_types[wi % len(doc_entity_types)]
node_id = f"ent-{doc.id[:8]}-{wi}"
graph_nodes.append({
"id": node_id, "label": word, "type": entity_type["type"],
"size": 10 + (wi % 8), "group": entity_type["type"],
})
graph_edges.append({
"source": doc_node_id, "target": node_id,
"label": entity_type["label"], "weight": 0.5 + (0.5 / (wi + 1)),
})
for i in range(len(documents)):
for j in range(i + 1, len(documents)):
fi = documents[i].filename.lower().rsplit(".", 1)[0]
fj = documents[j].filename.lower().rsplit(".", 1)[0]
words_i = set(fi.replace("_", " ").replace("-", " ").split())
words_j = set(fj.replace("_", " ").replace("-", " ").split())
common = words_i & words_j
if common and len(common) > 0:
graph_edges.append({
"source": f"doc-{documents[i].id[:8]}",
"target": f"doc-{documents[j].id[:8]}",
"label": f"共有关键词: {', '.join(list(common)[:3])}",
"weight": min(0.9, 0.3 * len(common)),
})
graph_data = {
"nodes": graph_nodes,
"edges": graph_edges,
"stats": {
"total_nodes": len(graph_nodes), "total_edges": len(graph_edges),
"document_count": len(documents),
},
}
kb.knowledge_graph_count = len(graph_nodes)
kb.updated_at = cst_now()
return graph_data
+240 -240
View File
@@ -1,240 +1,240 @@
# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
from fastapi import APIRouter, Depends, HTTPException, Query, Response
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 ..middleware.log_service import LogService
from ..capability_registry import resolve_auto_model
def _resolve_market_mcp(db: Session, mcp_ids: list, user_id: str) -> list:
"""
加载用户选择的 MCP 应用配置(MarketApp.type == 'mcp')。
返回: [{id, name, transport, endpoint, description}]
"""
if not mcp_ids:
return []
mcps = db.query(MarketApp).filter(
MarketApp.id.in_(mcp_ids),
MarketApp.type == "mcp"
).all()
results = []
for m in mcps:
config = {}
try:
if m.skill_content:
config = json.loads(m.skill_content)
except Exception:
config = {}
results.append({
"id": m.id,
"name": m.name,
"transport": m.mcp_transport or config.get("transport", "stdio"),
"endpoint": m.agent_endpoint or config.get("endpoint", "") or config.get("url", ""),
"description": m.description or config.get("description", ""),
})
return results
def _resolve_active_mcp_servers(db: Session, mcp_ids: list) -> list:
"""
解析用户选中的 MCP 服务连接信息(来自 mcp_servers 表,import 与后台添加通用)。
返回: [{id, name, description, transport, endpoint, url, headers}]
"""
if not mcp_ids:
return []
servers = db.query(MCPServer).filter(
MCPServer.id.in_(mcp_ids),
MCPServer.is_enabled == True,
).all()
out = []
for s in servers:
try:
headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
except Exception:
headers = {}
out.append({
"id": s.id,
"name": s.name,
"description": s.description or "",
"transport": s.transport_type or "streamable_http",
"endpoint": s.url or "",
"url": s.url or "",
"headers": headers,
})
return out
def _list_enabled_mcp_servers(db: Session) -> list:
"""返回平台所有已启用(is_enabled=True)的 MCP 服务清单,供 AI 准确回答"平台有哪些 MCP"。"""
servers = db.query(MCPServer).filter(MCPServer.is_enabled == True).all()
out = []
for s in servers:
try:
headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
except Exception:
headers = {}
out.append({
"id": s.id,
"name": s.name,
"description": s.description or "",
"transport": s.transport_type or "streamable_http",
"endpoint": s.url or "",
"url": s.url or "",
"headers": headers,
})
return out
# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
from fastapi import APIRouter, Depends, HTTPException, Query, Response
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 ..middleware.log_service import LogService
from ..capability_registry import resolve_auto_model
def _resolve_market_mcp(db: Session, mcp_ids: list, user_id: str) -> list:
"""
加载用户选择的 MCP 应用配置(MarketApp.type == 'mcp')。
返回: [{id, name, transport, endpoint, description}]
"""
if not mcp_ids:
return []
mcps = db.query(MarketApp).filter(
MarketApp.id.in_(mcp_ids),
MarketApp.type == "mcp"
).all()
results = []
for m in mcps:
config = {}
try:
if m.skill_content:
config = json.loads(m.skill_content)
except Exception:
config = {}
results.append({
"id": m.id,
"name": m.name,
"transport": m.mcp_transport or config.get("transport", "stdio"),
"endpoint": m.agent_endpoint or config.get("endpoint", "") or config.get("url", ""),
"description": m.description or config.get("description", ""),
})
return results
def _resolve_active_mcp_servers(db: Session, mcp_ids: list) -> list:
"""
解析用户选中的 MCP 服务连接信息(来自 mcp_servers 表,import 与后台添加通用)。
返回: [{id, name, description, transport, endpoint, url, headers}]
"""
if not mcp_ids:
return []
servers = db.query(MCPServer).filter(
MCPServer.id.in_(mcp_ids),
MCPServer.is_enabled == True,
).all()
out = []
for s in servers:
try:
headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
except Exception:
headers = {}
out.append({
"id": s.id,
"name": s.name,
"description": s.description or "",
"transport": s.transport_type or "streamable_http",
"endpoint": s.url or "",
"url": s.url or "",
"headers": headers,
})
return out
def _list_enabled_mcp_servers(db: Session) -> list:
"""返回平台所有已启用(is_enabled=True)的 MCP 服务清单,供 AI 准确回答"平台有哪些 MCP"。"""
servers = db.query(MCPServer).filter(MCPServer.is_enabled == True).all()
out = []
for s in servers:
try:
headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
except Exception:
headers = {}
out.append({
"id": s.id,
"name": s.name,
"description": s.description or "",
"transport": s.transport_type or "streamable_http",
"endpoint": s.url or "",
"url": s.url or "",
"headers": headers,
})
return out
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+282 -282
View File
@@ -1,282 +1,282 @@
"""
IM Bot 真实 LLM 调用。
原实现位于 channels.py(通用频道管理模块),该模块已移除。
为保持 IM Bot(微信机器人等)正常生成回复,将 _call_real_llm 迁移至此独立模块。
行为(对齐 AI 聊天 — 2026-07-23 同步至最新):
- Issue 7:图像/视频/语音意图优先走平台媒体模型(与 AI 聊天同一套 Auto 路由),
不再对"制作图片"这类请求只回文本。
- Issue 8:默认 Auto 模型;每轮先自动发现本账号已安装且平台已部署的技能,
多个适用时由模型自行判断选 1 个最优,通过 function calling 执行。
- 2026-07-15:修复 trust_env=False + proxy 处理,避免容器坏代理导致调用失败;
媒体生成前先经 _optimize_media_prompt 精炼提示词,提升 AGNES 生成质量。
- 2026-07-23:对齐 Web Chat 最新能力:
* 模型 failover(Auto 模式自动切换备选模型)
* 平台知识库注入(get_platform_knowledge,技能清单实时同步)
"""
import asyncio
import json
import logging
import re as _re
from datetime import datetime, timezone, timedelta
from typing import Optional
from urllib.parse import urlparse as _urlparse
import httpx
from app.core.llm_client import chat_complete # 统一模型层(07-24): web/IM 共用标准 OpenAI SDK 客户端
from sqlalchemy.orm import Session
from ..capability_registry import resolve_auto_model
from ..models import AIModel, ModelProvider, TokenUsageLog, gen_uuid
from app.core.observability import get_trace_id
logger = logging.getLogger(__name__)
# 香港出网代理(实测对 AGNES 延迟更低,2026-07-15 用户实测),与 chat.py 同值。
_HK_PROXY = "http://qyyfw:XHR521yhf@103.242.14.208:11495"
def _proxy_for_url(url: str) -> str | None:
"""按目标域名选择是否走代理(与 chat.py 同逻辑)。
2026-07-15 调整:AGNES 改走香港代理(实测延迟更低),不再直连。
国内域名(尊重 NO_PROXY)直连,更稳定。
"""
import os
host = (_urlparse(url).hostname or "").lower()
if "agnes-ai.com" in host or "agnes-ai.space" in host:
return _HK_PROXY
if "qihuyun.com" in host:
return None
_no_proxy = [h.strip().lower() for h in os.environ.get("NO_PROXY", "").split(",") if h.strip()]
for _np in _no_proxy:
if host == _np or host.endswith("." + _np) or _np 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, timeout=None):
"""在同步上下文中运行协程,兼容「无事件循环」与「已有事件循环」两种场景。
IM 个人微信在独立线程中调用(无事件循环)→ asyncio.run;
IM 多平台 Webhook 可能在 FastAPI 协程内调用(已有事件循环)→ 丢到线程池执行。
timeout: 可选,秒数;超时抛出 asyncio.TimeoutError(由调用方捕获)。
"""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
if timeout is not None:
async def _timed():
return await asyncio.wait_for(coro, timeout=timeout)
target = _timed()
else:
target = coro
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(target)).result(timeout=(timeout + 10) if timeout else None)
return asyncio.run(target)
def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, prompt: str):
"""IM Bot 媒体生成记账:把图片/视频消耗写入 TokenUsageLog,供驾驶舱统计。
此前 IM Bot 路径完全不写 TokenUsageLog,导致微信生成的图片/视频在驾驶舱里"查无此项"。
媒体 API 不返回 token 用量,这里用与 chat.py _generate_media_task 一致的估算。
"""
if not user_id or not ai_model:
return
try:
base = 1200 if media_type == "video" else 400
est = base + (len(prompt or "") // 4)
db.add(TokenUsageLog(
id=gen_uuid(),
user_id=user_id,
model_id=ai_model.id,
model_name=ai_model.name,
trace_id=get_trace_id(),
prompt_tokens=est,
completion_tokens=0,
total_tokens=est,
created_at=datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None),
))
db.commit()
except Exception as e:
try:
db.rollback()
except Exception:
pass
logger.warning(f"[im_llm] 媒体 Token 记账失败(忽略): {e}")
def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im", user_id: str = None) -> "str | dict":
"""调用七虎AI 生成回复,供 IM Bot 复用。支持媒体意图路由、已安装技能自动发现、模型 failover。"""
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), timeout=300)
except Exception:
# 任何未预期异常(如 AGNES 连接重置)统一兜底为友好提示,
# 绝不直接把原始异常(如 "Server disconnected")甩给用户
result = "⚠️ 媒体生成暂时失败(连接或服务波动),请稍后重试。"
if result:
# 媒体生成成功:返回结构化对象,交由 im_bot_user 在微信内联发送图片/视频
s = result.strip()
if s.startswith("<img") or s.startswith("<video"):
m = _re.search(r'src="([^"]+)"', s)
url = m.group(1) if m else ""
media_type = "image" if s.startswith("<img") else "video"
if not url:
return "(生成完成,但未能提取媒体地址)"
# IM Bot 路径此前完全不记账 → 补写媒体消耗,驾驶舱才看得到图片/视频数据
_log_media_usage(db, user_id, ai_model, media_type, prompt)
return {"__media__": True, "media_type": media_type, "url": url}
# 媒体生成失败:_generate_media 已返回友好中文(如限频/队列繁忙),
# 直接透传,绝不把原始提示词/标签/JSON 透传给用户
if s.startswith("⚠️") or s.startswith("(媒体"):
return s
return "(媒体生成未能返回有效内容,请重试)"
# ===== 2) 文本回复:模型 failover + 技能自动发现 + 平台知识库注入 =====
from ..platform_knowledge import get_platform_knowledge
# 构建候选模型列表(对齐 Web Chat failover)
ai_model = resolve_auto_model("text", db)
_try_models = [ai_model] if (ai_model and ai_model.is_enabled) else []
# Auto 模式附加备选(IM Bot 始终 Auto)
_fallbacks = db.query(AIModel).filter(
AIModel.is_enabled == True,
AIModel.id != (ai_model.id if ai_model else ""),
).order_by(AIModel.is_default.desc()).limit(4).all()
_try_models.extend([m for m in _fallbacks if m not in _try_models])
if not _try_models:
return "(未配置可用文本模型,无法生成回复)"
# 技能自动发现(提前执行一次,不必每个候选模型重复做)
skills = []
if user_id:
try:
skills = _discover_installed_skills(db, user_id)[:15]
except Exception:
skills = []
# 平台知识库注入(对齐 Web Chat——微信用户提问"平台有哪些技能"也能得到完整清单)
kb_context = ""
try:
pk = get_platform_knowledge(db=db)
if pk:
kb_context = f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}"
except Exception:
pass
# 尝试每个候选模型
_retry_log = []
for _try_model in _try_models:
ai_model = _try_model
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):
_retry_log.append(f"{ai_model.name}: 缺少密钥或端点")
continue
system_prompt, function_definitions = _build_system_prompt(
skills=skills,
agents=[],
kb_context=kb_context,
user_content=prompt,
mcp=[],
auto_discovered=bool(skills),
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt},
]
# 尝试调用
try:
max_rounds = 3
ai_content = ""
_http_ok = True
for round_num 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"
try:
data = chat_complete(
api_url,
api_key,
request_body,
proxy=_proxy_for_url(api_url),
timeout=ai_model.timeout or 120,
)
except Exception as _exc:
_retry_log.append(f"{ai_model.name}: {str(_exc)[:60]}")
logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {_exc}")
_http_ok = False
break
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
# 检查本轮调用是否成功
if _http_ok and ai_content:
return ai_content or "(AI 回复为空)"
# 本轮失败:记录日志,切换备选
logger.warning(f"[im_llm] 模型 {ai_model.name} 失败,切换备选...")
continue
except Exception as exc:
_retry_log.append(f"{ai_model.name}: {str(exc)[:60]}")
logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {exc}")
continue
# 全部失败
_summary = "; ".join(_retry_log[-3:])
return f"(已尝试 {len(_try_models)} 个模型均调用失败。最后错误: {_summary}。请稍后重试。)"
except Exception as e:
return f"(AI 回复生成失败:{e})"
"""
IM Bot 真实 LLM 调用。
原实现位于 channels.py(通用频道管理模块),该模块已移除。
为保持 IM Bot(微信机器人等)正常生成回复,将 _call_real_llm 迁移至此独立模块。
行为(对齐 AI 聊天 — 2026-07-23 同步至最新):
- Issue 7:图像/视频/语音意图优先走平台媒体模型(与 AI 聊天同一套 Auto 路由),
不再对"制作图片"这类请求只回文本。
- Issue 8:默认 Auto 模型;每轮先自动发现本账号已安装且平台已部署的技能,
多个适用时由模型自行判断选 1 个最优,通过 function calling 执行。
- 2026-07-15:修复 trust_env=False + proxy 处理,避免容器坏代理导致调用失败;
媒体生成前先经 _optimize_media_prompt 精炼提示词,提升 AGNES 生成质量。
- 2026-07-23:对齐 Web Chat 最新能力:
* 模型 failover(Auto 模式自动切换备选模型)
* 平台知识库注入(get_platform_knowledge,技能清单实时同步)
"""
import asyncio
import json
import logging
import re as _re
from datetime import datetime, timezone, timedelta
from typing import Optional
from urllib.parse import urlparse as _urlparse
import httpx
from app.core.llm_client import chat_complete # 统一模型层(07-24): web/IM 共用标准 OpenAI SDK 客户端
from sqlalchemy.orm import Session
from ..capability_registry import resolve_auto_model
from ..models import AIModel, ModelProvider, TokenUsageLog, gen_uuid
from app.core.observability import get_trace_id
logger = logging.getLogger(__name__)
# 香港出网代理(实测对 AGNES 延迟更低,2026-07-15 用户实测),与 chat.py 同值。
_HK_PROXY = "http://qyyfw:XHR521yhf@103.242.14.208:11495"
def _proxy_for_url(url: str) -> str | None:
"""按目标域名选择是否走代理(与 chat.py 同逻辑)。
2026-07-15 调整:AGNES 改走香港代理(实测延迟更低),不再直连。
国内域名(尊重 NO_PROXY)直连,更稳定。
"""
import os
host = (_urlparse(url).hostname or "").lower()
if "agnes-ai.com" in host or "agnes-ai.space" in host:
return _HK_PROXY
if "qihuyun.com" in host:
return None
_no_proxy = [h.strip().lower() for h in os.environ.get("NO_PROXY", "").split(",") if h.strip()]
for _np in _no_proxy:
if host == _np or host.endswith("." + _np) or _np 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, timeout=None):
"""在同步上下文中运行协程,兼容「无事件循环」与「已有事件循环」两种场景。
IM 个人微信在独立线程中调用(无事件循环)→ asyncio.run;
IM 多平台 Webhook 可能在 FastAPI 协程内调用(已有事件循环)→ 丢到线程池执行。
timeout: 可选,秒数;超时抛出 asyncio.TimeoutError(由调用方捕获)。
"""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
if timeout is not None:
async def _timed():
return await asyncio.wait_for(coro, timeout=timeout)
target = _timed()
else:
target = coro
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(target)).result(timeout=(timeout + 10) if timeout else None)
return asyncio.run(target)
def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, prompt: str):
"""IM Bot 媒体生成记账:把图片/视频消耗写入 TokenUsageLog,供驾驶舱统计。
此前 IM Bot 路径完全不写 TokenUsageLog,导致微信生成的图片/视频在驾驶舱里"查无此项"。
媒体 API 不返回 token 用量,这里用与 chat.py _generate_media_task 一致的估算。
"""
if not user_id or not ai_model:
return
try:
base = 1200 if media_type == "video" else 400
est = base + (len(prompt or "") // 4)
db.add(TokenUsageLog(
id=gen_uuid(),
user_id=user_id,
model_id=ai_model.id,
model_name=ai_model.name,
trace_id=get_trace_id(),
prompt_tokens=est,
completion_tokens=0,
total_tokens=est,
created_at=datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None),
))
db.commit()
except Exception as e:
try:
db.rollback()
except Exception:
pass
logger.warning(f"[im_llm] 媒体 Token 记账失败(忽略): {e}")
def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im", user_id: str = None) -> "str | dict":
"""调用七虎AI 生成回复,供 IM Bot 复用。支持媒体意图路由、已安装技能自动发现、模型 failover。"""
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), timeout=300)
except Exception:
# 任何未预期异常(如 AGNES 连接重置)统一兜底为友好提示,
# 绝不直接把原始异常(如 "Server disconnected")甩给用户
result = "⚠️ 媒体生成暂时失败(连接或服务波动),请稍后重试。"
if result:
# 媒体生成成功:返回结构化对象,交由 im_bot_user 在微信内联发送图片/视频
s = result.strip()
if s.startswith("<img") or s.startswith("<video"):
m = _re.search(r'src="([^"]+)"', s)
url = m.group(1) if m else ""
media_type = "image" if s.startswith("<img") else "video"
if not url:
return "(生成完成,但未能提取媒体地址)"
# IM Bot 路径此前完全不记账 → 补写媒体消耗,驾驶舱才看得到图片/视频数据
_log_media_usage(db, user_id, ai_model, media_type, prompt)
return {"__media__": True, "media_type": media_type, "url": url}
# 媒体生成失败:_generate_media 已返回友好中文(如限频/队列繁忙),
# 直接透传,绝不把原始提示词/标签/JSON 透传给用户
if s.startswith("⚠️") or s.startswith("(媒体"):
return s
return "(媒体生成未能返回有效内容,请重试)"
# ===== 2) 文本回复:模型 failover + 技能自动发现 + 平台知识库注入 =====
from ..platform_knowledge import get_platform_knowledge
# 构建候选模型列表(对齐 Web Chat failover)
ai_model = resolve_auto_model("text", db)
_try_models = [ai_model] if (ai_model and ai_model.is_enabled) else []
# Auto 模式附加备选(IM Bot 始终 Auto)
_fallbacks = db.query(AIModel).filter(
AIModel.is_enabled == True,
AIModel.id != (ai_model.id if ai_model else ""),
).order_by(AIModel.is_default.desc()).limit(4).all()
_try_models.extend([m for m in _fallbacks if m not in _try_models])
if not _try_models:
return "(未配置可用文本模型,无法生成回复)"
# 技能自动发现(提前执行一次,不必每个候选模型重复做)
skills = []
if user_id:
try:
skills = _discover_installed_skills(db, user_id)[:15]
except Exception:
skills = []
# 平台知识库注入(对齐 Web Chat——微信用户提问"平台有哪些技能"也能得到完整清单)
kb_context = ""
try:
pk = get_platform_knowledge(db=db)
if pk:
kb_context = f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}"
except Exception:
pass
# 尝试每个候选模型
_retry_log = []
for _try_model in _try_models:
ai_model = _try_model
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):
_retry_log.append(f"{ai_model.name}: 缺少密钥或端点")
continue
system_prompt, function_definitions = _build_system_prompt(
skills=skills,
agents=[],
kb_context=kb_context,
user_content=prompt,
mcp=[],
auto_discovered=bool(skills),
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt},
]
# 尝试调用
try:
max_rounds = 3
ai_content = ""
_http_ok = True
for round_num 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"
try:
data = chat_complete(
api_url,
api_key,
request_body,
proxy=_proxy_for_url(api_url),
timeout=ai_model.timeout or 120,
)
except Exception as _exc:
_retry_log.append(f"{ai_model.name}: {str(_exc)[:60]}")
logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {_exc}")
_http_ok = False
break
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
# 检查本轮调用是否成功
if _http_ok and ai_content:
return ai_content or "(AI 回复为空)"
# 本轮失败:记录日志,切换备选
logger.warning(f"[im_llm] 模型 {ai_model.name} 失败,切换备选...")
continue
except Exception as exc:
_retry_log.append(f"{ai_model.name}: {str(exc)[:60]}")
logger.warning(f"[im_llm] 模型 {ai_model.name} 异常: {exc}")
continue
# 全部失败
_summary = "; ".join(_retry_log[-3:])
return f"(已尝试 {len(_try_models)} 个模型均调用失败。最后错误: {_summary}。请稍后重试。)"
except Exception as e:
return f"(AI 回复生成失败:{e})"
File diff suppressed because it is too large. Load diff
+395 -395
View File
@@ -1,395 +1,395 @@
"""app/core/orchestrator.py — Phase 3.3 多 Agent 编排图引擎。
设计目标:
- 以轻量 DAG 描述多 Agent 协作:节点(llm/local/external/reduce) + 有向边(数据流)。
- 拓扑分层后按层并行执行(asyncio.gather),互相独立的节点真正并发。
- 失败隔离:每个节点异常不影响其它分支;per-node on_error=halt|continue。
- 节点执行复用既有原语(_build_system_prompt / _execute_skill_tool / _call_external_agent /
3.1 工具沙箱),完全向后兼容——agent.orchestration 为 null 时 chat_service 走原逻辑。
- 并发安全:每个节点用独立 SessionLocal 会话做 DB 读,避免共享请求会话跨协程竞态。
"""
from __future__ import annotations
import asyncio
import json
import logging
from typing import Any, Optional
import httpx
logger = logging.getLogger("aiflow.orchestrator")
VALID_NODE_TYPES = ("llm", "local", "external", "reduce")
class OrchestrationError(ValueError):
"""图结构非法或执行被 halt。"""
# ---------------------------------------------------------------------------
# 校验 + 拓扑
# ---------------------------------------------------------------------------
def validate_graph(graph: Any) -> None:
if not isinstance(graph, dict):
raise OrchestrationError("orchestration 必须为对象")
nodes = graph.get("nodes")
edges = graph.get("edges")
if not isinstance(nodes, list) or not nodes:
raise OrchestrationError("nodes 必须为非空数组")
if not isinstance(edges, list):
raise OrchestrationError("edges 必须为数组")
ids = set()
for i, n in enumerate(nodes):
if not isinstance(n, dict):
raise OrchestrationError(f"节点 #{i} 必须为对象")
nid = n.get("id")
if not nid or not isinstance(nid, str):
raise OrchestrationError(f"节点 #{i} 缺少字符串 id")
if nid in ids:
raise OrchestrationError(f"节点 id 重复: {nid}")
ids.add(nid)
ntype = n.get("type")
if ntype not in VALID_NODE_TYPES:
raise OrchestrationError(
f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(VALID_NODE_TYPES)})"
)
if ntype in ("local", "external") and not n.get("agent_id"):
raise OrchestrationError(f"节点 {nid} 类型 {ntype} 必须指定 agent_id")
for i, e in enumerate(edges):
if not isinstance(e, dict):
raise OrchestrationError(f"边 #{i} 必须为对象")
f = e.get("from")
t = e.get("to")
if f not in ids:
raise OrchestrationError(f"边 #{i} 的 from={f} 指向不存在的节点")
if t not in ids:
raise OrchestrationError(f"边 #{i} 的 to={t} 指向不存在的节点")
if f == t:
raise OrchestrationError(f"边 #{i} 存在自环: {f}")
def topological_layers(graph: dict) -> list:
"""Kahn 算法返回分层(每层内节点可并行);含环则抛错。"""
nodes = graph["nodes"]
edges = graph.get("edges", [])
indeg = {n["id"]: 0 for n in nodes}
adj = {n["id"]: [] for n in nodes}
for e in edges:
f, t = e["from"], e["to"]
adj[f].append(t)
indeg[t] += 1
order = {n["id"]: i for i, n in enumerate(nodes)}
layers: list = []
remaining = set(indeg)
while remaining:
layer = sorted([nid for nid in remaining if indeg[nid] == 0], key=lambda x: order[x])
if not layer:
cycle = sorted(remaining, key=lambda x: order[x])
raise OrchestrationError(f"检测到环路,无法拓扑排序: {cycle}")
layers.append(layer)
for nid in layer:
remaining.discard(nid)
for nxt in adj[nid]:
indeg[nxt] -= 1
return layers
def _node_by_id(graph: dict, nid: str) -> dict:
for n in graph["nodes"]:
if n["id"] == nid:
return n
return {}
def _predecessors(graph: dict, nid: str) -> list:
return [e["from"] for e in graph.get("edges", []) if e["to"] == nid]
def _successors(graph: dict, nid: str) -> list:
return [e["to"] for e in graph.get("edges", []) if e["from"] == nid]
def _find_output_node(graph: dict) -> str:
if graph.get("output_node"):
return graph["output_node"]
sinks = [n["id"] for n in graph["nodes"] if not _successors(graph, n["id"])]
if len(sinks) == 1:
return sinks[0]
for s in sinks:
if _node_by_id(graph, s).get("type") == "reduce":
return s
if sinks:
return sinks[0]
return graph["nodes"][-1]["id"]
def _assemble_node_input(query: str, upstream: str, is_entry: bool) -> str:
"""入口节点直接吃用户原问题;下游节点综合原问题 + 上游输出。"""
if is_entry:
return query or ""
if upstream:
return f"原始用户问题:\n{query}\n\n上游节点输出:\n{upstream}"
return query or ""
# ---------------------------------------------------------------------------
# 模型解析 + LLM 调用原语
# ---------------------------------------------------------------------------
async def _resolve_model_and_endpoint(model_id, db):
from app.models import AIModel, ModelProvider
from ..capability_registry import resolve_auto_model
ai_model = None
if model_id and model_id != "auto":
ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()
if not ai_model:
ai_model = resolve_auto_model("text", db)
if not ai_model:
raise OrchestrationError("无法解析到可用模型")
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_url or not api_key:
raise OrchestrationError(f"节点模型 {ai_model.name} 缺少 endpoint 或 api_key")
return ai_model, api_url, api_key
async def _llm_complete(system_prompt: str, user_content: str, model_id, db, user_id) -> str:
from app.api.chat_utils import _proxy_for_url
from app.core.observability import span
ai_model, api_url, api_key = await _resolve_model_and_endpoint(model_id, db)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_content},
]
request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
async with span("orchestrator.llm", model=ai_model.id):
async with httpx.AsyncClient(
timeout=ai_model.timeout or 120,
proxy=_proxy_for_url(api_url),
trust_env=False,
) as client:
resp = await client.post(
api_url,
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json=request_body,
)
if resp.status_code != 200:
raise OrchestrationError(f"LLM 调用失败 HTTP {resp.status_code}: {resp.text[:200]}")
data = resp.json()
return data["choices"][0]["message"].get("content") or ""
# ---------------------------------------------------------------------------
# local 节点:复用单 Agent system prompt + 受限 tool round + 3.1 沙箱
# ---------------------------------------------------------------------------
async def _run_local_agent_turn(agent: dict, node_input: str, db, user_id, conv_id) -> dict:
from app.api.chat_skill import _resolve_market_skills, _execute_skill_tool
from app.api.chat_context import _build_system_prompt, _retrieve_kb_context
from app.api.chat_service import _resolve_agent_tool_policy, _check_tool_policy
from app.api.chat_utils import _proxy_for_url
skills = _resolve_market_skills(db, agent.get("skill_ids") or [], user_id)
kb_context = await _retrieve_kb_context(db, agent.get("knowledge_base_ids") or [], node_input)
system_prompt, function_definitions = _build_system_prompt(
skills=skills,
agents=[agent],
kb_context=kb_context,
user_content=node_input,
auto_discovered=False,
)
ai_model, api_url, api_key = await _resolve_model_and_endpoint(agent.get("model_id"), db)
tool_policy = _resolve_agent_tool_policy([agent])
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": node_input},
]
called_skills: list = []
tool_call_count = 0
max_tool_rounds = 5
last_content = ""
async with httpx.AsyncClient(
timeout=ai_model.timeout or 120,
proxy=_proxy_for_url(api_url),
trust_env=False,
) as client:
for round_num in range(max_tool_rounds):
request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
if function_definitions:
request_body["tools"] = [
{"type": "function", "function": fd} for fd in function_definitions
]
request_body["tool_choice"] = "auto"
resp = await client.post(
api_url,
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json=request_body,
)
if resp.status_code != 200:
raise OrchestrationError(f"local 节点 LLM 失败 HTTP {resp.status_code}: {resp.text[:200]}")
data = resp.json()
message = data["choices"][0]["message"]
tool_calls = message.get("tool_calls") or []
content = message.get("content") or ""
last_content = content
if not tool_calls or not function_definitions:
return {"text": content, "used_tools": called_skills}
messages.append({"role": "assistant", "content": content, "tool_calls": tool_calls})
round_tool_count = 0
for tc in tool_calls:
func_name = tc.get("function", {}).get("name", "")
check = _check_tool_policy(
func_name, None, skills, tool_policy, tool_call_count, round_tool_count
)
if not check["allowed"]:
messages.append({
"role": "tool",
"tool_call_id": tc.get("id", f"call_{round_num}"),
"content": f"[工具调用已被策略拦截] {check.get('message', check.get('reason'))}",
})
tool_call_count += 1
round_tool_count += 1
continue
tool_call_count += 1
round_tool_count += 1
try:
func_args = json.loads(tc.get("function", {}).get("arguments", "{}") or "{}")
except Exception:
func_args = {}
res = _execute_skill_tool(db, func_name, func_args, skills, node_input)
called_skills.append(func_name)
messages.append({
"role": "tool",
"tool_call_id": tc.get("id", f"call_{round_num}"),
"content": res,
})
return {"text": last_content, "used_tools": called_skills}
# ---------------------------------------------------------------------------
# 单节点执行(含独立 DB 会话,保证并发安全)
# ---------------------------------------------------------------------------
async def _run_node(node, graph, blackboard, query, user_id, conv_id, trace_id) -> dict:
from app.api.chat_skill import _resolve_market_agents
from app.database import SessionLocal
nid = node["id"]
ntype = node["type"]
preds = _predecessors(graph, nid)
pred_results = [blackboard[p] for p in preds]
is_entry = not preds
if pred_results:
parts = []
for b in pred_results:
label = f"节点 {b.get('node_id')}/{b.get('node_type')}"
if b.get("ok"):
parts.append(f"【{label}】\n{b.get('output', '')}")
else:
parts.append(f"【{label} 执行失败】\n错误: {b.get('error', '')}")
upstream = "\n\n".join(parts)
else:
upstream = ""
with SessionLocal() as db:
if ntype in ("llm", "reduce"):
if ntype == "llm":
sys_prompt = node.get("system_prompt") or (
"你是一个处理节点,请根据上游输入完成指定的文本变换、摘要或路由判断,"
"只输出处理后的结果,不要多余解释。"
)
else:
sys_prompt = node.get("system_prompt") or (
"你是一个结果汇总器,请将多个子任务的输出综合成连贯、完整、面向用户的最终回答。"
)
uc = _assemble_node_input(query, upstream, is_entry)
text = await _llm_complete(sys_prompt, uc, node.get("model_id"), db, user_id)
return {"ok": True, "node_id": nid, "node_type": ntype, "output": text}
if ntype == "local":
agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
if not agents:
raise OrchestrationError(f"local 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
ni = _assemble_node_input(query, upstream, is_entry)
res = await _run_local_agent_turn(agents[0], ni, db, user_id, conv_id)
return {
"ok": True,
"node_id": nid,
"node_type": "local",
"output": res["text"],
"used_tools": res.get("used_tools"),
}
if ntype == "external":
from app.api.chat_agents import _call_external_agent
agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
if not agents:
raise OrchestrationError(f"external 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
ni = _assemble_node_input(query, upstream, is_entry)
answer, _ = await _call_external_agent(
db, agents[0], query=ni, user_id=user_id, ext_conversation_id=None, history=[]
)
return {"ok": True, "node_id": nid, "node_type": "external", "output": answer or ""}
raise OrchestrationError(f"未知节点类型 {ntype}")
# ---------------------------------------------------------------------------
# 入口
# ---------------------------------------------------------------------------
async def run_graph(graph, query, db, user_id, conv_id, trace_id=None):
"""执行编排图,返回 (answer: str, trace: dict)。
- 拓扑分层后逐层 asyncio.gather 并行执行。
- 节点异常按 on_error=halt|continue 处理(默认 continue,失败时下游拿到错误作为输入)。
"""
validate_graph(graph)
layers = topological_layers(graph)
blackboard: dict = {}
trace = {"nodes": [], "trace_id": trace_id, "layers": layers}
for layer in layers:
tasks = [
_run_node(_node_by_id(graph, nid), graph, blackboard, query, user_id, conv_id, trace_id)
for nid in layer
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for nid, res in zip(layer, results):
node = _node_by_id(graph, nid)
if isinstance(res, Exception):
on_error = node.get("on_error", "continue")
err_msg = str(res)
blackboard[nid] = {
"ok": False,
"node_id": nid,
"node_type": node["type"],
"error": err_msg,
}
trace["nodes"].append({"id": nid, "type": node["type"], "ok": False, "error": err_msg})
logger.warning(f"[orchestrator] 节点 {nid} 失败 on_error={on_error}: {err_msg}")
if on_error == "halt":
raise OrchestrationError(f"节点 {nid} 执行失败(halt): {err_msg}")
else:
blackboard[nid] = res
trace["nodes"].append({
"id": nid,
"type": res.get("node_type"),
"ok": True,
"used_tools": res.get("used_tools"),
})
output_node = _find_output_node(graph)
final = blackboard.get(output_node, {})
if final.get("ok"):
answer = final.get("output", "")
else:
answer = f"⚠️ 编排输出节点 {output_node} 执行失败: {final.get('error', '未知错误')}"
trace["output_node"] = output_node
return answer, trace
"""app/core/orchestrator.py — Phase 3.3 多 Agent 编排图引擎。
设计目标:
- 以轻量 DAG 描述多 Agent 协作:节点(llm/local/external/reduce) + 有向边(数据流)。
- 拓扑分层后按层并行执行(asyncio.gather),互相独立的节点真正并发。
- 失败隔离:每个节点异常不影响其它分支;per-node on_error=halt|continue。
- 节点执行复用既有原语(_build_system_prompt / _execute_skill_tool / _call_external_agent /
3.1 工具沙箱),完全向后兼容——agent.orchestration 为 null 时 chat_service 走原逻辑。
- 并发安全:每个节点用独立 SessionLocal 会话做 DB 读,避免共享请求会话跨协程竞态。
"""
from __future__ import annotations
import asyncio
import json
import logging
from typing import Any, Optional
import httpx
logger = logging.getLogger("aiflow.orchestrator")
VALID_NODE_TYPES = ("llm", "local", "external", "reduce")
class OrchestrationError(ValueError):
"""图结构非法或执行被 halt。"""
# ---------------------------------------------------------------------------
# 校验 + 拓扑
# ---------------------------------------------------------------------------
def validate_graph(graph: Any) -> None:
if not isinstance(graph, dict):
raise OrchestrationError("orchestration 必须为对象")
nodes = graph.get("nodes")
edges = graph.get("edges")
if not isinstance(nodes, list) or not nodes:
raise OrchestrationError("nodes 必须为非空数组")
if not isinstance(edges, list):
raise OrchestrationError("edges 必须为数组")
ids = set()
for i, n in enumerate(nodes):
if not isinstance(n, dict):
raise OrchestrationError(f"节点 #{i} 必须为对象")
nid = n.get("id")
if not nid or not isinstance(nid, str):
raise OrchestrationError(f"节点 #{i} 缺少字符串 id")
if nid in ids:
raise OrchestrationError(f"节点 id 重复: {nid}")
ids.add(nid)
ntype = n.get("type")
if ntype not in VALID_NODE_TYPES:
raise OrchestrationError(
f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(VALID_NODE_TYPES)})"
)
if ntype in ("local", "external") and not n.get("agent_id"):
raise OrchestrationError(f"节点 {nid} 类型 {ntype} 必须指定 agent_id")
for i, e in enumerate(edges):
if not isinstance(e, dict):
raise OrchestrationError(f"边 #{i} 必须为对象")
f = e.get("from")
t = e.get("to")
if f not in ids:
raise OrchestrationError(f"边 #{i} 的 from={f} 指向不存在的节点")
if t not in ids:
raise OrchestrationError(f"边 #{i} 的 to={t} 指向不存在的节点")
if f == t:
raise OrchestrationError(f"边 #{i} 存在自环: {f}")
def topological_layers(graph: dict) -> list:
"""Kahn 算法返回分层(每层内节点可并行);含环则抛错。"""
nodes = graph["nodes"]
edges = graph.get("edges", [])
indeg = {n["id"]: 0 for n in nodes}
adj = {n["id"]: [] for n in nodes}
for e in edges:
f, t = e["from"], e["to"]
adj[f].append(t)
indeg[t] += 1
order = {n["id"]: i for i, n in enumerate(nodes)}
layers: list = []
remaining = set(indeg)
while remaining:
layer = sorted([nid for nid in remaining if indeg[nid] == 0], key=lambda x: order[x])
if not layer:
cycle = sorted(remaining, key=lambda x: order[x])
raise OrchestrationError(f"检测到环路,无法拓扑排序: {cycle}")
layers.append(layer)
for nid in layer:
remaining.discard(nid)
for nxt in adj[nid]:
indeg[nxt] -= 1
return layers
def _node_by_id(graph: dict, nid: str) -> dict:
for n in graph["nodes"]:
if n["id"] == nid:
return n
return {}
def _predecessors(graph: dict, nid: str) -> list:
return [e["from"] for e in graph.get("edges", []) if e["to"] == nid]
def _successors(graph: dict, nid: str) -> list:
return [e["to"] for e in graph.get("edges", []) if e["from"] == nid]
def _find_output_node(graph: dict) -> str:
if graph.get("output_node"):
return graph["output_node"]
sinks = [n["id"] for n in graph["nodes"] if not _successors(graph, n["id"])]
if len(sinks) == 1:
return sinks[0]
for s in sinks:
if _node_by_id(graph, s).get("type") == "reduce":
return s
if sinks:
return sinks[0]
return graph["nodes"][-1]["id"]
def _assemble_node_input(query: str, upstream: str, is_entry: bool) -> str:
"""入口节点直接吃用户原问题;下游节点综合原问题 + 上游输出。"""
if is_entry:
return query or ""
if upstream:
return f"原始用户问题:\n{query}\n\n上游节点输出:\n{upstream}"
return query or ""
# ---------------------------------------------------------------------------
# 模型解析 + LLM 调用原语
# ---------------------------------------------------------------------------
async def _resolve_model_and_endpoint(model_id, db):
from app.models import AIModel, ModelProvider
from ..capability_registry import resolve_auto_model
ai_model = None
if model_id and model_id != "auto":
ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()
if not ai_model:
ai_model = resolve_auto_model("text", db)
if not ai_model:
raise OrchestrationError("无法解析到可用模型")
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_url or not api_key:
raise OrchestrationError(f"节点模型 {ai_model.name} 缺少 endpoint 或 api_key")
return ai_model, api_url, api_key
async def _llm_complete(system_prompt: str, user_content: str, model_id, db, user_id) -> str:
from app.api.chat_utils import _proxy_for_url
from app.core.observability import span
ai_model, api_url, api_key = await _resolve_model_and_endpoint(model_id, db)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_content},
]
request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
async with span("orchestrator.llm", model=ai_model.id):
async with httpx.AsyncClient(
timeout=ai_model.timeout or 120,
proxy=_proxy_for_url(api_url),
trust_env=False,
) as client:
resp = await client.post(
api_url,
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json=request_body,
)
if resp.status_code != 200:
raise OrchestrationError(f"LLM 调用失败 HTTP {resp.status_code}: {resp.text[:200]}")
data = resp.json()
return data["choices"][0]["message"].get("content") or ""
# ---------------------------------------------------------------------------
# local 节点:复用单 Agent system prompt + 受限 tool round + 3.1 沙箱
# ---------------------------------------------------------------------------
async def _run_local_agent_turn(agent: dict, node_input: str, db, user_id, conv_id) -> dict:
from app.api.chat_skill import _resolve_market_skills, _execute_skill_tool
from app.api.chat_context import _build_system_prompt, _retrieve_kb_context
from app.api.chat_service import _resolve_agent_tool_policy, _check_tool_policy
from app.api.chat_utils import _proxy_for_url
skills = _resolve_market_skills(db, agent.get("skill_ids") or [], user_id)
kb_context = await _retrieve_kb_context(db, agent.get("knowledge_base_ids") or [], node_input)
system_prompt, function_definitions = _build_system_prompt(
skills=skills,
agents=[agent],
kb_context=kb_context,
user_content=node_input,
auto_discovered=False,
)
ai_model, api_url, api_key = await _resolve_model_and_endpoint(agent.get("model_id"), db)
tool_policy = _resolve_agent_tool_policy([agent])
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": node_input},
]
called_skills: list = []
tool_call_count = 0
max_tool_rounds = 5
last_content = ""
async with httpx.AsyncClient(
timeout=ai_model.timeout or 120,
proxy=_proxy_for_url(api_url),
trust_env=False,
) as client:
for round_num in range(max_tool_rounds):
request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
if function_definitions:
request_body["tools"] = [
{"type": "function", "function": fd} for fd in function_definitions
]
request_body["tool_choice"] = "auto"
resp = await client.post(
api_url,
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json=request_body,
)
if resp.status_code != 200:
raise OrchestrationError(f"local 节点 LLM 失败 HTTP {resp.status_code}: {resp.text[:200]}")
data = resp.json()
message = data["choices"][0]["message"]
tool_calls = message.get("tool_calls") or []
content = message.get("content") or ""
last_content = content
if not tool_calls or not function_definitions:
return {"text": content, "used_tools": called_skills}
messages.append({"role": "assistant", "content": content, "tool_calls": tool_calls})
round_tool_count = 0
for tc in tool_calls:
func_name = tc.get("function", {}).get("name", "")
check = _check_tool_policy(
func_name, None, skills, tool_policy, tool_call_count, round_tool_count
)
if not check["allowed"]:
messages.append({
"role": "tool",
"tool_call_id": tc.get("id", f"call_{round_num}"),
"content": f"[工具调用已被策略拦截] {check.get('message', check.get('reason'))}",
})
tool_call_count += 1
round_tool_count += 1
continue
tool_call_count += 1
round_tool_count += 1
try:
func_args = json.loads(tc.get("function", {}).get("arguments", "{}") or "{}")
except Exception:
func_args = {}
res = _execute_skill_tool(db, func_name, func_args, skills, node_input)
called_skills.append(func_name)
messages.append({
"role": "tool",
"tool_call_id": tc.get("id", f"call_{round_num}"),
"content": res,
})
return {"text": last_content, "used_tools": called_skills}
# ---------------------------------------------------------------------------
# 单节点执行(含独立 DB 会话,保证并发安全)
# ---------------------------------------------------------------------------
async def _run_node(node, graph, blackboard, query, user_id, conv_id, trace_id) -> dict:
from app.api.chat_skill import _resolve_market_agents
from app.database import SessionLocal
nid = node["id"]
ntype = node["type"]
preds = _predecessors(graph, nid)
pred_results = [blackboard[p] for p in preds]
is_entry = not preds
if pred_results:
parts = []
for b in pred_results:
label = f"节点 {b.get('node_id')}/{b.get('node_type')}"
if b.get("ok"):
parts.append(f"【{label}】\n{b.get('output', '')}")
else:
parts.append(f"【{label} 执行失败】\n错误: {b.get('error', '')}")
upstream = "\n\n".join(parts)
else:
upstream = ""
with SessionLocal() as db:
if ntype in ("llm", "reduce"):
if ntype == "llm":
sys_prompt = node.get("system_prompt") or (
"你是一个处理节点,请根据上游输入完成指定的文本变换、摘要或路由判断,"
"只输出处理后的结果,不要多余解释。"
)
else:
sys_prompt = node.get("system_prompt") or (
"你是一个结果汇总器,请将多个子任务的输出综合成连贯、完整、面向用户的最终回答。"
)
uc = _assemble_node_input(query, upstream, is_entry)
text = await _llm_complete(sys_prompt, uc, node.get("model_id"), db, user_id)
return {"ok": True, "node_id": nid, "node_type": ntype, "output": text}
if ntype == "local":
agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
if not agents:
raise OrchestrationError(f"local 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
ni = _assemble_node_input(query, upstream, is_entry)
res = await _run_local_agent_turn(agents[0], ni, db, user_id, conv_id)
return {
"ok": True,
"node_id": nid,
"node_type": "local",
"output": res["text"],
"used_tools": res.get("used_tools"),
}
if ntype == "external":
from app.api.chat_agents import _call_external_agent
agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
if not agents:
raise OrchestrationError(f"external 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
ni = _assemble_node_input(query, upstream, is_entry)
answer, _ = await _call_external_agent(
db, agents[0], query=ni, user_id=user_id, ext_conversation_id=None, history=[]
)
return {"ok": True, "node_id": nid, "node_type": "external", "output": answer or ""}
raise OrchestrationError(f"未知节点类型 {ntype}")
# ---------------------------------------------------------------------------
# 入口
# ---------------------------------------------------------------------------
async def run_graph(graph, query, db, user_id, conv_id, trace_id=None):
"""执行编排图,返回 (answer: str, trace: dict)。
- 拓扑分层后逐层 asyncio.gather 并行执行。
- 节点异常按 on_error=halt|continue 处理(默认 continue,失败时下游拿到错误作为输入)。
"""
validate_graph(graph)
layers = topological_layers(graph)
blackboard: dict = {}
trace = {"nodes": [], "trace_id": trace_id, "layers": layers}
for layer in layers:
tasks = [
_run_node(_node_by_id(graph, nid), graph, blackboard, query, user_id, conv_id, trace_id)
for nid in layer
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for nid, res in zip(layer, results):
node = _node_by_id(graph, nid)
if isinstance(res, Exception):
on_error = node.get("on_error", "continue")
err_msg = str(res)
blackboard[nid] = {
"ok": False,
"node_id": nid,
"node_type": node["type"],
"error": err_msg,
}
trace["nodes"].append({"id": nid, "type": node["type"], "ok": False, "error": err_msg})
logger.warning(f"[orchestrator] 节点 {nid} 失败 on_error={on_error}: {err_msg}")
if on_error == "halt":
raise OrchestrationError(f"节点 {nid} 执行失败(halt): {err_msg}")
else:
blackboard[nid] = res
trace["nodes"].append({
"id": nid,
"type": res.get("node_type"),
"ok": True,
"used_tools": res.get("used_tools"),
})
output_node = _find_output_node(graph)
final = blackboard.get(output_node, {})
if final.get("ok"):
answer = final.get("output", "")
else:
answer = f"⚠️ 编排输出节点 {output_node} 执行失败: {final.get('error', '未知错误')}"
trace["output_node"] = output_node
return answer, trace
+131 -131
View File
@@ -1,131 +1,131 @@
"""
OpenTelemetry 桥接 (Phase 1.I 标准化)
=====================================
设计基准(标准对照表见 STANDARDS_AUDIT_2026-07-24.md §I):
- 标准:OpenTelemetry SDK + OTLP gRPC exporter —— 云原生可观测事实标准。
- traces:span 经 observability.span() 嵌套导出(见 observability.py)。
- metrics:MeterProvider 经 OTLP 导出 span 耗时直方图 + 调用计数,供 Grafana 出图。
- 失败安全(生产红线):若 OTLP 端点不可达 / 初始化异常,自动降级为 no-op,
绝不阻断业务请求、不产生未捕获异常。
- 门控:OTEL_ENABLED=1 默认开启;置 0 则零开销、不走 OTel。
- 发射目标:OTEL_EXPORTER_OTLP_ENDPOINT,默认 http://qihu-aiflow-alloy:4317
(经 Grafana Alloy 桥接 OTLP→Prometheus,再由 Prometheus 喂给 Grafana 出图)。
对外 API:
init_otel() 首次调用时构建 TracerProvider + MeterProvider(幂等)
get_tracer() 返回 tracer(未启用/失败则 None)
get_meter() 返回 meter(未启用/失败则 None)
current_span_cm(name) 返回一个 OTel span 的上下文管理器(供 observability 嵌套)
record_span_metrics(name, elapsed_ms, status) 记录 span 指标(耗时直方图 + 计数)
"""
import os
import logging
logger = logging.getLogger(__name__)
_ENABLED = os.getenv("OTEL_ENABLED", "1") not in ("0", "false", "False", "")
_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://qihu-aiflow-alloy:4317").rstrip("/")
if not _ENDPOINT.startswith("http://") and not _ENDPOINT.startswith("https://"):
_ENDPOINT = "http://" + _ENDPOINT
_tracer = None
_meter = None
_initialized = False
_duration_hist = None
_span_counter = None
def init_otel() -> None:
"""构建 TracerProvider + MeterProvider(幂等)。失败则降级为 no-op。"""
global _tracer, _meter, _initialized
if _initialized:
return
_initialized = True
if not _ENABLED:
logger.info("[otel] disabled (OTEL_ENABLED=0)")
return
try:
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
OTLPSpanExporter,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
OTLPMetricExporter,
)
resource = Resource.create({ResourceAttributes.SERVICE_NAME: "qihu-aiflow"})
# --- traces ---
trace_provider = TracerProvider(resource=resource)
trace_provider.add_span_processor(
BatchSpanProcessor(OTLPSpanExporter(endpoint=_ENDPOINT, insecure=True))
)
trace.set_tracer_provider(trace_provider)
_tracer = trace.get_tracer("qihu-aiflow")
# --- metrics ---
metric_reader = PeriodicExportingMetricReader(
OTLPMetricExporter(endpoint=_ENDPOINT, insecure=True)
)
meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
metrics.set_meter_provider(meter_provider)
_meter = metrics.get_meter("qihu-aiflow")
logger.info(f"[otel] tracer+meter ready -> {_ENDPOINT}")
except Exception as e: # pragma: no cover - 降级保护
logger.warning(f"[otel] init failed, degraded to no-op: {e}")
_tracer = None
_meter = None
def get_tracer():
"""返回 tracer;首次调用触发 init。未启用/失败返回 None。"""
if _tracer is None and _ENABLED:
init_otel()
return _tracer
def get_meter():
"""返回 meter;首次调用触发 init。未启用/失败返回 None。"""
if _meter is None and _ENABLED:
init_otel()
return _meter
def current_span_cm(name: str, attributes: dict | None = None):
"""返回一个 OTel span 的上下文管理器(同步/异步通用)。失败时返回 None。"""
tr = get_tracer()
if tr is None:
return None
try:
return tr.start_as_current_span(name, attributes=attributes or {})
except Exception as e: # pragma: no cover
logger.warning(f"[otel] start_span failed: {e}")
return None
def record_span_metrics(name: str, elapsed_ms: float, status: str) -> None:
"""记录 span 指标:耗时直方图 + 调用计数(供 Grafana 出图)。失败安全。"""
m = get_meter()
if m is None:
return
global _duration_hist, _span_counter
try:
if _duration_hist is None:
_duration_hist = m.create_histogram(
"aiflow_span_duration_ms", "Span duration in milliseconds"
)
_span_counter = m.create_counter("aiflow_span_count", "Span invocation count")
attrs = {"span": name, "status": status}
_duration_hist.record(elapsed_ms, attrs)
_span_counter.add(1, attrs)
except Exception: # pragma: no cover
pass
"""
OpenTelemetry 桥接 (Phase 1.I 标准化)
=====================================
设计基准(标准对照表见 STANDARDS_AUDIT_2026-07-24.md §I):
- 标准:OpenTelemetry SDK + OTLP gRPC exporter —— 云原生可观测事实标准。
- traces:span 经 observability.span() 嵌套导出(见 observability.py)。
- metrics:MeterProvider 经 OTLP 导出 span 耗时直方图 + 调用计数,供 Grafana 出图。
- 失败安全(生产红线):若 OTLP 端点不可达 / 初始化异常,自动降级为 no-op,
绝不阻断业务请求、不产生未捕获异常。
- 门控:OTEL_ENABLED=1 默认开启;置 0 则零开销、不走 OTel。
- 发射目标:OTEL_EXPORTER_OTLP_ENDPOINT,默认 http://qihu-aiflow-alloy:4317
(经 Grafana Alloy 桥接 OTLP→Prometheus,再由 Prometheus 喂给 Grafana 出图)。
对外 API:
init_otel() 首次调用时构建 TracerProvider + MeterProvider(幂等)
get_tracer() 返回 tracer(未启用/失败则 None)
get_meter() 返回 meter(未启用/失败则 None)
current_span_cm(name) 返回一个 OTel span 的上下文管理器(供 observability 嵌套)
record_span_metrics(name, elapsed_ms, status) 记录 span 指标(耗时直方图 + 计数)
"""
import os
import logging
logger = logging.getLogger(__name__)
_ENABLED = os.getenv("OTEL_ENABLED", "1") not in ("0", "false", "False", "")
_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://qihu-aiflow-alloy:4317").rstrip("/")
if not _ENDPOINT.startswith("http://") and not _ENDPOINT.startswith("https://"):
_ENDPOINT = "http://" + _ENDPOINT
_tracer = None
_meter = None
_initialized = False
_duration_hist = None
_span_counter = None
def init_otel() -> None:
"""构建 TracerProvider + MeterProvider(幂等)。失败则降级为 no-op。"""
global _tracer, _meter, _initialized
if _initialized:
return
_initialized = True
if not _ENABLED:
logger.info("[otel] disabled (OTEL_ENABLED=0)")
return
try:
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
OTLPSpanExporter,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
OTLPMetricExporter,
)
resource = Resource.create({ResourceAttributes.SERVICE_NAME: "qihu-aiflow"})
# --- traces ---
trace_provider = TracerProvider(resource=resource)
trace_provider.add_span_processor(
BatchSpanProcessor(OTLPSpanExporter(endpoint=_ENDPOINT, insecure=True))
)
trace.set_tracer_provider(trace_provider)
_tracer = trace.get_tracer("qihu-aiflow")
# --- metrics ---
metric_reader = PeriodicExportingMetricReader(
OTLPMetricExporter(endpoint=_ENDPOINT, insecure=True)
)
meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
metrics.set_meter_provider(meter_provider)
_meter = metrics.get_meter("qihu-aiflow")
logger.info(f"[otel] tracer+meter ready -> {_ENDPOINT}")
except Exception as e: # pragma: no cover - 降级保护
logger.warning(f"[otel] init failed, degraded to no-op: {e}")
_tracer = None
_meter = None
def get_tracer():
"""返回 tracer;首次调用触发 init。未启用/失败返回 None。"""
if _tracer is None and _ENABLED:
init_otel()
return _tracer
def get_meter():
"""返回 meter;首次调用触发 init。未启用/失败返回 None。"""
if _meter is None and _ENABLED:
init_otel()
return _meter
def current_span_cm(name: str, attributes: dict | None = None):
"""返回一个 OTel span 的上下文管理器(同步/异步通用)。失败时返回 None。"""
tr = get_tracer()
if tr is None:
return None
try:
return tr.start_as_current_span(name, attributes=attributes or {})
except Exception as e: # pragma: no cover
logger.warning(f"[otel] start_span failed: {e}")
return None
def record_span_metrics(name: str, elapsed_ms: float, status: str) -> None:
"""记录 span 指标:耗时直方图 + 调用计数(供 Grafana 出图)。失败安全。"""
m = get_meter()
if m is None:
return
global _duration_hist, _span_counter
try:
if _duration_hist is None:
_duration_hist = m.create_histogram(
"aiflow_span_duration_ms", "Span duration in milliseconds"
)
_span_counter = m.create_counter("aiflow_span_count", "Span invocation count")
attrs = {"span": name, "status": status}
_duration_hist.record(elapsed_ms, attrs)
_span_counter.add(1, attrs)
except Exception: # pragma: no cover
pass
+271 -271
View File
@@ -1,271 +1,271 @@
"""
知识库路由层
============
将检索请求智能路由到本地BM25、向量检索、或外部KB服务(Dify)。
根据知识库配置自动选择最优检索策略。
使用方式:
from app.kb_router import kb_router_search
results = await kb_router_search(query, kb_id, top_k=5)
"""
import logging
from typing import List, Dict, Optional
logger = logging.getLogger("aiflow.kb_router")
async def _get_external_services(db_session, enabled_only: bool = True) -> List[Dict]:
"""获取已配置的外部KB服务(仅 dify)"""
from sqlalchemy import text
try:
query = "SELECT * FROM external_kb_services"
if enabled_only:
query += " WHERE is_enabled = 1"
query += " ORDER BY created_at DESC"
result = db_session.execute(text(query))
columns = result.keys()
return [dict(zip(columns, row)) for row in result.fetchall()]
except Exception as e:
logger.error(f"Failed to get external services: {e}")
return []
async def _search_dify(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
"""检索 Dify 知识库(走 app.dify_kb 客户端)"""
try:
from app.dify_kb import dify_retrieve
api_url = service.get("api_url", "")
api_key = service.get("api_key", "")
dataset_name = None
cfg = service.get("config")
if isinstance(cfg, str):
try:
import json
cfg = json.loads(cfg)
except Exception:
cfg = None
if isinstance(cfg, dict):
dataset_name = cfg.get("dataset_name")
hits = dify_retrieve(api_url, api_key, query, top_k=top_k, dataset_name=dataset_name)
return hits
except Exception as e:
logger.error(f"Dify search error: {e}")
return []
async def _search_external(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
"""路由到对应的外部KB服务(当前仅支持 dify)"""
service_type = service.get("service_type", "")
if service_type == "dify":
return await _search_dify(service, query, top_k)
else:
logger.warning(f"Unknown external KB type: {service_type}")
return []
async def _get_web_services(db_session, enabled_only: bool = True) -> List[Dict]:
"""获取全局 AnySearch 联网检索配置(从 site_settings 读取)。"""
from app.anysearch_client import load_anysearch_config
cfg = load_anysearch_config(db_session)
if not cfg:
return []
if enabled_only and not cfg.get("is_enabled"):
return []
api_key = cfg.get("api_key", "")
if not api_key:
return []
return [{
"id": "anysearch_global",
"service_type": "anysearch",
"api_key": api_key,
"api_url": cfg.get("api_url") or "https://api.anysearch.com/v1/search",
"config": {
"zone": cfg.get("zone", "cn"),
"freshness": cfg.get("freshness"),
"top_k": cfg.get("top_k", 5),
},
"is_enabled": cfg.get("is_enabled", False),
}]
async def _search_anysearch(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
"""调用 AnySearch 联网检索(走 app.anysearch_client)。"""
try:
from app.anysearch_client import anysearch_search
api_key = service.get("api_key", "")
api_url = service.get("api_url") or "https://api.anysearch.com/v1/search"
cfg = service.get("config")
if isinstance(cfg, str):
try:
import json
cfg = json.loads(cfg)
except Exception:
cfg = None
cfg = cfg or {}
zone = cfg.get("zone", "cn")
freshness = cfg.get("freshness")
k = int(cfg.get("top_k", top_k))
return await anysearch_search(api_key, api_url, query, top_k=k, zone=zone, freshness=freshness)
except Exception as e:
logger.error(f"AnySearch search error: {e}")
return []
async def kb_router_search(
query: str,
kb_id: Optional[str] = None,
kb_ids: Optional[List[str]] = None,
top_k: int = 5,
bm25_results: Optional[List[Dict]] = None,
db_session=None,
use_vector: bool = True,
use_external: bool = True,
use_tree: bool = True,
use_web: bool = False,
) -> List[Dict]:
"""
知识库路由层 — 混合检索
检索策略:
1. 本地 BM25 检索 (MySQL, 已有)
2. 向量语义检索 (PostgreSQL+pgvector, 新增)
3. 结构化分片段检索 (TreeSearch, 新增: 知识库目录树 + chunk 级精准命中)
4. 外部KB检索 (Dify, 按配置)
5. 联网检索 (AnySearch, 可选 use_web)
融合方式: Reciprocal Rank Fusion (RRF)
Args:
query: 查询文本
kb_id: 知识库ID
top_k: 返回结果数
bm25_results: BM25检索结果
db_session: 数据库会话(用于获取外部KB配置与分片段检索)
use_vector: 是否使用向量检索
use_external: 是否使用外部KB检索
use_tree: 是否使用 TreeSearch 结构化分片段检索
use_web: 是否使用 AnySearch 联网检索
Returns:
融合后的检索结果列表
"""
all_results = []
# 1. BM25 结果
if bm25_results:
for rank, item in enumerate(bm25_results):
all_results.append({
"content": item.get("content", item.get("text", "")),
"score": item.get("score", 1.0 / (60 + rank + 1)),
"source": "local_bm25",
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
# 2. 向量检索
if use_vector:
try:
from app.pgvector_search import vector_search
kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else [None])
vec_results = []
for _kid in kb_targets:
try:
vec_results.extend(await vector_search(query, _kid, top_k))
except Exception as e:
logger.warning(f"Vector search failed for kb={_kid}: {e}")
for rank, item in enumerate(vec_results):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.5),
"source": "vector_pg",
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
except ImportError:
logger.debug("pgvector_search not available, skipping vector search")
except Exception as e:
logger.warning(f"Vector search failed: {e}")
# 2.5 结构化分片段检索 (TreeSearch, Level 2)
if use_tree and db_session is not None:
try:
from app.tree_search import tree_search
kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else None)
tree_hits = await tree_search(
db_session, query, kb_ids=kb_targets, top_k=top_k, mode="hybrid",
)
for rank, item in enumerate(tree_hits):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.5),
"source": "tree_chunk",
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
except Exception as e:
logger.warning(f"TreeSearch failed: {e}")
# 3. 外部KB检索 (Dify)
if use_external and db_session:
services = await _get_external_services(db_session)
for service in services:
if service.get("status") == "connected":
ext_results = await _search_external(service, query, top_k)
for rank, item in enumerate(ext_results):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.5),
"source": item.get("source", "external_dify"),
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
# 4. 联网检索 (AnySearch, 可选)
if use_web and db_session is not None:
try:
web_services = await _get_web_services(db_session)
for svc in web_services:
web_hits = await _search_anysearch(svc, query, top_k)
for rank, item in enumerate(web_hits):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.6),
"source": item.get("source", "web_anysearch"),
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
except Exception as e:
logger.warning(f"AnySearch web search failed: {e}")
if not all_results:
return []
# RRF 融合
rrf_k = 60
score_map = {}
for item in all_results:
content_key = item["content"][:200] # 用内容前200字符作为去重键
rrf_score = 1.0 / (rrf_k + item.get("rank", 1))
if content_key in score_map:
score_map[content_key]["score"] += rrf_score
score_map[content_key]["sources"].append(item["source"])
else:
score_map[content_key] = {
"content": item["content"],
"score": rrf_score,
"sources": [item["source"]],
"metadata": item.get("metadata", {}),
}
# 按融合分数排序
merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True)
# 标准化分数到 0-1
if merged:
max_score = merged[0]["score"]
for item in merged:
item["score"] = item["score"] / max_score if max_score > 0 else 0
item["sources"] = list(set(item["sources"]))
return merged[:top_k]
"""
知识库路由层
============
将检索请求智能路由到本地BM25、向量检索、或外部KB服务(Dify)。
根据知识库配置自动选择最优检索策略。
使用方式:
from app.kb_router import kb_router_search
results = await kb_router_search(query, kb_id, top_k=5)
"""
import logging
from typing import List, Dict, Optional
logger = logging.getLogger("aiflow.kb_router")
async def _get_external_services(db_session, enabled_only: bool = True) -> List[Dict]:
"""获取已配置的外部KB服务(仅 dify)"""
from sqlalchemy import text
try:
query = "SELECT * FROM external_kb_services"
if enabled_only:
query += " WHERE is_enabled = 1"
query += " ORDER BY created_at DESC"
result = db_session.execute(text(query))
columns = result.keys()
return [dict(zip(columns, row)) for row in result.fetchall()]
except Exception as e:
logger.error(f"Failed to get external services: {e}")
return []
async def _search_dify(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
"""检索 Dify 知识库(走 app.dify_kb 客户端)"""
try:
from app.dify_kb import dify_retrieve
api_url = service.get("api_url", "")
api_key = service.get("api_key", "")
dataset_name = None
cfg = service.get("config")
if isinstance(cfg, str):
try:
import json
cfg = json.loads(cfg)
except Exception:
cfg = None
if isinstance(cfg, dict):
dataset_name = cfg.get("dataset_name")
hits = dify_retrieve(api_url, api_key, query, top_k=top_k, dataset_name=dataset_name)
return hits
except Exception as e:
logger.error(f"Dify search error: {e}")
return []
async def _search_external(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
"""路由到对应的外部KB服务(当前仅支持 dify)"""
service_type = service.get("service_type", "")
if service_type == "dify":
return await _search_dify(service, query, top_k)
else:
logger.warning(f"Unknown external KB type: {service_type}")
return []
async def _get_web_services(db_session, enabled_only: bool = True) -> List[Dict]:
"""获取全局 AnySearch 联网检索配置(从 site_settings 读取)。"""
from app.anysearch_client import load_anysearch_config
cfg = load_anysearch_config(db_session)
if not cfg:
return []
if enabled_only and not cfg.get("is_enabled"):
return []
api_key = cfg.get("api_key", "")
if not api_key:
return []
return [{
"id": "anysearch_global",
"service_type": "anysearch",
"api_key": api_key,
"api_url": cfg.get("api_url") or "https://api.anysearch.com/v1/search",
"config": {
"zone": cfg.get("zone", "cn"),
"freshness": cfg.get("freshness"),
"top_k": cfg.get("top_k", 5),
},
"is_enabled": cfg.get("is_enabled", False),
}]
async def _search_anysearch(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
"""调用 AnySearch 联网检索(走 app.anysearch_client)。"""
try:
from app.anysearch_client import anysearch_search
api_key = service.get("api_key", "")
api_url = service.get("api_url") or "https://api.anysearch.com/v1/search"
cfg = service.get("config")
if isinstance(cfg, str):
try:
import json
cfg = json.loads(cfg)
except Exception:
cfg = None
cfg = cfg or {}
zone = cfg.get("zone", "cn")
freshness = cfg.get("freshness")
k = int(cfg.get("top_k", top_k))
return await anysearch_search(api_key, api_url, query, top_k=k, zone=zone, freshness=freshness)
except Exception as e:
logger.error(f"AnySearch search error: {e}")
return []
async def kb_router_search(
query: str,
kb_id: Optional[str] = None,
kb_ids: Optional[List[str]] = None,
top_k: int = 5,
bm25_results: Optional[List[Dict]] = None,
db_session=None,
use_vector: bool = True,
use_external: bool = True,
use_tree: bool = True,
use_web: bool = False,
) -> List[Dict]:
"""
知识库路由层 — 混合检索
检索策略:
1. 本地 BM25 检索 (MySQL, 已有)
2. 向量语义检索 (PostgreSQL+pgvector, 新增)
3. 结构化分片段检索 (TreeSearch, 新增: 知识库目录树 + chunk 级精准命中)
4. 外部KB检索 (Dify, 按配置)
5. 联网检索 (AnySearch, 可选 use_web)
融合方式: Reciprocal Rank Fusion (RRF)
Args:
query: 查询文本
kb_id: 知识库ID
top_k: 返回结果数
bm25_results: BM25检索结果
db_session: 数据库会话(用于获取外部KB配置与分片段检索)
use_vector: 是否使用向量检索
use_external: 是否使用外部KB检索
use_tree: 是否使用 TreeSearch 结构化分片段检索
use_web: 是否使用 AnySearch 联网检索
Returns:
融合后的检索结果列表
"""
all_results = []
# 1. BM25 结果
if bm25_results:
for rank, item in enumerate(bm25_results):
all_results.append({
"content": item.get("content", item.get("text", "")),
"score": item.get("score", 1.0 / (60 + rank + 1)),
"source": "local_bm25",
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
# 2. 向量检索
if use_vector:
try:
from app.pgvector_search import vector_search
kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else [None])
vec_results = []
for _kid in kb_targets:
try:
vec_results.extend(await vector_search(query, _kid, top_k))
except Exception as e:
logger.warning(f"Vector search failed for kb={_kid}: {e}")
for rank, item in enumerate(vec_results):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.5),
"source": "vector_pg",
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
except ImportError:
logger.debug("pgvector_search not available, skipping vector search")
except Exception as e:
logger.warning(f"Vector search failed: {e}")
# 2.5 结构化分片段检索 (TreeSearch, Level 2)
if use_tree and db_session is not None:
try:
from app.tree_search import tree_search
kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else None)
tree_hits = await tree_search(
db_session, query, kb_ids=kb_targets, top_k=top_k, mode="hybrid",
)
for rank, item in enumerate(tree_hits):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.5),
"source": "tree_chunk",
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
except Exception as e:
logger.warning(f"TreeSearch failed: {e}")
# 3. 外部KB检索 (Dify)
if use_external and db_session:
services = await _get_external_services(db_session)
for service in services:
if service.get("status") == "connected":
ext_results = await _search_external(service, query, top_k)
for rank, item in enumerate(ext_results):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.5),
"source": item.get("source", "external_dify"),
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
# 4. 联网检索 (AnySearch, 可选)
if use_web and db_session is not None:
try:
web_services = await _get_web_services(db_session)
for svc in web_services:
web_hits = await _search_anysearch(svc, query, top_k)
for rank, item in enumerate(web_hits):
all_results.append({
"content": item.get("content", ""),
"score": item.get("score", 0.6),
"source": item.get("source", "web_anysearch"),
"rank": rank + 1,
"metadata": item.get("metadata", {}),
})
except Exception as e:
logger.warning(f"AnySearch web search failed: {e}")
if not all_results:
return []
# RRF 融合
rrf_k = 60
score_map = {}
for item in all_results:
content_key = item["content"][:200] # 用内容前200字符作为去重键
rrf_score = 1.0 / (rrf_k + item.get("rank", 1))
if content_key in score_map:
score_map[content_key]["score"] += rrf_score
score_map[content_key]["sources"].append(item["source"])
else:
score_map[content_key] = {
"content": item["content"],
"score": rrf_score,
"sources": [item["source"]],
"metadata": item.get("metadata", {}),
}
# 按融合分数排序
merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True)
# 标准化分数到 0-1
if merged:
max_score = merged[0]["score"]
for item in merged:
item["score"] = item["score"] / max_score if max_score > 0 else 0
item["sources"] = list(set(item["sources"]))
return merged[:top_k]
+379 -379
View File
@@ -1,379 +1,379 @@
"""
PGVector 向量检索模块
====================
使用 PostgreSQL + pgvector 实现语义向量检索,
与现有 MySQL BM25 检索配合,实现混合检索。
使用方式:
from app.pgvector_search import vector_search
results = await vector_search(query, kb_id, top_k=5)
"""
import os
import logging
import asyncio
from typing import List, Dict, Optional
from datetime import datetime
logger = logging.getLogger("aiflow.pgvector")
# PostgreSQL 连接配置
PG_HOST = os.environ.get("POSTGRES_HOST", "qihu-aiflow-pgvector")
PG_PORT = int(os.environ.get("POSTGRES_PORT", "5432"))
PG_USER = os.environ.get("POSTGRES_USER", "aiflow")
PG_PASSWORD = os.environ.get("POSTGRES_PASSWORD", "pwd2026")
PG_DB = os.environ.get("POSTGRES_DB", "aiflow_vectors")
# 嵌入模型配置
EMBEDDING_API_URL = os.environ.get("SILICONFLOW_BASE_URL", "https://api.siliconflow.cn")
EMBEDDING_API_KEY = os.environ.get("SILICONFLOW_API_KEY", "")
EMBEDDING_MODEL = os.environ.get("EMBEDDING_MODEL", "BAAI/bge-large-zh-v1.5")
# ---- 本地嵌入缓存(SQLite 持久化):破解"每次嵌入消耗外部 API 额度"的隐性成本障碍 ----
import sqlite3, hashlib, json as _json
_EMBED_CACHE_DB = os.environ.get("EMBED_CACHE_DB", "/app/backend/python/.embed_cache.db")
def _embed_cache_get(text, model):
try:
key = hashlib.sha256(f"{model}::{text}".encode("utf-8")).hexdigest()
conn = sqlite3.connect(_EMBED_CACHE_DB, timeout=5)
row = conn.execute("SELECT vec FROM embed_cache WHERE k=?", (key,)).fetchone()
conn.close()
if row:
return _json.loads(row[0])
except Exception:
return None
return None
def _embed_cache_put(text, model, vec):
try:
key = hashlib.sha256(f"{model}::{text}".encode("utf-8")).hexdigest()
conn = sqlite3.connect(_EMBED_CACHE_DB, timeout=5)
conn.execute("CREATE TABLE IF NOT EXISTS embed_cache (k TEXT PRIMARY KEY, vec TEXT, ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP)")
conn.execute("INSERT OR REPLACE INTO embed_cache (k, vec) VALUES (?, ?)", (key, _json.dumps(vec)))
conn.commit()
conn.close()
except Exception:
pass
# 连接池
_pg_pool = None
_pool_lock = asyncio.Lock()
async def get_pg_pool():
"""获取 PostgreSQL 异步连接池"""
global _pg_pool
if _pg_pool is not None:
return _pg_pool
async with _pool_lock:
if _pg_pool is not None:
return _pg_pool
try:
import asyncpg
except ImportError:
logger.warning("asyncpg not installed, vector search disabled")
return None
try:
_pg_pool = await asyncpg.create_pool(
host=PG_HOST,
port=PG_PORT,
user=PG_USER,
password=PG_PASSWORD,
database=PG_DB,
min_size=2,
max_size=10,
command_timeout=30,
)
logger.info(f"PostgreSQL connection pool created: {PG_HOST}:{PG_PORT}/{PG_DB}")
# 初始化表结构
async with _pg_pool.acquire() as conn:
await conn.execute("""
CREATE TABLE IF NOT EXISTS document_embeddings (
id SERIAL PRIMARY KEY,
doc_id VARCHAR(64) NOT NULL,
kb_id VARCHAR(64) NOT NULL,
chunk_id VARCHAR(64),
content TEXT NOT NULL,
embedding vector(1024),
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_embeddings_kb
ON document_embeddings(kb_id);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_embeddings_doc
ON document_embeddings(doc_id);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_embeddings_vector
ON document_embeddings USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
""")
logger.info("PGVector tables initialized")
return _pg_pool
except Exception as e:
logger.error(f"Failed to create PostgreSQL pool: {e}")
_pg_pool = None
return None
async def get_embedding(text: str) -> Optional[List[float]]:
"""调用嵌入API获取文本向量"""
if not EMBEDDING_API_KEY:
logger.warning("No embedding API key configured")
return None
cached = _embed_cache_get(text, EMBEDDING_MODEL)
if cached:
return cached
try:
# 收敛到共享标准模型层(app.core.llm_client.aembed),与聊天共用代理/超时/failover 语义
from app.core.llm_client import aembed
embedding = await aembed(
EMBEDDING_API_URL, EMBEDDING_API_KEY, text[:8000], EMBEDDING_MODEL, timeout=30
)
if embedding:
_embed_cache_put(text, EMBEDDING_MODEL, embedding)
return embedding
logger.error("Embedding API returned empty")
return None
except Exception as e:
logger.error(f"Embedding generation failed: {e}")
return None
async def vector_search(
query: str,
kb_id: Optional[str] = None,
top_k: int = 5,
threshold: float = 0.5,
) -> List[Dict]:
"""
向量语义检索
Args:
query: 查询文本
kb_id: 知识库ID (可选,不指定则搜索全部)
top_k: 返回结果数
threshold: 相似度阈值 (0-1)
Returns:
[{"doc_id": ..., "content": ..., "score": ..., "metadata": ...}, ...]
"""
pool = await get_pg_pool()
if pool is None:
return []
# 获取查询向量
query_embedding = await get_embedding(query)
if query_embedding is None:
return []
try:
async with pool.acquire() as conn:
# 将向量转换为 PostgreSQL 格式
embedding_str = f"[{','.join(str(x) for x in query_embedding)}]"
if kb_id:
rows = await conn.fetch(
"""
SELECT doc_id, chunk_id, content, metadata,
1 - (embedding <=> $1::vector) as score
FROM document_embeddings
WHERE kb_id = $2
ORDER BY embedding <=> $1::vector
LIMIT $3;
""",
embedding_str, kb_id, top_k,
)
else:
rows = await conn.fetch(
"""
SELECT doc_id, chunk_id, content, metadata,
1 - (embedding <=> $1::vector) as score
FROM document_embeddings
ORDER BY embedding <=> $1::vector
LIMIT $2;
""",
embedding_str, top_k,
)
results = []
for row in rows:
score = float(row["score"])
if score >= threshold:
results.append({
"doc_id": row["doc_id"],
"chunk_id": row["chunk_id"],
"content": row["content"],
"score": score,
"metadata": row["metadata"] if isinstance(row["metadata"], dict) else {},
})
return results
except Exception as e:
logger.error(f"Vector search failed: {e}")
return []
async def index_document(
doc_id: str,
kb_id: str,
content: str,
chunk_id: Optional[str] = None,
metadata: Optional[Dict] = None,
) -> bool:
"""
将文档内容索引到 PGVector
Args:
doc_id: 文档ID
kb_id: 知识库ID
content: 文档内容
chunk_id: 分块ID (可选)
metadata: 元数据
Returns:
是否成功
"""
pool = await get_pg_pool()
if pool is None:
return False
# 分块处理(简单按长度分块)
chunks = []
chunk_size = 500
overlap = 50
if len(content) <= chunk_size:
chunks = [(content, chunk_id or f"{doc_id}_0")]
else:
for i in range(0, len(content), chunk_size - overlap):
chunk = content[i:i + chunk_size]
if len(chunk.strip()) > 10:
chunks.append((chunk, f"{doc_id}_{i // (chunk_size - overlap)}"))
success_count = 0
for chunk_text, cid in chunks:
embedding = await get_embedding(chunk_text)
if embedding is None:
continue
try:
async with pool.acquire() as conn:
embedding_str = f"[{','.join(str(x) for x in embedding)}]"
import json
await conn.execute(
"""
INSERT INTO document_embeddings (doc_id, kb_id, chunk_id, content, embedding, metadata)
VALUES ($1, $2, $3, $4, $5::vector, $6)
ON CONFLICT DO NOTHING;
""",
doc_id, kb_id, cid, chunk_text, embedding_str,
json.dumps(metadata or {}),
)
success_count += 1
except Exception as e:
logger.error(f"Failed to index chunk {cid}: {e}")
logger.info(f"Indexed document {doc_id}: {success_count}/{len(chunks)} chunks")
return success_count > 0
async def hybrid_search(
query: str,
kb_id: Optional[str] = None,
top_k: int = 5,
bm25_results: Optional[List[Dict]] = None,
) -> List[Dict]:
"""
混合检索:BM25 + 向量检索融合
Args:
query: 查询文本
kb_id: 知识库ID
top_k: 返回结果数
bm25_results: BM25 检索结果(来自MySQL)
Returns:
融合后的检索结果
"""
# 获取向量检索结果
vec_results = await vector_search(query, kb_id, top_k)
if not vec_results and not bm25_results:
return []
if not vec_results:
return bm25_results or []
if not bm25_results:
return vec_results
# 融合:Reciprocal Rank Fusion (RRF)
rrf_k = 60
score_map = {}
for rank, item in enumerate(bm25_results):
doc_id = item.get("doc_id", item.get("id", ""))
score_map[doc_id] = {
"content": item.get("content", item.get("text", "")),
"score": 1.0 / (rrf_k + rank + 1),
"bm25_rank": rank + 1,
"metadata": item.get("metadata", {}),
}
for rank, item in enumerate(vec_results):
doc_id = item["doc_id"]
if doc_id in score_map:
score_map[doc_id]["score"] += 1.0 / (rrf_k + rank + 1)
score_map[doc_id]["vec_score"] = item["score"]
score_map[doc_id]["vec_rank"] = rank + 1
else:
score_map[doc_id] = {
"content": item["content"],
"score": 1.0 / (rrf_k + rank + 1),
"vec_score": item["score"],
"vec_rank": rank + 1,
"metadata": item.get("metadata", {}),
}
# 按融合分数排序
merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True)
return merged[:top_k]
async def delete_document_embeddings(doc_id: str) -> bool:
"""删除文档的所有向量索引"""
pool = await get_pg_pool()
if pool is None:
return False
try:
async with pool.acquire() as conn:
await conn.execute(
"DELETE FROM document_embeddings WHERE doc_id = $1;",
doc_id,
)
logger.info(f"Deleted embeddings for document {doc_id}")
return True
except Exception as e:
logger.error(f"Failed to delete embeddings: {e}")
return False
async def close_pool():
"""关闭连接池"""
global _pg_pool
if _pg_pool is not None:
await _pg_pool.close()
_pg_pool = None
logger.info("PostgreSQL connection pool closed")
"""
PGVector 向量检索模块
====================
使用 PostgreSQL + pgvector 实现语义向量检索,
与现有 MySQL BM25 检索配合,实现混合检索。
使用方式:
from app.pgvector_search import vector_search
results = await vector_search(query, kb_id, top_k=5)
"""
import os
import logging
import asyncio
from typing import List, Dict, Optional
from datetime import datetime
logger = logging.getLogger("aiflow.pgvector")
# PostgreSQL 连接配置
PG_HOST = os.environ.get("POSTGRES_HOST", "qihu-aiflow-pgvector")
PG_PORT = int(os.environ.get("POSTGRES_PORT", "5432"))
PG_USER = os.environ.get("POSTGRES_USER", "aiflow")
PG_PASSWORD = os.environ.get("POSTGRES_PASSWORD", "pwd2026")
PG_DB = os.environ.get("POSTGRES_DB", "aiflow_vectors")
# 嵌入模型配置
EMBEDDING_API_URL = os.environ.get("SILICONFLOW_BASE_URL", "https://api.siliconflow.cn")
EMBEDDING_API_KEY = os.environ.get("SILICONFLOW_API_KEY", "")
EMBEDDING_MODEL = os.environ.get("EMBEDDING_MODEL", "BAAI/bge-large-zh-v1.5")
# ---- 本地嵌入缓存(SQLite 持久化):破解"每次嵌入消耗外部 API 额度"的隐性成本障碍 ----
import sqlite3, hashlib, json as _json
_EMBED_CACHE_DB = os.environ.get("EMBED_CACHE_DB", "/app/backend/python/.embed_cache.db")
def _embed_cache_get(text, model):
try:
key = hashlib.sha256(f"{model}::{text}".encode("utf-8")).hexdigest()
conn = sqlite3.connect(_EMBED_CACHE_DB, timeout=5)
row = conn.execute("SELECT vec FROM embed_cache WHERE k=?", (key,)).fetchone()
conn.close()
if row:
return _json.loads(row[0])
except Exception:
return None
return None
def _embed_cache_put(text, model, vec):
try:
key = hashlib.sha256(f"{model}::{text}".encode("utf-8")).hexdigest()
conn = sqlite3.connect(_EMBED_CACHE_DB, timeout=5)
conn.execute("CREATE TABLE IF NOT EXISTS embed_cache (k TEXT PRIMARY KEY, vec TEXT, ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP)")
conn.execute("INSERT OR REPLACE INTO embed_cache (k, vec) VALUES (?, ?)", (key, _json.dumps(vec)))
conn.commit()
conn.close()
except Exception:
pass
# 连接池
_pg_pool = None
_pool_lock = asyncio.Lock()
async def get_pg_pool():
"""获取 PostgreSQL 异步连接池"""
global _pg_pool
if _pg_pool is not None:
return _pg_pool
async with _pool_lock:
if _pg_pool is not None:
return _pg_pool
try:
import asyncpg
except ImportError:
logger.warning("asyncpg not installed, vector search disabled")
return None
try:
_pg_pool = await asyncpg.create_pool(
host=PG_HOST,
port=PG_PORT,
user=PG_USER,
password=PG_PASSWORD,
database=PG_DB,
min_size=2,
max_size=10,
command_timeout=30,
)
logger.info(f"PostgreSQL connection pool created: {PG_HOST}:{PG_PORT}/{PG_DB}")
# 初始化表结构
async with _pg_pool.acquire() as conn:
await conn.execute("""
CREATE TABLE IF NOT EXISTS document_embeddings (
id SERIAL PRIMARY KEY,
doc_id VARCHAR(64) NOT NULL,
kb_id VARCHAR(64) NOT NULL,
chunk_id VARCHAR(64),
content TEXT NOT NULL,
embedding vector(1024),
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_embeddings_kb
ON document_embeddings(kb_id);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_embeddings_doc
ON document_embeddings(doc_id);
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS idx_embeddings_vector
ON document_embeddings USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
""")
logger.info("PGVector tables initialized")
return _pg_pool
except Exception as e:
logger.error(f"Failed to create PostgreSQL pool: {e}")
_pg_pool = None
return None
async def get_embedding(text: str) -> Optional[List[float]]:
"""调用嵌入API获取文本向量"""
if not EMBEDDING_API_KEY:
logger.warning("No embedding API key configured")
return None
cached = _embed_cache_get(text, EMBEDDING_MODEL)
if cached:
return cached
try:
# 收敛到共享标准模型层(app.core.llm_client.aembed),与聊天共用代理/超时/failover 语义
from app.core.llm_client import aembed
embedding = await aembed(
EMBEDDING_API_URL, EMBEDDING_API_KEY, text[:8000], EMBEDDING_MODEL, timeout=30
)
if embedding:
_embed_cache_put(text, EMBEDDING_MODEL, embedding)
return embedding
logger.error("Embedding API returned empty")
return None
except Exception as e:
logger.error(f"Embedding generation failed: {e}")
return None
async def vector_search(
query: str,
kb_id: Optional[str] = None,
top_k: int = 5,
threshold: float = 0.5,
) -> List[Dict]:
"""
向量语义检索
Args:
query: 查询文本
kb_id: 知识库ID (可选,不指定则搜索全部)
top_k: 返回结果数
threshold: 相似度阈值 (0-1)
Returns:
[{"doc_id": ..., "content": ..., "score": ..., "metadata": ...}, ...]
"""
pool = await get_pg_pool()
if pool is None:
return []
# 获取查询向量
query_embedding = await get_embedding(query)
if query_embedding is None:
return []
try:
async with pool.acquire() as conn:
# 将向量转换为 PostgreSQL 格式
embedding_str = f"[{','.join(str(x) for x in query_embedding)}]"
if kb_id:
rows = await conn.fetch(
"""
SELECT doc_id, chunk_id, content, metadata,
1 - (embedding <=> $1::vector) as score
FROM document_embeddings
WHERE kb_id = $2
ORDER BY embedding <=> $1::vector
LIMIT $3;
""",
embedding_str, kb_id, top_k,
)
else:
rows = await conn.fetch(
"""
SELECT doc_id, chunk_id, content, metadata,
1 - (embedding <=> $1::vector) as score
FROM document_embeddings
ORDER BY embedding <=> $1::vector
LIMIT $2;
""",
embedding_str, top_k,
)
results = []
for row in rows:
score = float(row["score"])
if score >= threshold:
results.append({
"doc_id": row["doc_id"],
"chunk_id": row["chunk_id"],
"content": row["content"],
"score": score,
"metadata": row["metadata"] if isinstance(row["metadata"], dict) else {},
})
return results
except Exception as e:
logger.error(f"Vector search failed: {e}")
return []
async def index_document(
doc_id: str,
kb_id: str,
content: str,
chunk_id: Optional[str] = None,
metadata: Optional[Dict] = None,
) -> bool:
"""
将文档内容索引到 PGVector
Args:
doc_id: 文档ID
kb_id: 知识库ID
content: 文档内容
chunk_id: 分块ID (可选)
metadata: 元数据
Returns:
是否成功
"""
pool = await get_pg_pool()
if pool is None:
return False
# 分块处理(简单按长度分块)
chunks = []
chunk_size = 500
overlap = 50
if len(content) <= chunk_size:
chunks = [(content, chunk_id or f"{doc_id}_0")]
else:
for i in range(0, len(content), chunk_size - overlap):
chunk = content[i:i + chunk_size]
if len(chunk.strip()) > 10:
chunks.append((chunk, f"{doc_id}_{i // (chunk_size - overlap)}"))
success_count = 0
for chunk_text, cid in chunks:
embedding = await get_embedding(chunk_text)
if embedding is None:
continue
try:
async with pool.acquire() as conn:
embedding_str = f"[{','.join(str(x) for x in embedding)}]"
import json
await conn.execute(
"""
INSERT INTO document_embeddings (doc_id, kb_id, chunk_id, content, embedding, metadata)
VALUES ($1, $2, $3, $4, $5::vector, $6)
ON CONFLICT DO NOTHING;
""",
doc_id, kb_id, cid, chunk_text, embedding_str,
json.dumps(metadata or {}),
)
success_count += 1
except Exception as e:
logger.error(f"Failed to index chunk {cid}: {e}")
logger.info(f"Indexed document {doc_id}: {success_count}/{len(chunks)} chunks")
return success_count > 0
async def hybrid_search(
query: str,
kb_id: Optional[str] = None,
top_k: int = 5,
bm25_results: Optional[List[Dict]] = None,
) -> List[Dict]:
"""
混合检索:BM25 + 向量检索融合
Args:
query: 查询文本
kb_id: 知识库ID
top_k: 返回结果数
bm25_results: BM25 检索结果(来自MySQL)
Returns:
融合后的检索结果
"""
# 获取向量检索结果
vec_results = await vector_search(query, kb_id, top_k)
if not vec_results and not bm25_results:
return []
if not vec_results:
return bm25_results or []
if not bm25_results:
return vec_results
# 融合:Reciprocal Rank Fusion (RRF)
rrf_k = 60
score_map = {}
for rank, item in enumerate(bm25_results):
doc_id = item.get("doc_id", item.get("id", ""))
score_map[doc_id] = {
"content": item.get("content", item.get("text", "")),
"score": 1.0 / (rrf_k + rank + 1),
"bm25_rank": rank + 1,
"metadata": item.get("metadata", {}),
}
for rank, item in enumerate(vec_results):
doc_id = item["doc_id"]
if doc_id in score_map:
score_map[doc_id]["score"] += 1.0 / (rrf_k + rank + 1)
score_map[doc_id]["vec_score"] = item["score"]
score_map[doc_id]["vec_rank"] = rank + 1
else:
score_map[doc_id] = {
"content": item["content"],
"score": 1.0 / (rrf_k + rank + 1),
"vec_score": item["score"],
"vec_rank": rank + 1,
"metadata": item.get("metadata", {}),
}
# 按融合分数排序
merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True)
return merged[:top_k]
async def delete_document_embeddings(doc_id: str) -> bool:
"""删除文档的所有向量索引"""
pool = await get_pg_pool()
if pool is None:
return False
try:
async with pool.acquire() as conn:
await conn.execute(
"DELETE FROM document_embeddings WHERE doc_id = $1;",
doc_id,
)
logger.info(f"Deleted embeddings for document {doc_id}")
return True
except Exception as e:
logger.error(f"Failed to delete embeddings: {e}")
return False
async def close_pool():
"""关闭连接池"""
global _pg_pool
if _pg_pool is not None:
await _pg_pool.close()
_pg_pool = None
logger.info("PostgreSQL connection pool closed")
+249 -249
View File
@@ -1,250 +1,250 @@
"""
Qihu-AIFlow FastAPI 主应用
完整后端API,监听18080端口,所有接口以 /app 为前缀
"""
import sys
import os
import time
import uuid
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.api import auth, chat, knowledge, market, token_usage, payment, profile, notifications, assets, evolution
from app.casdoor_auth import router as casdoor_router
from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin, observability_admin
from app.api.im_bot_user import router as im_bot_user_router
from app.api.im_bot_platform import router as im_bot_platform_router
from app.api.portal import user_portal
from app.api.admin.developer_tools_admin import router as developer_tools_router
from app.api.admin.mcp_market_admin import router as mcp_market_router
from app.api.admin import asset_admin, achievement_admin
from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD
from app.core.observability import set_trace_id
from app.database import SessionLocal
from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel
# 导入平台适配器触发注册(AstrBot借鉴优化方案 P1)
import app.platform_adapters # noqa: F401 - 触发所有平台适配器装饰器注册
# 初始化文件日志(按天轮转)
access_logger = setup_file_logging()
logger = logging.getLogger("aiflow")
class APITrackingMiddleware(BaseHTTPMiddleware):
"""API调用统计中间件 - 记录请求耗时,标记慢请求"""
async def dispatch(self, request: Request, call_next):
request_id = str(uuid.uuid4())
set_trace_id(request_id)
start_time = time.time()
path = request.url.path
method = request.method
# 跳过健康检查和静态资源
skip_paths = ["/app/api/health", "/favicon.ico"]
should_skip = any(path.startswith(p) for p in skip_paths)
try:
response = await call_next(request)
elapsed = time.time() - start_time
status_code = response.status_code
response.headers["X-Trace-Id"] = request_id
if not should_skip:
access_logger.info(
f"[{request_id}] {method} {path} -> {status_code} ({elapsed:.3f}s)"
)
# 异步记录API调用统计
try:
db = SessionLocal()
try:
_record_api_call(db, method, path, status_code, elapsed)
finally:
db.close()
except Exception as e:
logger.warning(f"API统计记录失败: {e}")
return response
except Exception as e:
elapsed = time.time() - start_time
if not should_skip:
access_logger.info(
f"[{request_id}] {method} {path} -> ERROR ({elapsed:.3f}s): {str(e)}"
)
raise
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
await super().__call__(scope, receive, send)
def _record_api_call(db, method: str, path: str, status_code: int, elapsed: float):
"""记录API调用到数据库统计表"""
# 简化路径以减少基数(去掉动态ID)
parts = path.split("/")
simplified_parts = []
for part in parts:
# 将UUID样式的路径段替换为{id}
if len(part) == 36 and part.count("-") == 4:
simplified_parts.append("{id}")
# 将纯数字路径段替换为{id}
elif part.isdigit():
simplified_parts.append("{id}")
else:
simplified_parts.append(part)
simplified_path = "/".join(simplified_parts)
# 统计记录通过数据库实现,简化版记录到日志
is_error = status_code >= 400
is_slow = elapsed >= SLOW_REQUEST_THRESHOLD
if is_slow:
logger.warning(
f"慢请求: {method} {simplified_path} -> {status_code} ({elapsed:.3f}s, 阈值: {SLOW_REQUEST_THRESHOLD}s)"
)
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Qihu-AIFlow v2.0.0 启动中...")
# 启动后台 Evolution Pipeline(知识库图谱已存在的用户配置自动恢复运行)
try:
from app.services.evolution_service import EvolutionWatcherManager
from app.models import UserEvolutionConfig
db = SessionLocal()
try:
configs = db.query(UserEvolutionConfig).filter(
UserEvolutionConfig.watcher_status == "running"
).all()
if configs:
watcher = EvolutionWatcherManager()
for cfg in configs:
try:
watcher.start_user_evolution(cfg.user_id, lambda: SessionLocal())
logger.info(f"[Evolution] 恢复用户 {cfg.user_id} 的进化Pipeline")
except Exception as e:
logger.warning(f"[Evolution] 恢复用户 {cfg.user_id} 失败: {e}")
logger.info(f"[Evolution] 已恢复 {len(configs)} 个用户的进化Pipeline")
else:
logger.info("[Evolution] 无待恢复的进化配置")
finally:
db.close()
except Exception as e:
logger.warning(f"[Evolution] Pipeline自动启动异常(不影响正常服务): {e}")
# 启动Redis事件总线
try:
from app.event_bus import get_event_bus
event_bus = get_event_bus()
event_bus.start(num_consumers=2)
logger.info("[EventBus] Redis事件总线已启动")
except Exception as e:
logger.warning(f"[EventBus] 启动失败(不影响正常服务): {e}")
logger.info("Qihu-AIFlow v2.0.0 启动完成")
yield
# 停止Redis事件总线
try:
from app.event_bus import get_event_bus
get_event_bus().stop()
logger.info("[EventBus] Redis事件总线已停止")
except Exception:
pass
# 优雅关闭:停止所有 Evolution Pipeline
try:
from app.services.evolution_service import EvolutionWatcherManager
EvolutionWatcherManager().shutdown_all()
logger.info("[Evolution] 所有Pipeline已停止")
except Exception:
pass
logger.info("Qihu-AIFlow 正在关闭...")
app = FastAPI(
title="Qihu-AIFlow - 智能自动化PaaS平台",
version="1.0.0",
lifespan=lifespan,
)
# API调用追踪中间件
app.add_middleware(APITrackingMiddleware)
# CORS配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 前台 API
app.include_router(auth.router, prefix="/app/api/auth", tags=["认证"])
app.include_router(chat.router, prefix="/app/api/chat", tags=["AI聊天"])
app.include_router(knowledge.router, prefix="/app/api/knowledge", tags=["知识库"])
app.include_router(market.router, prefix="/app/api/market", tags=["应用市场"])
app.include_router(token_usage.router, prefix="/app/api/token", tags=["Token驾驶舱"])
app.include_router(payment.router, prefix="/app/api/payment", tags=["付费中心"])
app.include_router(profile.router, prefix="/app/api/profile", tags=["个人中心"])
app.include_router(notifications.router, prefix="/app/api/notifications", tags=["消息通知"])
app.include_router(assets.router, prefix="/app/api/assets", tags=["资产管理"])
app.include_router(evolution.router, prefix="/app/api/evolution", tags=["AI自进化"])
# 后台 API
app.include_router(dashboard.router, prefix="/app/api/admin/dashboard", tags=["后台-仪表盘"])
app.include_router(users.router, prefix="/app/api/admin/users", tags=["后台-用户管理"])
app.include_router(models.router, prefix="/app/api/admin/models", tags=["后台-模型管理"])
app.include_router(settings.router, prefix="/app/api/admin/settings", tags=["后台-网站设置"])
app.include_router(payment_admin.router, prefix="/app/api/admin/payment", tags=["后台-支付配置"])
app.include_router(services.router, prefix="/app/api/admin/services", tags=["后台-增值服务"])
app.include_router(membership_admin.router, prefix="/app/api/admin/membership", tags=["后台-会员管理"])
app.include_router(ads.router, prefix="/app/api/admin/ads", tags=["后台-广告管理"])
app.include_router(market_admin.router, prefix="/app/api/admin/market", tags=["后台-应用审核"])
app.include_router(logs.router, prefix="/app/api/admin/logs", tags=["后台-日志风控"])
app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=["后台-技能管理"])
app.include_router(agents_admin.router, prefix="/app/api/admin/agents", tags=["后台-Agent管理"])
app.include_router(knowledge_admin.router, prefix="/app/api/admin/knowledge", tags=["后台-知识库管理"])
app.include_router(workflow_admin.router, prefix="/app/api/admin/workflows", tags=["后台-工作流编排"])
app.include_router(asset_admin.router, prefix="/app/api/admin/assets", tags=["后台-资产管理"])
app.include_router(storage_admin.router, prefix="/app/api/admin/storage", tags=["后台-存储配置"])
app.include_router(automation_admin.router, prefix="/app/api/admin/automation", tags=["后台-自动化任务"])
app.include_router(mcp_admin.router, prefix="/app/api/admin/mcp", tags=["后台-MCP服务"])
app.include_router(developer_admin.router, prefix="/app/api/admin/developer", tags=["后台-开发者API"])
app.include_router(developer_tools_router, prefix="/app/api/admin/developer-tools", tags=["dev-tools"])
app.include_router(apps_admin.router, prefix="/app/api/admin/apps", tags=["admin-ai-apps"])
app.include_router(mcp_market_router, prefix="/app/api/admin/mcp-market", tags=["mcp-market"])
app.include_router(code_exec_admin.router, prefix="/app/api/admin/code-exec", tags=["code-exec"])
app.include_router(distro_admin.router, prefix="/app/api/admin/distributions", tags=["distributions"])
app.include_router(privacy_admin.router, prefix="/app/api/admin/privacy", tags=["privacy"])
app.include_router(im_bot_admin.router, prefix="/app/api/admin/im-bot", tags=["im-bot"])
app.include_router(im_bot_user_router, prefix="/app/api/im/wechat", tags=["im-wechat-user"])
app.include_router(im_bot_platform_router, prefix="/app/api/im/platform", tags=["im-platform-user"])
app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", tags=["promotion"])
app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"])
app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"])
app.include_router(analytics_admin.router, prefix="/app/api/admin/analytics", tags=["analytics"])
app.include_router(observability_admin.router, prefix="/app/api/admin/observability", tags=["后台-调用可观测"])
app.include_router(user_portal.router, prefix="/app/api/portal", tags=["portal"])
app.include_router(achievement_admin.router, prefix="/app/api/admin/achievement", tags=["achievement"])
app.include_router(casdoor_router, prefix="/app/api/auth", tags=["Casdoor SSO"])
@app.get("/app/api/health")
async def health_check():
"""
Qihu-AIFlow FastAPI 主应用
完整后端API,监听18080端口,所有接口以 /app 为前缀
"""
import sys
import os
import time
import uuid
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.api import auth, chat, knowledge, market, token_usage, payment, profile, notifications, assets, evolution
from app.casdoor_auth import router as casdoor_router
from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin, observability_admin
from app.api.im_bot_user import router as im_bot_user_router
from app.api.im_bot_platform import router as im_bot_platform_router
from app.api.portal import user_portal
from app.api.admin.developer_tools_admin import router as developer_tools_router
from app.api.admin.mcp_market_admin import router as mcp_market_router
from app.api.admin import asset_admin, achievement_admin
from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD
from app.core.observability import set_trace_id
from app.database import SessionLocal
from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel
# 导入平台适配器触发注册(AstrBot借鉴优化方案 P1)
import app.platform_adapters # noqa: F401 - 触发所有平台适配器装饰器注册
# 初始化文件日志(按天轮转)
access_logger = setup_file_logging()
logger = logging.getLogger("aiflow")
class APITrackingMiddleware(BaseHTTPMiddleware):
"""API调用统计中间件 - 记录请求耗时,标记慢请求"""
async def dispatch(self, request: Request, call_next):
request_id = str(uuid.uuid4())
set_trace_id(request_id)
start_time = time.time()
path = request.url.path
method = request.method
# 跳过健康检查和静态资源
skip_paths = ["/app/api/health", "/favicon.ico"]
should_skip = any(path.startswith(p) for p in skip_paths)
try:
response = await call_next(request)
elapsed = time.time() - start_time
status_code = response.status_code
response.headers["X-Trace-Id"] = request_id
if not should_skip:
access_logger.info(
f"[{request_id}] {method} {path} -> {status_code} ({elapsed:.3f}s)"
)
# 异步记录API调用统计
try:
db = SessionLocal()
try:
_record_api_call(db, method, path, status_code, elapsed)
finally:
db.close()
except Exception as e:
logger.warning(f"API统计记录失败: {e}")
return response
except Exception as e:
elapsed = time.time() - start_time
if not should_skip:
access_logger.info(
f"[{request_id}] {method} {path} -> ERROR ({elapsed:.3f}s): {str(e)}"
)
raise
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
await super().__call__(scope, receive, send)
def _record_api_call(db, method: str, path: str, status_code: int, elapsed: float):
"""记录API调用到数据库统计表"""
# 简化路径以减少基数(去掉动态ID)
parts = path.split("/")
simplified_parts = []
for part in parts:
# 将UUID样式的路径段替换为{id}
if len(part) == 36 and part.count("-") == 4:
simplified_parts.append("{id}")
# 将纯数字路径段替换为{id}
elif part.isdigit():
simplified_parts.append("{id}")
else:
simplified_parts.append(part)
simplified_path = "/".join(simplified_parts)
# 统计记录通过数据库实现,简化版记录到日志
is_error = status_code >= 400
is_slow = elapsed >= SLOW_REQUEST_THRESHOLD
if is_slow:
logger.warning(
f"慢请求: {method} {simplified_path} -> {status_code} ({elapsed:.3f}s, 阈值: {SLOW_REQUEST_THRESHOLD}s)"
)
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Qihu-AIFlow v2.0.0 启动中...")
# 启动后台 Evolution Pipeline(知识库图谱已存在的用户配置自动恢复运行)
try:
from app.services.evolution_service import EvolutionWatcherManager
from app.models import UserEvolutionConfig
db = SessionLocal()
try:
configs = db.query(UserEvolutionConfig).filter(
UserEvolutionConfig.watcher_status == "running"
).all()
if configs:
watcher = EvolutionWatcherManager()
for cfg in configs:
try:
watcher.start_user_evolution(cfg.user_id, lambda: SessionLocal())
logger.info(f"[Evolution] 恢复用户 {cfg.user_id} 的进化Pipeline")
except Exception as e:
logger.warning(f"[Evolution] 恢复用户 {cfg.user_id} 失败: {e}")
logger.info(f"[Evolution] 已恢复 {len(configs)} 个用户的进化Pipeline")
else:
logger.info("[Evolution] 无待恢复的进化配置")
finally:
db.close()
except Exception as e:
logger.warning(f"[Evolution] Pipeline自动启动异常(不影响正常服务): {e}")
# 启动Redis事件总线
try:
from app.event_bus import get_event_bus
event_bus = get_event_bus()
event_bus.start(num_consumers=2)
logger.info("[EventBus] Redis事件总线已启动")
except Exception as e:
logger.warning(f"[EventBus] 启动失败(不影响正常服务): {e}")
logger.info("Qihu-AIFlow v2.0.0 启动完成")
yield
# 停止Redis事件总线
try:
from app.event_bus import get_event_bus
get_event_bus().stop()
logger.info("[EventBus] Redis事件总线已停止")
except Exception:
pass
# 优雅关闭:停止所有 Evolution Pipeline
try:
from app.services.evolution_service import EvolutionWatcherManager
EvolutionWatcherManager().shutdown_all()
logger.info("[Evolution] 所有Pipeline已停止")
except Exception:
pass
logger.info("Qihu-AIFlow 正在关闭...")
app = FastAPI(
title="Qihu-AIFlow - 智能自动化PaaS平台",
version="1.0.0",
lifespan=lifespan,
)
# API调用追踪中间件
app.add_middleware(APITrackingMiddleware)
# CORS配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 前台 API
app.include_router(auth.router, prefix="/app/api/auth", tags=["认证"])
app.include_router(chat.router, prefix="/app/api/chat", tags=["AI聊天"])
app.include_router(knowledge.router, prefix="/app/api/knowledge", tags=["知识库"])
app.include_router(market.router, prefix="/app/api/market", tags=["应用市场"])
app.include_router(token_usage.router, prefix="/app/api/token", tags=["Token驾驶舱"])
app.include_router(payment.router, prefix="/app/api/payment", tags=["付费中心"])
app.include_router(profile.router, prefix="/app/api/profile", tags=["个人中心"])
app.include_router(notifications.router, prefix="/app/api/notifications", tags=["消息通知"])
app.include_router(assets.router, prefix="/app/api/assets", tags=["资产管理"])
app.include_router(evolution.router, prefix="/app/api/evolution", tags=["AI自进化"])
# 后台 API
app.include_router(dashboard.router, prefix="/app/api/admin/dashboard", tags=["后台-仪表盘"])
app.include_router(users.router, prefix="/app/api/admin/users", tags=["后台-用户管理"])
app.include_router(models.router, prefix="/app/api/admin/models", tags=["后台-模型管理"])
app.include_router(settings.router, prefix="/app/api/admin/settings", tags=["后台-网站设置"])
app.include_router(payment_admin.router, prefix="/app/api/admin/payment", tags=["后台-支付配置"])
app.include_router(services.router, prefix="/app/api/admin/services", tags=["后台-增值服务"])
app.include_router(membership_admin.router, prefix="/app/api/admin/membership", tags=["后台-会员管理"])
app.include_router(ads.router, prefix="/app/api/admin/ads", tags=["后台-广告管理"])
app.include_router(market_admin.router, prefix="/app/api/admin/market", tags=["后台-应用审核"])
app.include_router(logs.router, prefix="/app/api/admin/logs", tags=["后台-日志风控"])
app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=["后台-技能管理"])
app.include_router(agents_admin.router, prefix="/app/api/admin/agents", tags=["后台-Agent管理"])
app.include_router(knowledge_admin.router, prefix="/app/api/admin/knowledge", tags=["后台-知识库管理"])
app.include_router(workflow_admin.router, prefix="/app/api/admin/workflows", tags=["后台-工作流编排"])
app.include_router(asset_admin.router, prefix="/app/api/admin/assets", tags=["后台-资产管理"])
app.include_router(storage_admin.router, prefix="/app/api/admin/storage", tags=["后台-存储配置"])
app.include_router(automation_admin.router, prefix="/app/api/admin/automation", tags=["后台-自动化任务"])
app.include_router(mcp_admin.router, prefix="/app/api/admin/mcp", tags=["后台-MCP服务"])
app.include_router(developer_admin.router, prefix="/app/api/admin/developer", tags=["后台-开发者API"])
app.include_router(developer_tools_router, prefix="/app/api/admin/developer-tools", tags=["dev-tools"])
app.include_router(apps_admin.router, prefix="/app/api/admin/apps", tags=["admin-ai-apps"])
app.include_router(mcp_market_router, prefix="/app/api/admin/mcp-market", tags=["mcp-market"])
app.include_router(code_exec_admin.router, prefix="/app/api/admin/code-exec", tags=["code-exec"])
app.include_router(distro_admin.router, prefix="/app/api/admin/distributions", tags=["distributions"])
app.include_router(privacy_admin.router, prefix="/app/api/admin/privacy", tags=["privacy"])
app.include_router(im_bot_admin.router, prefix="/app/api/admin/im-bot", tags=["im-bot"])
app.include_router(im_bot_user_router, prefix="/app/api/im/wechat", tags=["im-wechat-user"])
app.include_router(im_bot_platform_router, prefix="/app/api/im/platform", tags=["im-platform-user"])
app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", tags=["promotion"])
app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"])
app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"])
app.include_router(analytics_admin.router, prefix="/app/api/admin/analytics", tags=["analytics"])
app.include_router(observability_admin.router, prefix="/app/api/admin/observability", tags=["后台-调用可观测"])
app.include_router(user_portal.router, prefix="/app/api/portal", tags=["portal"])
app.include_router(achievement_admin.router, prefix="/app/api/admin/achievement", tags=["achievement"])
app.include_router(casdoor_router, prefix="/app/api/auth", tags=["Casdoor SSO"])
@app.get("/app/api/health")
async def health_check():
return {"status": "ok", "service": "Qihu-AIFlow"}
+30 -30
View File
@@ -1,30 +1,30 @@
# AIFlow 后端依赖
fastapi==0.115.6
uvicorn==0.34.0
sqlalchemy==2.0.36
pymysql==1.1.1
cryptography==44.0.0
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0
python-multipart==0.0.18
redis==5.2.1
pydantic==2.10.3
pydantic-settings==2.7.0
python-dotenv==1.0.1
httpx==0.28.1
aiofiles==24.1.0
celery==5.4.0
qrcode==8.2
pillow==12.2.0
xskill==0.5.2
# ---- pinned celery ecosystem (known-good, extracted from running container) ----
# 避免 pip 浮动解析到与 celery 5.4.0 冲突的新版传递依赖(如 click-repl 0.2.0 的 prompt-toolkit 冲突)
amqp==5.3.1
kombu==5.6.2
billiard==4.2.4
vine==5.1.0
click==8.4.2
click-didyoumean==0.3.1
click-plugins==1.1.1.2
click-repl==0.3.0
# AIFlow 后端依赖
fastapi==0.115.6
uvicorn==0.34.0
sqlalchemy==2.0.36
pymysql==1.1.1
cryptography==44.0.0
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0
python-multipart==0.0.18
redis==5.2.1
pydantic==2.10.3
pydantic-settings==2.7.0
python-dotenv==1.0.1
httpx==0.28.1
aiofiles==24.1.0
celery==5.4.0
qrcode==8.2
pillow==12.2.0
xskill==0.5.2
# ---- pinned celery ecosystem (known-good, extracted from running container) ----
# 避免 pip 浮动解析到与 celery 5.4.0 冲突的新版传递依赖(如 click-repl 0.2.0 的 prompt-toolkit 冲突)
amqp==5.3.1
kombu==5.6.2
billiard==4.2.4
vine==5.1.0
click==8.4.2
click-didyoumean==0.3.1
click-plugins==1.1.1.2
click-repl==0.3.0