"""Build bounded, traceable seat input from MAF roundtable context. Converts the shared roundtable state (task, previous turns, fact ledger, stage instructions) into a single human message for a seat's DeerFlow thread. Design constraints (MAF-02-06): - Bounded: total output STRICTLY never exceeds ``max_context_chars``. - Trimmable: each section is independently truncatable. - Traceable: includes turn references so the seat can cite prior work. - Never injects credentials, tokens, or raw binary content. - All content is sanitized before inclusion. """ from __future__ import annotations from typing import Any from app.maf_roundtable.sanitize import sanitize _DEFAULT_MAX_CONTEXT_CHARS = 12000 _DEFAULT_MAX_TURN_CHARS = 2000 _DEFAULT_MAX_FACT_CHARS = 3000 def build_seat_input( *, task_text: str, stage_instruction: str, role_description: str, prior_turns: list[dict[str, Any]], fact_ledger: dict[str, Any] | None = None, max_context_chars: int = _DEFAULT_MAX_CONTEXT_CHARS, max_turn_chars: int = _DEFAULT_MAX_TURN_CHARS, max_fact_chars: int = _DEFAULT_MAX_FACT_CHARS, ) -> str: """Assemble the human message for a seat's DeerFlow run. The total output length is STRICTLY <= max_context_chars. Truncation happens per-section first, then a final hard cap ensures no overflow. """ sections: list[str] = [] sections.append(f"【任务】\n{task_text}") if role_description: sections.append(f"【你的角色】\n{role_description}") if stage_instruction: sections.append(f"【本阶段要求】\n{stage_instruction}") if prior_turns: turns_text = _format_prior_turns(prior_turns, max_turn_chars) if turns_text: sections.append(f"【已有讨论】\n{turns_text}") if fact_ledger: sanitized_ledger = sanitize(fact_ledger) if isinstance(sanitized_ledger, dict): facts_text = _format_fact_ledger(sanitized_ledger, max_fact_chars) if facts_text: sections.append(f"【共享事实账本】\n{facts_text}") result = "\n\n".join(sections) if len(result) > max_context_chars: result = result[:max_context_chars] return result def build_manager_synthesis_input( *, task_text: str, seat_results: list[dict[str, Any]], fact_ledger: dict[str, Any] | None = None, max_context_chars: int = _DEFAULT_MAX_CONTEXT_CHARS, ) -> str: """Build the manager's final synthesis prompt.""" sections: list[str] = [] sections.append(f"【原始任务】\n{task_text}") sections.append("【各席位分析结果】") for r in seat_results: agent_id = r.get("agent_id", "unknown") content = str(r.get("content", "")) if len(content) > 3000: content = content[:3000] + "..." sections.append(f"--- {agent_id} ---\n{content}") if fact_ledger: sanitized_ledger = sanitize(fact_ledger) if isinstance(sanitized_ledger, dict): facts_text = _format_fact_ledger(sanitized_ledger, _DEFAULT_MAX_FACT_CHARS) if facts_text: sections.append(f"【共享事实账本】\n{facts_text}") sections.append( "【综合要求】\n" "请基于以上各席位分析和事实账本，生成最终结论。要求：\n" "1. 明确标注共识点和分歧点\n" "2. 区分「建议动作」和「已实际执行的动作」\n" "3. 引用具体证据来源\n" "4. 列出失败或状态不确定的动作" ) result = "\n\n".join(sections) if len(result) > max_context_chars: result = result[:max_context_chars] return result # ── internal helpers ────────────────────────────────────────────────────────── def _format_prior_turns(turns: list[dict[str, Any]], max_per_turn: int) -> str: lines: list[str] = [] for t in turns: agent_id = t.get("agent_id", "unknown") turn_idx = t.get("turn_index", "?") summary = str(t.get("summary", "")) if len(summary) > max_per_turn: summary = summary[:max_per_turn] + "..." lines.append(f"[轮次{turn_idx}] {agent_id}: {summary}") return "\n".join(lines) def _format_fact_ledger(ledger: dict[str, Any], max_chars: int) -> str: parts: list[str] = [] candidates = ledger.get("candidate_facts") if candidates and isinstance(candidates, list): # Phase-3: candidate facts are UNVERIFIED seat claims — label them # explicitly so downstream seats / the finalizer never treat them as # established conclusions. parts.append("未经核验的席位主张（不能作为结论）:") for c in candidates[:20]: if isinstance(c, dict): parts.append(f" - {c.get('text', c)}") else: parts.append(f" - {c}") facts = ledger.get("confirmed_facts") if facts and isinstance(facts, list): parts.append("已确认事实（经交叉审查核验）:") for f in facts[:20]: if isinstance(f, dict): parts.append(f" - {f.get('text', f)}") else: parts.append(f" - {f}") evidence = ledger.get("evidence") if evidence and isinstance(evidence, list): parts.append("证据引用:") for e in evidence[:10]: if isinstance(e, dict): parts.append(f" - {e.get('ref') or e.get('text') or e}") else: parts.append(f" - {e}") tool_results = ledger.get("tool_results") if tool_results and isinstance(tool_results, list): parts.append("工具执行结果:") for tr in tool_results[:10]: if isinstance(tr, dict): parts.append(f" - {tr.get('tool_name', '?')}: {tr.get('status', '?')}") else: parts.append(f" - {tr}") proposed = ledger.get("proposed_actions") if proposed and isinstance(proposed, list): parts.append("建议动作(未执行):") for p in proposed[:8]: parts.append(f" - {p}") executed = ledger.get("executed_actions") if executed and isinstance(executed, list): parts.append("已执行外部写入:") for e in executed[:8]: if isinstance(e, dict): parts.append(f" - {e.get('tool_name', '?')} ({e.get('action_id', '')})") else: parts.append(f" - {e}") conflicts = ledger.get("conflicts") if conflicts and isinstance(conflicts, list): parts.append("未解决冲突:") for c in conflicts[:5]: parts.append(f" - {c}") pending = ledger.get("pending_actions") if pending and isinstance(pending, list): parts.append("等待确认动作:") for p in pending[:5]: parts.append(f" - {p}") failed = ledger.get("failed_or_unknown_actions") if failed and isinstance(failed, list): parts.append("失败或状态不确定的动作:") for f in failed[:5]: parts.append(f" - {f}") result = "\n".join(parts) if len(result) > max_chars: result = result[:max_chars] return result