| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162 |
- from __future__ import annotations
- from typing import Any
- from .models import ContractNarration
- TASK_LABELS = {
- "direction": "创作方向",
- "pattern-retrieval": "模式检索",
- "decode-retrieval": "案例检索",
- "external-retrieval": "外部资料检索",
- "knowledge-retrieval": "知识检索",
- "structure": "整体结构",
- "paragraph": "内容段落",
- "element-set": "内容元素",
- "compare": "候选比较",
- "compose": "完整脚本组合",
- "candidate-portfolio": "候选方案集",
- "root": "最终交付",
- }
- def task_label(task_kind: str) -> str:
- return TASK_LABELS.get(task_kind, task_kind.replace("-", " "))
- def narrate_contract(contract: dict[str, Any]) -> ContractNarration:
- task_kind = str(contract.get("task_kind") or "task")
- label = task_label(task_kind)
- objective = _plain(contract.get("objective")) or f"完成{label}"
- goals = [str(value) for value in contract.get("goal_ids") or [] if str(value).strip()]
- inputs = []
- for item in contract.get("input_decision_refs") or []:
- if not isinstance(item, dict):
- continue
- source_kind = task_label(str(item.get("expected_task_kind") or "上游成果"))
- inputs.append(f"已接受的{source_kind}")
- if contract.get("base_artifact_ref"):
- inputs.append("已冻结的基础成果")
- if not inputs:
- inputs.append("当前已冻结的创作输入")
- criteria = [
- _plain(item.get("description"))
- for item in contract.get("criteria") or []
- if isinstance(item, dict) and _plain(item.get("description"))
- ]
- return ContractNarration(
- action=f"创建{label} Task",
- reasoning=f"根据任务合同：{objective}",
- goals=goals,
- inputs=_unique(inputs),
- output=label,
- criteria=_unique(criteria),
- )
- def _plain(value: Any) -> str:
- return " ".join(str(value or "").split())
- def _unique(values: list[str]) -> list[str]:
- return list(dict.fromkeys(value for value in values if value))
|
