435 lines
18 KiB
Python
435 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Callable
|
||
|
||
from ...config.ai_prompts import PROMPT_EXTERNAL_QUERY_PLAN
|
||
from ...schemas.project_operating_context import ProjectFitBrief
|
||
from .schemas import ExternalQueryPlan
|
||
|
||
|
||
class ExternalQueryPlanner:
|
||
FUNCTION_NAME = "external_query_plan"
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
llm_service: Any,
|
||
db: Any,
|
||
tenant_id: int,
|
||
query_builder: Callable[[str], list[str]] | None = None,
|
||
) -> None:
|
||
self.llm_service = llm_service
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id or 1)
|
||
self.query_builder = query_builder
|
||
|
||
async def build_plan(
|
||
self,
|
||
*,
|
||
question: str,
|
||
query_semantics: dict[str, Any] | None = None,
|
||
fit_brief: dict[str, Any] | None = None,
|
||
) -> tuple[ExternalQueryPlan, dict[str, Any]]:
|
||
fallback_plan = self._build_fallback_plan(
|
||
question=question,
|
||
query_semantics=query_semantics,
|
||
fit_brief=fit_brief,
|
||
)
|
||
generator = getattr(self.llm_service, "generate_structured_object", None)
|
||
if not callable(generator):
|
||
return fallback_plan, {
|
||
"planner": "heuristic",
|
||
"fallback_taken": True,
|
||
"fit_brief_summary": fallback_plan.fit_brief_summary,
|
||
}
|
||
|
||
normalized_fit_brief = self._normalize_fit_brief(fit_brief)
|
||
prompt = PROMPT_EXTERNAL_QUERY_PLAN.format(
|
||
question=str(question or "").strip()[:1200],
|
||
query_semantics=json.dumps(query_semantics or {}, ensure_ascii=False)[:4000],
|
||
fit_brief=json.dumps(normalized_fit_brief, ensure_ascii=False)[:2500],
|
||
fallback_queries=json.dumps(fallback_plan.search_queries, ensure_ascii=False),
|
||
).strip()
|
||
try:
|
||
structured = await generator(
|
||
prompt=prompt,
|
||
db=self.db,
|
||
tenant_id=self.tenant_id,
|
||
function_name=self.FUNCTION_NAME,
|
||
use_prompt_profile=False,
|
||
temperature=0.0,
|
||
response_mode="fast",
|
||
)
|
||
payload = getattr(structured, "payload", None)
|
||
if payload is None and hasattr(structured, "model_dump"):
|
||
payload = structured.model_dump()
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("external_query_plan_payload_not_dict")
|
||
merged = fallback_plan.model_dump()
|
||
for key, value in payload.items():
|
||
if value in (None, "", [], {}):
|
||
continue
|
||
merged[key] = value
|
||
plan = ExternalQueryPlan.model_validate(merged)
|
||
if fallback_plan.freshness_required:
|
||
plan.freshness_required = True
|
||
recommendation_queries = self._build_signal_source_recommendation_queries(
|
||
question=question,
|
||
fit_brief=normalized_fit_brief,
|
||
)
|
||
if recommendation_queries:
|
||
plan.search_queries = self._prepend_priority_queries(
|
||
recommendation_queries,
|
||
list(plan.search_queries or []),
|
||
)
|
||
elif not plan.search_queries:
|
||
plan.search_queries = list(fallback_plan.search_queries)
|
||
if not plan.preferred_domains:
|
||
plan.preferred_domains = list(fallback_plan.preferred_domains)
|
||
if not plan.topic:
|
||
plan.topic = fallback_plan.topic
|
||
plan = self._apply_freshness_search_contract(plan)
|
||
return plan, {
|
||
"planner": "llm_structured",
|
||
"fallback_taken": False,
|
||
"fit_brief_summary": plan.fit_brief_summary or fallback_plan.fit_brief_summary,
|
||
}
|
||
except Exception as exc:
|
||
return fallback_plan, {
|
||
"planner": "heuristic",
|
||
"fallback_taken": True,
|
||
"reason": str(exc)[:200],
|
||
"fit_brief_summary": fallback_plan.fit_brief_summary,
|
||
}
|
||
|
||
def _build_fallback_plan(
|
||
self,
|
||
*,
|
||
question: str,
|
||
query_semantics: dict[str, Any] | None = None,
|
||
fit_brief: dict[str, Any] | None = None,
|
||
) -> ExternalQueryPlan:
|
||
text = re.sub(r"\s+", " ", str(question or "")).strip()
|
||
semantics = query_semantics if isinstance(query_semantics, dict) else {}
|
||
query_profile = semantics.get("query_profile") if isinstance(semantics.get("query_profile"), dict) else {}
|
||
interaction_level = self._resolve_interaction_level(semantics)
|
||
normalized_fit_brief = self._normalize_fit_brief(fit_brief)
|
||
object_mentions = [
|
||
str(item or "").strip()
|
||
for item in list(semantics.get("object_mentions") or [])
|
||
if str(item or "").strip()
|
||
]
|
||
|
||
query_type = str(semantics.get("query_type") or "").strip().lower()
|
||
if query_type == "compare":
|
||
answer_contract = "compare_with_sources"
|
||
elif bool(query_profile.get("needs_freshness")):
|
||
answer_contract = "summary_with_sources"
|
||
else:
|
||
answer_contract = "direct_fact"
|
||
|
||
topic = text[:120] or "外部查询"
|
||
search_queries = self._build_search_queries(
|
||
text,
|
||
fit_brief=normalized_fit_brief,
|
||
)
|
||
preferred_domains = self._infer_preferred_domains(text)
|
||
subject_entities = object_mentions[:6]
|
||
location_hint = None
|
||
time_range_hint = self._infer_time_range_hint(text)
|
||
requires_multi_source = answer_contract == "compare_with_sources"
|
||
clarify_on_ambiguity = False
|
||
result_limit = 3 if requires_multi_source or bool(query_profile.get("needs_freshness")) else 2
|
||
fit_brief_summary = self._summarize_fit_brief(normalized_fit_brief)
|
||
project_binding, project_title = self._extract_project_binding(normalized_fit_brief)
|
||
project_object_anchors = self._collect_project_object_anchors(normalized_fit_brief)
|
||
|
||
plan = ExternalQueryPlan(
|
||
plan_version="v1",
|
||
query_kind="generic_web" if text else "unknown",
|
||
interaction_level=interaction_level,
|
||
answer_contract=answer_contract,
|
||
freshness_required=bool(query_profile.get("needs_freshness")),
|
||
topic=topic,
|
||
subject_entities=subject_entities,
|
||
location_hint=location_hint,
|
||
time_range_hint=time_range_hint,
|
||
preferred_domains=preferred_domains,
|
||
search_queries=search_queries,
|
||
result_limit=result_limit,
|
||
requires_multi_source=requires_multi_source,
|
||
clarify_on_ambiguity=clarify_on_ambiguity,
|
||
fit_brief_summary=fit_brief_summary,
|
||
project_binding=project_binding,
|
||
project_title=project_title,
|
||
project_scope_hint="current_project_and_repo" if fit_brief_summary else None,
|
||
recommendation_goal=self._build_recommendation_goal(
|
||
question=text,
|
||
fit_brief=normalized_fit_brief,
|
||
interaction_level=interaction_level,
|
||
),
|
||
project_object_anchors=project_object_anchors,
|
||
signal_source_objectives=self._collect_signal_source_objectives(normalized_fit_brief),
|
||
)
|
||
return self._apply_freshness_search_contract(plan)
|
||
|
||
def _build_search_queries(
|
||
self,
|
||
question: str,
|
||
*,
|
||
fit_brief: dict[str, Any] | None = None,
|
||
) -> list[str]:
|
||
recommendation_queries = self._build_signal_source_recommendation_queries(
|
||
question=question,
|
||
fit_brief=fit_brief,
|
||
)
|
||
if callable(self.query_builder):
|
||
try:
|
||
queries = self.query_builder(question)
|
||
normalized = [str(item or "").strip() for item in list(queries or []) if str(item or "").strip()]
|
||
if normalized:
|
||
return self._merge_fit_brief_queries(
|
||
self._prepend_priority_queries(recommendation_queries, normalized),
|
||
fit_brief=fit_brief,
|
||
)
|
||
except Exception:
|
||
pass
|
||
text = re.sub(r"\s+", " ", str(question or "")).strip()
|
||
queries = self._prepend_priority_queries(recommendation_queries, [text] if text else [])
|
||
return self._merge_fit_brief_queries(queries, fit_brief=fit_brief)
|
||
|
||
@staticmethod
|
||
def _infer_preferred_domains(question: str) -> list[str]:
|
||
_ = question
|
||
return []
|
||
|
||
@staticmethod
|
||
def _infer_time_range_hint(question: str) -> str | None:
|
||
text = str(question or "")
|
||
for pattern in (r"未来\d+天", r"未来一周", r"今天", r"截至今天", r"截至现在", r"目前", r"最新"):
|
||
match = re.search(pattern, text)
|
||
if match:
|
||
return str(match.group(0) or "").strip() or None
|
||
return None
|
||
|
||
@classmethod
|
||
def _apply_freshness_search_contract(cls, plan: ExternalQueryPlan) -> ExternalQueryPlan:
|
||
if not bool(plan.freshness_required):
|
||
return plan
|
||
current_year = datetime.now(timezone.utc).year
|
||
previous_year = current_year - 1
|
||
topic = str(plan.topic or "").strip()
|
||
subject = topic or "外部公开信息"
|
||
freshness_queries = [
|
||
f"{subject} {previous_year} {current_year}",
|
||
f"{subject} {current_year}",
|
||
]
|
||
existing_queries = list(plan.search_queries or [])
|
||
if cls._is_signal_source_recommendation_plan(plan):
|
||
plan.search_queries = cls._prepend_priority_queries(existing_queries, freshness_queries)
|
||
else:
|
||
plan.search_queries = cls._prepend_priority_queries(freshness_queries, existing_queries)
|
||
time_hint = str(plan.time_range_hint or "").strip()
|
||
if not time_hint or not re.search(r"20\d{2}", time_hint):
|
||
plan.time_range_hint = f"{previous_year}-{current_year}"
|
||
return plan
|
||
|
||
@staticmethod
|
||
def _is_signal_source_recommendation_plan(plan: ExternalQueryPlan) -> bool:
|
||
return bool(plan.signal_source_objectives or plan.project_object_anchors)
|
||
|
||
@staticmethod
|
||
def _normalize_fit_brief(payload: dict[str, Any] | None) -> dict[str, Any]:
|
||
if not isinstance(payload, dict):
|
||
return {}
|
||
try:
|
||
return ProjectFitBrief.model_validate(payload).model_dump()
|
||
except Exception:
|
||
return {}
|
||
|
||
@staticmethod
|
||
def _summarize_fit_brief(fit_brief: dict[str, Any] | None) -> str:
|
||
brief = fit_brief if isinstance(fit_brief, dict) else {}
|
||
parts: list[str] = []
|
||
project_summary = str(brief.get("project_summary") or "").strip()
|
||
repo_summary = str(brief.get("repo_summary") or "").strip()
|
||
capabilities = [
|
||
str(item or "").strip()
|
||
for item in list(brief.get("current_capabilities") or [])
|
||
if str(item or "").strip()
|
||
]
|
||
objectives = [
|
||
str(item or "").strip()
|
||
for item in list(brief.get("signal_source_objectives") or [])
|
||
if str(item or "").strip()
|
||
]
|
||
if project_summary:
|
||
parts.append(project_summary)
|
||
if repo_summary:
|
||
parts.append(repo_summary)
|
||
if capabilities:
|
||
parts.append(f"当前能力：{'、'.join(capabilities[:4])}")
|
||
if objectives:
|
||
parts.append(f"任务目标：{'；'.join(objectives[:2])}")
|
||
return " ".join(parts)[:600]
|
||
|
||
@staticmethod
|
||
def _extract_project_binding(fit_brief: dict[str, Any] | None) -> tuple[int | None, str | None]:
|
||
brief = fit_brief if isinstance(fit_brief, dict) else {}
|
||
project_summary = str(brief.get("project_summary") or "").strip()
|
||
match = re.search(r"当前项目：#(?P<id>\d+)\s+(?P<title>.+?)(?:\s*\(|$)", project_summary)
|
||
if not match:
|
||
return None, None
|
||
try:
|
||
project_id = int(match.group("id"))
|
||
except Exception:
|
||
return None, None
|
||
title = str(match.group("title") or "").strip() or None
|
||
return project_id, title
|
||
|
||
@staticmethod
|
||
def _collect_project_object_anchors(fit_brief: dict[str, Any] | None) -> list[str]:
|
||
brief = fit_brief if isinstance(fit_brief, dict) else {}
|
||
anchors: list[str] = []
|
||
for pool_name in ("signal_source_anchors", "signal_source_objectives", "current_capabilities", "recommendation_focus"):
|
||
for item in list(brief.get(pool_name) or []):
|
||
normalized = str(item or "").strip()
|
||
if normalized and normalized not in anchors:
|
||
anchors.append(normalized)
|
||
return anchors[:6]
|
||
|
||
@staticmethod
|
||
def _collect_signal_source_objectives(fit_brief: dict[str, Any] | None) -> list[str]:
|
||
brief = fit_brief if isinstance(fit_brief, dict) else {}
|
||
objectives: list[str] = []
|
||
for item in list(brief.get("signal_source_objectives") or []):
|
||
normalized = str(item or "").strip()
|
||
if normalized and normalized not in objectives:
|
||
objectives.append(normalized)
|
||
return objectives[:4]
|
||
|
||
@staticmethod
|
||
def _resolve_interaction_level(query_semantics: dict[str, Any] | None) -> str | None:
|
||
semantics = query_semantics if isinstance(query_semantics, dict) else {}
|
||
query_profile = semantics.get("query_profile") if isinstance(semantics.get("query_profile"), dict) else {}
|
||
for raw in (
|
||
semantics.get("interaction_level"),
|
||
query_profile.get("interaction_level"),
|
||
):
|
||
normalized = str(raw or "").strip().lower()
|
||
if normalized in {"status", "recommendation", "draft", "execute", "clarify"}:
|
||
return normalized
|
||
return None
|
||
|
||
def _build_recommendation_goal(
|
||
self,
|
||
*,
|
||
question: str,
|
||
fit_brief: dict[str, Any] | None,
|
||
interaction_level: str | None,
|
||
) -> str | None:
|
||
focus = [
|
||
str(item or "").strip()
|
||
for item in list((fit_brief or {}).get("recommendation_focus") or [])
|
||
if str(item or "").strip()
|
||
]
|
||
if focus:
|
||
return "；".join(focus[:3])[:240]
|
||
if str(interaction_level or "").strip().lower() != "recommendation":
|
||
return None
|
||
text = str(question or "").strip()
|
||
if not text:
|
||
return None
|
||
return f"判断“{text[:80]}”是否适合当前项目接入或跟进。"
|
||
|
||
@staticmethod
|
||
def _prepend_priority_queries(priority: list[str], existing: list[str]) -> list[str]:
|
||
merged: list[str] = []
|
||
for item in list(priority or []) + list(existing or []):
|
||
normalized = str(item or "").strip()
|
||
if normalized and normalized not in merged:
|
||
merged.append(normalized)
|
||
return merged[:5]
|
||
|
||
@staticmethod
|
||
def _extract_project_label(question: str, fit_brief: dict[str, Any] | None) -> str | None:
|
||
text = re.sub(r"\s+", " ", str(question or "")).strip()
|
||
match = re.search(r"([\u4e00-\u9fffA-Za-z0-9_-]{2,24})这个项目", text)
|
||
if match:
|
||
candidate = str(match.group(1) or "").strip("“”\"' ,，。")
|
||
if candidate:
|
||
return candidate[:24]
|
||
summary = str((fit_brief or {}).get("project_summary") or "").strip()
|
||
if not summary:
|
||
return None
|
||
summary = re.sub(r"^当前项目：", "", summary).strip()
|
||
summary = re.sub(r"^#\d+\s*", "", summary).strip()
|
||
summary = re.split(r"\s*\(|（|status=", summary)[0].strip()
|
||
for delimiter in ("，", ",", "。", "；", ";", "：", ":"):
|
||
if delimiter in summary:
|
||
summary = str(summary.split(delimiter, 1)[0] or "").strip()
|
||
break
|
||
return summary[:24] or None
|
||
|
||
@staticmethod
|
||
def _build_signal_source_recommendation_queries(
|
||
*,
|
||
question: str,
|
||
fit_brief: dict[str, Any] | None = None,
|
||
) -> list[str]:
|
||
project_label = ExternalQueryPlanner._extract_project_label(question, fit_brief)
|
||
objectives = ExternalQueryPlanner._collect_signal_source_objectives(fit_brief)
|
||
anchors = ExternalQueryPlanner._collect_project_object_anchors(fit_brief)
|
||
if not objectives and not anchors:
|
||
return []
|
||
queries: list[str] = []
|
||
if objectives:
|
||
queries.append(
|
||
" ".join(
|
||
item for item in [project_label or "", *objectives[:2]] if item
|
||
)
|
||
)
|
||
queries.append(
|
||
" ".join(
|
||
item
|
||
for item in [
|
||
project_label or "",
|
||
objectives[0],
|
||
]
|
||
if item
|
||
)
|
||
)
|
||
else:
|
||
queries.append(
|
||
" ".join(
|
||
item for item in [project_label or "", *anchors[:3]] if item
|
||
)
|
||
)
|
||
|
||
normalized: list[str] = []
|
||
for item in queries:
|
||
candidate = str(item or "").strip()
|
||
if candidate and candidate not in normalized:
|
||
normalized.append(candidate)
|
||
return normalized[:3]
|
||
|
||
def _merge_fit_brief_queries(
|
||
self,
|
||
queries: list[str],
|
||
*,
|
||
fit_brief: dict[str, Any] | None = None,
|
||
) -> list[str]:
|
||
normalized = [str(item or "").strip() for item in list(queries or []) if str(item or "").strip()]
|
||
if not normalized:
|
||
return []
|
||
anchors = self._collect_project_object_anchors(fit_brief)
|
||
if anchors:
|
||
anchor = anchors[0]
|
||
candidate = f"{normalized[0]} {anchor}".strip()
|
||
if candidate and candidate not in normalized:
|
||
normalized.append(candidate)
|
||
return normalized[:5]
|