439 lines
18 KiB
Python
439 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable
|
|
|
|
import httpx
|
|
from sqlalchemy import and_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ....models import Integration
|
|
from ..schemas import ExternalAdapterResult, ExternalEvidenceItem, ExternalQueryPlan
|
|
|
|
|
|
PLATFORM_RUNTIME_TENANT_ID = int(os.getenv("PMFLOW_PLATFORM_RUNTIME_TENANT_ID", "1") or "1")
|
|
|
|
|
|
class DashScopeResponsesSearchAdapter:
|
|
adapter_name = "dashscope_responses"
|
|
supported_query_kinds = {"weather", "news", "market", "company", "model", "generic_web", "unknown"}
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
api_key: str,
|
|
base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1",
|
|
model: str = "qwen3.6-plus",
|
|
client_factory: Callable[[], httpx.AsyncClient] | None = None,
|
|
tenant_id: int | None = None,
|
|
user_id: int | None = None,
|
|
timeout_seconds: float | None = None,
|
|
routing_meta: dict[str, Any] | None = None,
|
|
) -> None:
|
|
self.api_key = str(api_key or "").strip()
|
|
self.base_url = str(base_url or "https://dashscope.aliyuncs.com/compatible-mode/v1").strip()
|
|
self.model = str(model or "qwen3.6-plus").strip()
|
|
self.timeout_seconds = self._normalize_timeout(timeout_seconds)
|
|
self.client_factory = client_factory or (
|
|
lambda: httpx.AsyncClient(timeout=self.timeout_seconds, trust_env=False, follow_redirects=True)
|
|
)
|
|
self.tenant_id = int(tenant_id or PLATFORM_RUNTIME_TENANT_ID)
|
|
self.user_id = int(user_id) if user_id is not None else None
|
|
self.routing_meta = dict(routing_meta or {}) if isinstance(routing_meta, dict) else {}
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
return bool(self.api_key and self.model)
|
|
|
|
async def execute(self, plan: ExternalQueryPlan) -> ExternalAdapterResult:
|
|
if not self.available:
|
|
return ExternalAdapterResult(
|
|
adapter_name=self.adapter_name,
|
|
status="unavailable",
|
|
evidence=[],
|
|
meta={"reason": "dashscope_responses_runtime_missing"},
|
|
)
|
|
query = self._build_input(plan)
|
|
if not query:
|
|
return ExternalAdapterResult(
|
|
adapter_name=self.adapter_name,
|
|
status="empty",
|
|
evidence=[],
|
|
meta={"reason": "empty_query"},
|
|
)
|
|
|
|
endpoint = f"{self.base_url.rstrip('/')}/responses"
|
|
payload = {
|
|
"model": self.model,
|
|
"input": query,
|
|
"tools": [{"type": "web_search"}],
|
|
"search_options": {"forced_search": True},
|
|
"enable_thinking": True,
|
|
}
|
|
if plan.freshness_required:
|
|
freshness_days = self._freshness_days(plan)
|
|
if freshness_days:
|
|
payload["search_options"]["freshness"] = freshness_days
|
|
|
|
response_payload: dict[str, Any] | None = None
|
|
status = "error"
|
|
error_summary = ""
|
|
started_at = time.perf_counter()
|
|
try:
|
|
async with self.client_factory() as client:
|
|
response = await client.post(
|
|
endpoint,
|
|
headers={
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json=payload,
|
|
)
|
|
try:
|
|
response_payload = response.json()
|
|
except Exception:
|
|
response_payload = {"raw_text": str(response.text or "")}
|
|
response.raise_for_status()
|
|
evidence = self._evidence_from_response(plan=plan, payload=response_payload, query=query)
|
|
status = "success" if evidence else "empty"
|
|
return ExternalAdapterResult(
|
|
adapter_name=self.adapter_name,
|
|
status=status,
|
|
evidence=evidence[: max(1, int(plan.result_limit or 3))],
|
|
meta={
|
|
"query": query,
|
|
"provider": self.adapter_name,
|
|
"model": self.model,
|
|
"search_result_count": len(self._extract_search_results(response_payload)),
|
|
"tool_usage": self._extract_tool_usage(response_payload),
|
|
"latency_ms": int((time.perf_counter() - started_at) * 1000),
|
|
"timeout_seconds": self.timeout_seconds,
|
|
"fetched_pages": 0,
|
|
},
|
|
)
|
|
except Exception as exc:
|
|
error_summary = f"{type(exc).__name__}: {exc!s}".strip()[:500]
|
|
if error_summary.endswith(":"):
|
|
error_summary = repr(exc)[:500]
|
|
return ExternalAdapterResult(
|
|
adapter_name=self.adapter_name,
|
|
status="error",
|
|
evidence=[],
|
|
meta={
|
|
"query": query,
|
|
"provider": self.adapter_name,
|
|
"model": self.model,
|
|
"reason": "dashscope_responses_request_failed",
|
|
"error": error_summary,
|
|
"latency_ms": int((time.perf_counter() - started_at) * 1000),
|
|
"timeout_seconds": self.timeout_seconds,
|
|
},
|
|
)
|
|
finally:
|
|
await self._log_ai_call(
|
|
endpoint=endpoint,
|
|
payload=payload,
|
|
response_payload=response_payload,
|
|
status=status,
|
|
error_summary=error_summary,
|
|
latency_ms=int((time.perf_counter() - started_at) * 1000),
|
|
)
|
|
|
|
@staticmethod
|
|
def _build_input(plan: ExternalQueryPlan) -> str:
|
|
queries = [str(item or "").strip() for item in list(plan.search_queries or []) if str(item or "").strip()]
|
|
topic = str(plan.topic or "").strip()
|
|
primary = queries[0] if queries else topic
|
|
if not primary:
|
|
return ""
|
|
context_bits = []
|
|
if plan.freshness_required:
|
|
context_bits.append("优先检索最近公开来源。")
|
|
if plan.requires_multi_source:
|
|
context_bits.append("尽量覆盖多个独立公开来源。")
|
|
if plan.subject_entities:
|
|
context_bits.append("关注对象：" + "、".join(str(item) for item in plan.subject_entities[:5]))
|
|
suffix = "\n".join(context_bits)
|
|
return f"{primary}\n{suffix}".strip()
|
|
|
|
@staticmethod
|
|
def _freshness_days(plan: ExternalQueryPlan) -> int | None:
|
|
hint = str(plan.time_range_hint or "").lower()
|
|
if "24" in hint or "today" in hint or "今天" in hint:
|
|
return 1
|
|
if "week" in hint or "7" in hint or "一周" in hint or "最近" in hint:
|
|
return 7
|
|
if "month" in hint or "30" in hint or "一个月" in hint:
|
|
return 30
|
|
return 30
|
|
|
|
@staticmethod
|
|
def _normalize_timeout(value: float | None) -> float:
|
|
raw = value
|
|
if raw is None:
|
|
env_value = str(os.getenv("DASHSCOPE_RESPONSES_TIMEOUT_SECONDS") or "").strip()
|
|
try:
|
|
raw = float(env_value) if env_value else 120.0
|
|
except Exception:
|
|
raw = 120.0
|
|
try:
|
|
normalized = float(raw)
|
|
except Exception:
|
|
normalized = 120.0
|
|
return max(30.0, min(normalized, 240.0))
|
|
|
|
@classmethod
|
|
def _evidence_from_response(
|
|
cls,
|
|
*,
|
|
plan: ExternalQueryPlan,
|
|
payload: dict[str, Any] | None,
|
|
query: str,
|
|
) -> list[ExternalEvidenceItem]:
|
|
data = payload if isinstance(payload, dict) else {}
|
|
search_results = cls._extract_search_results(data)
|
|
output_text = cls._extract_output_text(data)
|
|
allow_output_text_as_single_source_snippet = len(search_results) == 1
|
|
retrieved_at = datetime.now(timezone.utc).isoformat()
|
|
evidence: list[ExternalEvidenceItem] = []
|
|
for raw in search_results:
|
|
target_url = cls._first_text(raw, "url", "link", "href", "contentUrl", "site_url")
|
|
if not target_url.startswith("http://") and not target_url.startswith("https://"):
|
|
continue
|
|
title = cls._first_text(raw, "title", "name", "source_name") or target_url
|
|
snippet = cls._first_text(raw, "snippet", "summary", "content", "text", "description")
|
|
if not snippet and output_text and allow_output_text_as_single_source_snippet:
|
|
snippet = output_text
|
|
metadata = {
|
|
"provider": "dashscope_responses",
|
|
"query": query,
|
|
"result_query": query,
|
|
"model": str(data.get("model") or ""),
|
|
"published_at": cls._first_text(raw, "published_at", "date", "time"),
|
|
}
|
|
evidence.append(
|
|
ExternalEvidenceItem(
|
|
source_type="url",
|
|
source_label=title[:255],
|
|
title=title[:255],
|
|
target_url=target_url,
|
|
snippet=str(snippet or "")[:1800],
|
|
retrieved_at=retrieved_at,
|
|
freshness_score=0.86 if plan.freshness_required else 0.74,
|
|
authority_score=0.68,
|
|
grounded_fields=["dashscope_responses_web_search"],
|
|
raw_metadata={key: value for key, value in metadata.items() if value},
|
|
)
|
|
)
|
|
return evidence
|
|
|
|
@classmethod
|
|
def _extract_search_results(cls, payload: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
data = payload if isinstance(payload, dict) else {}
|
|
candidates: list[Any] = []
|
|
search_info = data.get("search_info") if isinstance(data.get("search_info"), dict) else {}
|
|
candidates.append(search_info.get("search_results"))
|
|
output = data.get("output")
|
|
if isinstance(output, dict):
|
|
nested_search_info = output.get("search_info") if isinstance(output.get("search_info"), dict) else {}
|
|
candidates.append(nested_search_info.get("search_results"))
|
|
candidates.append(output.get("search_results"))
|
|
if isinstance(output, list):
|
|
for item in output:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
item_search_info = item.get("search_info") if isinstance(item.get("search_info"), dict) else {}
|
|
candidates.append(item_search_info.get("search_results"))
|
|
candidates.append(item.get("search_results"))
|
|
if str(item.get("type") or "") == "web_search_call":
|
|
action = item.get("action") if isinstance(item.get("action"), dict) else {}
|
|
sources = action.get("sources")
|
|
if isinstance(sources, list):
|
|
candidates.append(
|
|
[
|
|
{
|
|
"url": source.get("url"),
|
|
"title": source.get("title") or source.get("url"),
|
|
"snippet": source.get("snippet") or source.get("summary") or "",
|
|
}
|
|
for source in sources
|
|
if isinstance(source, dict)
|
|
]
|
|
)
|
|
for candidate in candidates:
|
|
if isinstance(candidate, list):
|
|
rows = [dict(item) for item in candidate if isinstance(item, dict)]
|
|
if rows:
|
|
return rows
|
|
return []
|
|
|
|
@staticmethod
|
|
def _extract_tool_usage(payload: dict[str, Any] | None) -> dict[str, Any]:
|
|
data = payload if isinstance(payload, dict) else {}
|
|
usage = data.get("usage") if isinstance(data.get("usage"), dict) else {}
|
|
tools = usage.get("x_tools") if isinstance(usage.get("x_tools"), dict) else {}
|
|
if tools:
|
|
return dict(tools)
|
|
details = usage.get("x_details") if isinstance(usage.get("x_details"), list) else []
|
|
for item in details:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
plugins = item.get("plugins") if isinstance(item.get("plugins"), dict) else {}
|
|
if plugins:
|
|
return dict(plugins)
|
|
return {}
|
|
|
|
@classmethod
|
|
def _extract_output_text(cls, payload: dict[str, Any]) -> str:
|
|
direct = str(payload.get("output_text") or "").strip()
|
|
if direct:
|
|
return direct
|
|
output = payload.get("output")
|
|
texts: list[str] = []
|
|
if isinstance(output, list):
|
|
for item in output:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
content = item.get("content")
|
|
if isinstance(content, list):
|
|
for content_item in content:
|
|
if isinstance(content_item, dict):
|
|
text = str(content_item.get("text") or "").strip()
|
|
if text:
|
|
texts.append(text)
|
|
text = str(item.get("text") or "").strip()
|
|
if text:
|
|
texts.append(text)
|
|
elif isinstance(output, dict):
|
|
text = str(output.get("text") or output.get("output_text") or "").strip()
|
|
if text:
|
|
texts.append(text)
|
|
return "\n".join(texts).strip()
|
|
|
|
@staticmethod
|
|
def _first_text(raw: dict[str, Any], *keys: str) -> str:
|
|
for key in keys:
|
|
value = str(raw.get(key) or "").strip()
|
|
if value:
|
|
return value
|
|
return ""
|
|
|
|
async def _log_ai_call(
|
|
self,
|
|
*,
|
|
endpoint: str,
|
|
payload: dict[str, Any],
|
|
response_payload: dict[str, Any] | None,
|
|
status: str,
|
|
error_summary: str,
|
|
latency_ms: int | None,
|
|
) -> None:
|
|
try:
|
|
from ...llm_service import llm_service
|
|
|
|
routing_meta = {
|
|
**self.routing_meta,
|
|
"scene_key": "external_query",
|
|
"slot_key": "external_query",
|
|
"slot_provider_id": self.adapter_name,
|
|
"routing_decision": "external_query_provider_registry",
|
|
"route_source": "active_slot",
|
|
"response_api": "dashscope_responses",
|
|
"latency_ms": latency_ms,
|
|
"provider_timeout_seconds": self.timeout_seconds,
|
|
"tools": [str(item.get("type") or "") for item in list(payload.get("tools") or []) if isinstance(item, dict)],
|
|
}
|
|
await llm_service._log_ai_call(
|
|
tenant_id=self.tenant_id,
|
|
user_id=self.user_id,
|
|
function_name="external_query_dashscope_responses",
|
|
prompt=str(payload.get("input") or "")[:4000],
|
|
response=str(response_payload or error_summary or "")[:4000],
|
|
status=f"external_query_{status}",
|
|
provider="dashscope",
|
|
model=self.model,
|
|
provider_request_payload={**payload, "api_key": None},
|
|
provider_response_payload=response_payload,
|
|
provider_attempts=[
|
|
{
|
|
"provider": "dashscope",
|
|
"model": self.model,
|
|
"endpoint": endpoint,
|
|
"status": status,
|
|
"latency_ms": latency_ms,
|
|
"request_payload": {**payload, "api_key": None},
|
|
"response_payload": response_payload,
|
|
"error": error_summary or None,
|
|
}
|
|
],
|
|
routing_meta=routing_meta,
|
|
)
|
|
except Exception:
|
|
return
|
|
|
|
|
|
async def build_dashscope_responses_adapter_from_runtime(
|
|
*,
|
|
db: AsyncSession | None,
|
|
tenant_id: int | None = None,
|
|
user_id: int | None = None,
|
|
timeout_seconds: float | None = None,
|
|
routing_meta: dict[str, Any] | None = None,
|
|
) -> DashScopeResponsesSearchAdapter:
|
|
runtime = await _resolve_dashscope_runtime(db=db, tenant_id=tenant_id)
|
|
return DashScopeResponsesSearchAdapter(
|
|
api_key=str(runtime.get("api_key") or ""),
|
|
base_url=str(runtime.get("base_url") or "https://dashscope.aliyuncs.com/compatible-mode/v1"),
|
|
model=str(runtime.get("model") or "qwen3.6-plus"),
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
timeout_seconds=timeout_seconds if timeout_seconds is not None else runtime.get("timeout_seconds"),
|
|
routing_meta=routing_meta,
|
|
)
|
|
|
|
|
|
async def _resolve_dashscope_runtime(*, db: AsyncSession | None, tenant_id: int | None) -> dict[str, Any]:
|
|
env_key = str(os.getenv("DASHSCOPE_API_KEY") or "").strip()
|
|
fallback = {
|
|
"api_key": env_key,
|
|
"base_url": str(os.getenv("DASHSCOPE_BASE_URL") or "https://dashscope.aliyuncs.com/compatible-mode/v1"),
|
|
"model": str(os.getenv("DASHSCOPE_RESPONSES_MODEL") or "qwen3.6-plus"),
|
|
"timeout_seconds": str(os.getenv("DASHSCOPE_RESPONSES_TIMEOUT_SECONDS") or "120"),
|
|
}
|
|
if db is None:
|
|
return fallback
|
|
try:
|
|
row = (
|
|
await db.execute(
|
|
select(Integration).where(
|
|
and_(
|
|
Integration.tenant_id == int(tenant_id or PLATFORM_RUNTIME_TENANT_ID),
|
|
Integration.type == "llm_engine",
|
|
)
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
except Exception:
|
|
return fallback
|
|
config = row.config if row is not None and isinstance(row.config, dict) else {}
|
|
engines = [item for item in list(config.get("engines") or []) if isinstance(item, dict)]
|
|
candidates = [
|
|
item
|
|
for item in engines
|
|
if str(item.get("provider") or "").strip().lower() == "dashscope"
|
|
and bool(item.get("enabled", True))
|
|
and str(item.get("model") or "").strip().lower().startswith("qwen3.6")
|
|
and str(item.get("api_key") or "").strip()
|
|
]
|
|
if not candidates:
|
|
return fallback
|
|
selected = candidates[0]
|
|
return {
|
|
"api_key": str(selected.get("api_key") or env_key),
|
|
"base_url": str(selected.get("base_url") or fallback["base_url"]),
|
|
"model": str(selected.get("model") or fallback["model"]),
|
|
"timeout_seconds": selected.get("responses_timeout_seconds") or fallback["timeout_seconds"],
|
|
}
|