325 lines
14 KiB
Python
325 lines
14 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
from .shared import *
|
||
|
||
class LLMStructuredCandidatesMixin:
|
||
@staticmethod
|
||
def _normalize_string_list(value: Any) -> list[str]:
|
||
if value is None:
|
||
return []
|
||
if not isinstance(value, (list, tuple, set)):
|
||
value = [value]
|
||
normalized: list[str] = []
|
||
seen: set[str] = set()
|
||
for raw in value:
|
||
text = str(raw or "").strip()
|
||
if not text:
|
||
continue
|
||
key = text.lower()
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
normalized.append(text[:120])
|
||
return normalized
|
||
|
||
@staticmethod
|
||
def _normalize_int_list(value: Any) -> list[int]:
|
||
if value is None:
|
||
return []
|
||
if not isinstance(value, (list, tuple, set)):
|
||
value = [value]
|
||
normalized: list[int] = []
|
||
seen: set[int] = set()
|
||
for raw in value:
|
||
if isinstance(raw, bool):
|
||
continue
|
||
try:
|
||
number = int(str(raw).strip())
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if number <= 0 or number in seen:
|
||
continue
|
||
seen.add(number)
|
||
normalized.append(number)
|
||
return normalized
|
||
|
||
@classmethod
|
||
def _coerce_task_breakdown_item(cls, payload: dict[str, Any]) -> dict[str, Any]:
|
||
normalized = dict(payload)
|
||
title = str(normalized.get("title") or normalized.get("task_name") or "").strip()
|
||
description = normalized.get("description")
|
||
task_type = str(normalized.get("type") or normalized.get("task_type") or "backend").strip().lower() or "backend"
|
||
task_key = str(normalized.get("task_key") or normalized.get("key") or "").strip()
|
||
if not task_key and title:
|
||
task_key = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or title
|
||
normalized["task_key"] = task_key
|
||
normalized["title"] = title
|
||
if description is not None:
|
||
normalized["description"] = str(description or "").strip() or None
|
||
normalized["type"] = task_type
|
||
|
||
estimated_hours = cls._coerce_number(normalized.get("estimated_hours"))
|
||
normalized["estimated_hours"] = float(estimated_hours or 0.0)
|
||
|
||
required_skill_names = normalized.get("required_skill_names")
|
||
normalized["required_skill_names"] = cls._normalize_string_list(required_skill_names)
|
||
normalized["required_skill_tag_ids"] = cls._normalize_int_list(normalized.get("required_skill_tag_ids"))
|
||
normalized["depends_on_task_keys"] = cls._normalize_string_list(normalized.get("depends_on_task_keys"))
|
||
|
||
risk_level = str(normalized.get("risk_level") or "medium").strip().lower() or "medium"
|
||
normalized["risk_level"] = risk_level
|
||
risk_notes = normalized.get("risk_notes")
|
||
normalized["risk_notes"] = str(risk_notes or "").strip() or None
|
||
|
||
confidence = cls._coerce_number(normalized.get("confidence"))
|
||
if confidence is None:
|
||
confidence = 0.0
|
||
if confidence > 1 and confidence <= 100:
|
||
confidence = confidence / 100.0
|
||
normalized["confidence"] = max(0.0, min(float(confidence), 1.0))
|
||
|
||
normalized["evidence_refs"] = cls._normalize_string_list(normalized.get("evidence_refs"))
|
||
normalized["priority"] = str(normalized.get("priority") or "medium").strip().lower() or "medium"
|
||
return normalized
|
||
|
||
@classmethod
|
||
def _coerce_task_breakdown_payload(cls, payload: dict[str, Any]) -> dict[str, Any]:
|
||
normalized = dict(payload)
|
||
tasks = normalized.get("tasks")
|
||
if isinstance(tasks, list):
|
||
normalized["tasks"] = [
|
||
cls._coerce_task_breakdown_item(item)
|
||
for item in tasks
|
||
if isinstance(item, dict)
|
||
]
|
||
normalized["detected_skills"] = cls._normalize_string_list(normalized.get("detected_skills"))
|
||
source_context = normalized.get("source_context")
|
||
if source_context is not None and not isinstance(source_context, dict):
|
||
normalized["source_context"] = source_context if isinstance(source_context, dict) else None
|
||
return normalized
|
||
|
||
@staticmethod
|
||
def _validate_task_breakdown_item(payload: dict[str, Any]) -> bool:
|
||
required_text_fields = ("task_key", "title", "type", "risk_level")
|
||
for key in required_text_fields:
|
||
value = str(payload.get(key) or "").strip()
|
||
if not value:
|
||
return False
|
||
if not isinstance(payload.get("estimated_hours"), (int, float)) or isinstance(payload.get("estimated_hours"), bool):
|
||
return False
|
||
if not isinstance(payload.get("confidence"), (int, float)) or isinstance(payload.get("confidence"), bool):
|
||
return False
|
||
for key in ("required_skill_names", "required_skill_tag_ids", "depends_on_task_keys", "evidence_refs"):
|
||
if not isinstance(payload.get(key), list):
|
||
return False
|
||
priority = payload.get("priority")
|
||
if priority is not None and not isinstance(priority, str):
|
||
return False
|
||
risk_notes = payload.get("risk_notes")
|
||
if risk_notes is not None and not isinstance(risk_notes, str):
|
||
return False
|
||
description = payload.get("description")
|
||
if description is not None and not isinstance(description, str):
|
||
return False
|
||
return True
|
||
|
||
def _iter_structured_candidates(
|
||
self,
|
||
*,
|
||
message: Any,
|
||
choice: Any,
|
||
response: Any,
|
||
include_reasoning: bool = True,
|
||
) -> list[tuple[str, Any]]:
|
||
candidates: list[tuple[str, Any]] = []
|
||
if isinstance(message, dict):
|
||
tool_calls = message.get("tool_calls")
|
||
if isinstance(tool_calls, list) and tool_calls:
|
||
first = tool_calls[0] if isinstance(tool_calls[0], dict) else {}
|
||
fn = first.get("function") if isinstance(first, dict) else {}
|
||
candidates.append(("tool_calls.function.arguments", fn.get("arguments") if isinstance(fn, dict) else None))
|
||
candidates.append(("message.content", message.get("content")))
|
||
if isinstance(choice, dict):
|
||
candidates.append(("choice.text", choice.get("text")))
|
||
if isinstance(response, dict):
|
||
candidates.append(("response.output_text", response.get("output_text")))
|
||
if include_reasoning:
|
||
if isinstance(message, dict):
|
||
candidates.append(("message.reasoning_content", message.get("reasoning_content")))
|
||
candidates.append(("message.reasoning_content", message.get("reasoning")))
|
||
if isinstance(choice, dict):
|
||
candidates.append(("choice.reasoning_content", choice.get("reasoning_content")))
|
||
candidates.append(("choice.reasoning_content", choice.get("reasoning")))
|
||
if isinstance(response, dict):
|
||
candidates.append(("response.reasoning_content", response.get("reasoning_content")))
|
||
candidates.append(("response.reasoning_content", response.get("reasoning")))
|
||
return candidates
|
||
|
||
@classmethod
|
||
def _unwrap_structured_payload_once(cls, payload: Any) -> Any:
|
||
if not isinstance(payload, dict):
|
||
return payload
|
||
for key in ("json", "data", "result", "arguments"):
|
||
if key not in payload:
|
||
continue
|
||
inner = payload.get(key)
|
||
if isinstance(inner, str):
|
||
candidate = cls._extract_json_candidate(cls._strip_code_fence(inner))
|
||
if not candidate:
|
||
continue
|
||
try:
|
||
return json.loads(candidate)
|
||
except Exception:
|
||
return inner
|
||
if isinstance(inner, (dict, list)):
|
||
return inner
|
||
return payload
|
||
|
||
@staticmethod
|
||
def _matches_expected_type(value: Any, expected: str) -> bool:
|
||
if expected == "bool":
|
||
return isinstance(value, bool)
|
||
if expected == "str":
|
||
return isinstance(value, str)
|
||
if expected == "list":
|
||
return isinstance(value, list)
|
||
if expected == "number":
|
||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||
if expected == "number_or_none":
|
||
return value is None or (isinstance(value, (int, float)) and not isinstance(value, bool))
|
||
if expected == "str_or_none":
|
||
return value is None or isinstance(value, str)
|
||
if expected == "dict_or_none":
|
||
return value is None or isinstance(value, dict)
|
||
return True
|
||
|
||
@staticmethod
|
||
def _coerce_number(value: Any) -> Optional[float]:
|
||
if isinstance(value, bool):
|
||
return None
|
||
if isinstance(value, (int, float)):
|
||
return float(value)
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return None
|
||
try:
|
||
return float(text)
|
||
except Exception:
|
||
return None
|
||
|
||
@staticmethod
|
||
def _coerce_int(value: Any) -> Optional[int]:
|
||
number = LLMStructuredCandidatesMixin._coerce_number(value)
|
||
if number is None:
|
||
return None
|
||
try:
|
||
return int(number)
|
||
except Exception:
|
||
return None
|
||
|
||
@staticmethod
|
||
def _coerce_bool(value: Any) -> Optional[bool]:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||
return bool(int(value))
|
||
text = str(value or "").strip().lower()
|
||
if not text:
|
||
return None
|
||
if text in BOOL_TRUE_VALUES:
|
||
return True
|
||
if text in {"0", "false", "no", "off", "n"}:
|
||
return False
|
||
return None
|
||
|
||
@classmethod
|
||
def _coerce_intent_structured_payload(cls, payload: dict[str, Any]) -> dict[str, Any]:
|
||
normalized = dict(payload)
|
||
confidence = cls._coerce_number(normalized.get("confidence"))
|
||
if confidence is None:
|
||
for alias in ("score", "probability", "confidence_score"):
|
||
confidence = cls._coerce_number(normalized.get(alias))
|
||
if confidence is not None:
|
||
break
|
||
if confidence is not None:
|
||
normalized["confidence"] = confidence
|
||
|
||
reasoning_value = normalized.get("reasoning")
|
||
if isinstance(reasoning_value, list):
|
||
text_items = [str(item).strip() for item in reasoning_value if str(item).strip()]
|
||
normalized["reasoning"] = "；".join(text_items)
|
||
elif isinstance(reasoning_value, dict):
|
||
normalized["reasoning"] = json.dumps(reasoning_value, ensure_ascii=False)
|
||
else:
|
||
reasoning_text = str(reasoning_value or "").strip()
|
||
if reasoning_text:
|
||
normalized["reasoning"] = reasoning_text
|
||
else:
|
||
for alias in ("reason", "rationale", "analysis", "explanation"):
|
||
alias_text = str(normalized.get(alias) or "").strip()
|
||
if alias_text:
|
||
normalized["reasoning"] = alias_text
|
||
break
|
||
|
||
if "requires_confirmation" in normalized:
|
||
bool_value = cls._coerce_bool(normalized.get("requires_confirmation"))
|
||
if bool_value is not None:
|
||
normalized["requires_confirmation"] = bool_value
|
||
|
||
for field in ("project_id", "requested_count"):
|
||
if field in normalized:
|
||
number = cls._coerce_int(normalized.get(field))
|
||
normalized[field] = number if number is not None else None
|
||
|
||
route_hint = str(
|
||
normalized.get("workflow_route") or normalized.get("route") or ""
|
||
).strip().lower()
|
||
if route_hint and "." in route_hint:
|
||
family, action = route_hint.split(".", 1)
|
||
if not str(normalized.get("family") or "").strip():
|
||
normalized["family"] = family
|
||
if not str(normalized.get("action") or "").strip():
|
||
normalized["action"] = action
|
||
return normalized
|
||
|
||
@classmethod
|
||
def _coerce_structured_payload(cls, *, function_name: str, payload: Any) -> Any:
|
||
fn = str(function_name or "").strip().lower()
|
||
if not isinstance(payload, dict):
|
||
return payload
|
||
if fn in {"generate_task_breakdown", "sstc_ai_breakdown"}:
|
||
return cls._coerce_task_breakdown_payload(payload)
|
||
return payload
|
||
|
||
def _validate_structured_payload(self, *, function_name: str, payload: Any) -> Optional[str]:
|
||
fn = str(function_name or "").strip().lower()
|
||
schema = STRUCTURED_OBJECT_SCHEMAS.get(fn)
|
||
payload = self._coerce_structured_payload(function_name=fn, payload=payload)
|
||
if schema is None:
|
||
return None if isinstance(payload, (dict, list)) else "structured_json_invalid"
|
||
if not isinstance(payload, dict):
|
||
return "structured_schema_invalid"
|
||
required_keys = schema.get("required_keys") or ()
|
||
for key in required_keys:
|
||
if key not in payload:
|
||
return "structured_schema_invalid"
|
||
type_checks = schema.get("type_checks") or {}
|
||
for key, expected in type_checks.items():
|
||
if not self._matches_expected_type(payload.get(key), str(expected)):
|
||
return "structured_schema_invalid"
|
||
optional_type_checks = schema.get("optional_type_checks") or {}
|
||
for key, expected_values in optional_type_checks.items():
|
||
if key not in payload:
|
||
continue
|
||
if not any(self._matches_expected_type(payload.get(key), str(expected)) for expected in expected_values):
|
||
return "structured_schema_invalid"
|
||
if fn in {"generate_task_breakdown", "sstc_ai_breakdown"}:
|
||
tasks = payload.get("tasks")
|
||
if not isinstance(tasks, list) or not tasks:
|
||
return "structured_schema_invalid"
|
||
if not all(isinstance(item, dict) and self._validate_task_breakdown_item(item) for item in tasks):
|
||
return "structured_schema_invalid"
|
||
return None
|