wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 圆桌第3步大屏多页改用专职智能体 + 健壮 report-json 抽取

Browse Source 
修复「大屏多页抽取 json 又失败」：旧实现复用画 HTML 的 roundtable-report
智能体 + rfind 围栏 + 单次 json.loads 的简单粗暴抽取，弱模型跑偏/格式偏差即失败。

- 新内置专职单例 roundtable-dashboard「大屏多页数据智能体」：SOUL 专讲逐席位
  分析 → report-json 契约，禁工具/禁文件/禁 HTML；新 run policy dashboard 禁用一切工具
- harness 新增 report_json.py：多策略抽取（report-json/json/裸 JSON/截断）+ schema
  归一化（移植前端 extractReportData/coerceReportData），返回可直接渲染的归一化 JSON
- 后台 run_dashboard_data 改用专职 agent + dashboard policy + 健壮抽取
- 新增 /api/multi-agent/dashboard/init；前端 initDashboardThread + useStep3Dashboard 改用之

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
董宏凯andClaude Opus 4.8 committed 2026-06-17 21:47:21 +08:00
1 parent bb05f7b1de
commit ed887f762e
9 files changed
+452 -29
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontend-web/src/roundtable-planning/api/multi-agent.ts
+26
View File
@@ -835,6 +835,32 @@ export async function initReportThread(
return { agentName: json.agent_name, threadId: json.thread_id };
}
/**
* Create a thread for the Step 3「大屏多页」dashboard-data agent
* (`roundtable-dashboard`, a built-in singleton — distinct from `roundtable-report`;
* its run policy disables ALL tools so it only emits a ```report-json block). Run it
* with {@link streamMultiAgent} (`agentType: "special"`, `agentName: "roundtable-dashboard"`,
* `dAgentThreadId: { [agentName]: threadId }`); the frontend then `extractReportData`s
* the streamed text. Follow-up Q&A / refinements reuse the SAME thread (do not re-init)
* so the agent keeps the prior dashboard context.
*/
export async function initDashboardThread(
options: { signal?: AbortSignal } = {},
): Promise<ReportThreadResponse> {
const res = await apiFetch(`${getBackendBaseURL()}/api/multi-agent/dashboard/init`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: "{}",
signal: options.signal,
});
if (!res.ok) {
const text = await res.text().catch(() => "");
throw new Error(`init dashboard thread failed (${res.status}): ${text}`);
}
const json = (await res.json()) as { agent_name: string; thread_id: string };
return { agentName: json.agent_name, threadId: json.thread_id };
}
/**
* Create a thread for the Step 3「总结报告」summary agent (`roundtable-summary`,
* a built-in singleton — distinct from `roundtable-report`; this only spins up a
frontend-web/src/roundtable-planning/hooks/useStep3Dashboard.ts
+3 -3
View File
@@ -15,7 +15,7 @@ import { useCallback, useEffect, useRef, useState } from "react";
import { useArtifacts } from "@/components/workspace/artifacts";
import type { DraftStep3Snapshot } from "@/roundtable-planning/api/drafts";
import {
initReportThread,
initDashboardThread,
streamMultiAgent,
type StreamPhase,
} from "@/roundtable-planning/api/multi-agent";
@@ -35,7 +35,7 @@ import {
} from "@/roundtable-planning/lib/report-template";
import { nowTime } from "@/roundtable-planning/lib/roundtable-constants";
const DASH_AGENT_FALLBACK = "roundtable-report";
const DASH_AGENT_FALLBACK = "roundtable-dashboard";
export type DashboardPhase = "idle" | "creating" | "streaming" | "done" | "error";
@@ -269,7 +269,7 @@ export function useStep3Dashboard(options: UseStep3DashboardOptions): UseStep3Da
const { bubbleId, message, aborter } = args;
let threadId = liveThreadRef.current;
if (!threadId) {
const init = await initReportThread({ signal: aborter.signal });
const init = await initDashboardThread({ signal: aborter.signal });
if (aborter.signal.aborted) return;
threadId = init.threadId;
agentNameRef.current = init.agentName || DASH_AGENT_FALLBACK;
offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py
+25 -24
View File
@@ -15,7 +15,6 @@
from __future__ import annotations
import asyncio
import json
import logging
import os
import uuid
@@ -28,10 +27,12 @@ from langgraph.checkpoint.base import empty_checkpoint
from app.gateway.roundtable_run_policy import roundtable_run_policy
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
DASHBOARD_AGENT_ID,
REPORT_AGENT_ID,
SUMMARY_AGENT_ID,
ensure_roundtable_functional_agents,
)
from deerflow.agents.roundtable_orchestrator.report_json import extract_report_json
from deerflow.agents.roundtable_orchestrator.types import (
LeaderTurn,
ReportResult,
@@ -331,30 +332,24 @@ _SUMMARY_RETRY_MISSING = (
"先写前半部分、再用 write_file 的 append=true 参数分段续写。写完后把报告正文也输出到对话回复里。"
)
_DASHBOARD_FENCE = "```report-json"
_DASHBOARD_RETRY = (
"你刚才没有输出有效的 ```report-json 代码块（或其中的 JSON 无法解析）。"
"请重新在回复**最末尾**输出一个完整且**严格合法**的 ```report-json 代码块"
"（标准 JSON：无注释、无尾逗号、字符串用双引号），代码块之外只允许一两句说明；"
"你刚才没有输出可用的大屏数据（缺少 ```report-json 代码块，或其中的 JSON 无法解析、"
"字段不符合契约）。请重新**逐席位分析**研讨成果，在回复**最末尾**输出一个完整且**严格合法**"
"的 ```report-json 代码块：标准 JSON（双引号、无注释、无尾逗号、数字不加引号），seats 覆盖"
"每一个有交付的席位，dependsOn 只引用本清单内的席位 id；代码块之外只允许一两句说明；"
"不要调用任何工具、不要写文件、不要输出 HTML。"
)
def _extract_report_json(text: str) -> str | None:
"""从模型回复里提取最后一个 ```report-json 代码块的 JSON 字符串（须可解析）。"""
start = text.rfind(_DASHBOARD_FENCE)
if start < 0:
return None
after = text[start + len(_DASHBOARD_FENCE) :]
close = after.find("```")
body = (after[:close] if close >= 0 else after).strip()
if not body:
return None
try:
json.loads(body)
except Exception: # noqa: BLE001
return None
return body
"""从模型回复里健壮抽取大屏 report-json，返回**归一化后的** JSON 字符串（失败 → None）。
旧实现是「找最后一个 ```report-json 围栏 + json.loads」的简单粗暴抽取，模型一旦用
```json / 裸 JSON / 流式截断 / 尾逗号 就直接失败（即用户反馈的「大屏多页抽取 json 又失败」）。
现委托给 harness 的 :func:`extract_report_json`（多策略抽取 + schema 归一化，与前端
``report-template.ts`` 同源），返回的 JSON 字段齐全、可被前端直接 JSON.parse。
"""
return extract_report_json(text)
def _html_complete(html: str | None) -> bool:
@@ -569,19 +564,25 @@ class InProcessRoundtableGateway:
return SummaryResult(md=final_md, closing=closing, model=model or self._model, thread_id=summary_thread_id)
async def run_dashboard_data(self, *, prompt, model) -> str | None:
"""大屏多页（模式二）的结构化数据：进程内驱动 roundtable-report 输出
```report-json（prompt 已声明禁工具/禁文件），返回可解析的 JSON 字符串。
缺失 / 坏 JSON → 同线程纠偏重试（同 run_report/run_summary 的自愈语义）。"""
"""大屏多页的结构化数据：进程内驱动**专职**「大屏多页数据智能体」
(``roundtable-dashboard``) 逐席位分析研讨成果，在对话里输出 ```report-json。
与旧版的差异（修复「大屏多页抽取 json 又失败」）：
- 用**专职** dashboard 智能体（SOUL 专讲「逐席位分析 → 结构化数据」）而非复用
画 HTML 的 report 智能体；``dashboard`` policy **禁用一切工具**，杜绝弱模型跑去写文件。
- 抽取走 :func:`_extract_report_json` → harness 健壮多策略抽取 + schema 归一化，
容忍 ```json / 裸 JSON / 流式截断，并保证返回的 JSON 字段齐全可直接渲染。
- 仍保留同线程纠偏重试（缺失/坏 JSON 时追加 ``_DASHBOARD_RETRY``）。"""
ensure_roundtable_functional_agents()
thread_id = await _create_thread(self._app)
policy = roundtable_run_policy("report")
policy = roundtable_run_policy("dashboard")
message = prompt
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
messages = await _run_agent_turn(
self._app,
user_id=self._user_id,
thread_id=thread_id,
agent_name=REPORT_AGENT_ID,
agent_name=DASHBOARD_AGENT_ID,
message=message,
model=model or self._model,
excluded_tools=policy.excluded_tools,
offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py
+31 -1
View File
@@ -119,6 +119,29 @@ _REPORT_EXCLUDED: list[str] = [*_SEAT_EXCLUDED, "str_replace", "read_peer_delive
# 完全一致——可写文件(write_file 整文件重写),禁 str_replace(整文件重写才能让右侧 md 预览
# 实时刷新)/bash/联网/派活/呈现文件/看图/读前序席位。单列出来便于以后独立调优。
_SUMMARY_EXCLUDED: list[str] = [*_SEAT_EXCLUDED, "str_replace", "read_peer_delivery"]
# dashboard：Step3「大屏多页」数据智能体(roundtable-dashboard)——纯「研讨成果 → report-json」
# 提炼器,**禁用一切工具**(同 leader 的黑名单全集)。它只需读 prompt、分析各席位、在对话里
# 输出一个 report-json 代码块;留任何干活类工具(尤其 write_file / bash)只会让弱模型跑偏去
# 写文件而不输出 JSON(report agent 上已实锤这类行为)。thinking + medium:需要逐席位分析归纳。
_DASHBOARD_EXCLUDED: list[str] = [
"web_search",
"view_image",
"bash",
"ls",
"read_file",
"write_file",
"str_replace",
"task",
"write_todos",
"memory",
"hindsight_recall",
"hindsight_reflect",
"hindsight_retain",
"read_peer_delivery",
"agent_orchestration",
"ask_clarification",
"present_files",
]
# leader 内禀的 skill_stop：``agent_orchestration`` 调用后停图。调用方可在此基础上
# 再叠加自己的 skill_stop（见 ``merge_skill_stop_names``）。
@@ -181,8 +204,15 @@ def roundtable_run_policy(role: str) -> RoundtableRunPolicy:
thinking_enabled=True,
reasoning_effort="medium",
)
if role == "dashboard":
return RoundtableRunPolicy(
excluded_tools=list(_DASHBOARD_EXCLUDED),
skill_stop_names=[],
thinking_enabled=True,
reasoning_effort="medium",
)
raise ValueError(
f"unknown roundtable role: {role!r} (expected leader/seat/seat_parallel/report/summary)"
f"unknown roundtable role: {role!r} (expected leader/seat/seat_parallel/report/summary/dashboard)"
)
offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py
+10 -1
View File
@@ -67,6 +67,13 @@ REPORT_AGENT_ID = "roundtable-report"
# 总结报告(写入 outputs/方案总结报告.md),并按业务链总控编排提示组织。与 report 并存
# 的另一内置单例(report 画 HTML 看板,summary 写 Markdown 报告 + 支持问答/增量改)。
SUMMARY_AGENT_ID = "roundtable-summary"
# Step 3「大屏多页」用的结构化数据智能体:读各席位交付 + 共识,逐席位分析并提炼成
# 一份大屏展示用的结构化数据(report-json),供前端按模板渲染成多页大屏。**不调任何工具、
# 不写文件、不输出 HTML**——只在对话里输出一个严格合法的 ```report-json 代码块。与 report
# (画 HTML 看板)/summary(写 Markdown)并存的第三个 Step3 内置单例,把「大屏抽取」从
# 旧版「复用 report agent + 简单粗暴字符串抽取」改为专职智能体 + 健壮抽取(见
# roundtable_orchestrator.report_json)。
DASHBOARD_AGENT_ID = "roundtable-dashboard"
# 验证版 SOUL.md + config.yaml 的源码资源目录(随 git 发布)。每个子目录名即 agent id,
# 内含该 agent 的 config.yaml 与 SOUL.md,是 seeder 落盘时字节级复制的来源。
@@ -74,13 +81,15 @@ _ASSETS_DIR = Path(__file__).parent / "_roundtable_seed_assets"
# 这几个功能型/内置 agent 任何部署都必须存在;顺序无关紧要(逐个独立补建)。
# 前三个支撑 Step 1 澄清 / 推荐 / Step 2 总控;roundtable-report 支撑 Step 3 结果绘制
# (HTML 看板);roundtable-summary 支撑 Step 3 总结报告(Markdown + 问答/增量改)。
# (HTML 看板);roundtable-summary 支撑 Step 3 总结报告(Markdown + 问答/增量改);
# roundtable-dashboard 支撑 Step 3 大屏多页结构化数据(report-json + 问答/增量改)。
_FUNCTIONAL_AGENT_IDS: tuple[str, ...] = (
INTENT_AGENT_ID,
RECOMMENDER_AGENT_ID,
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
SUMMARY_AGENT_ID,
DASHBOARD_AGENT_ID,
)
# seeder 会复制资源目录里的这些文件(存在即复制,缺失则跳过)。
offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-dashboard/SOUL.md
+57
View File
@@ -0,0 +1,57 @@
# 圆桌大屏多页数据智能体
## 核心定位
你是多智能体圆桌研讨 **Step 3** 的「大屏多页数据智能体」。各席位（情报收集 / 环境评估 / 方案设计 / 风险审查 / 执行规划 / 总结陈述等）已经完成研讨并产出交付，总控也给出了最终共识。
你的**唯一职责**：通读这些文字成果，**逐个席位地分析、提炼**，把整场研讨归纳成一份**大屏展示用的结构化数据**，在回复**最末尾**输出一个**严格合法**的 ```report-json 代码块。
这份数据会被前端用既定模板渲染成「多页大屏」（总览页 + 每个席位一页）。所以数据的**结构正确性与可解析性是第一位的**——宁可内容朴素，也必须是一份能被 `JSON.parse` 解析、字段齐全的 JSON。
你**不**调研、**不**联网搜索、**不**派活、**不**追问用户、**不**写文件、**不**输出 HTML——你是一个「研讨成果 → 结构化数据」的提炼器，只在对话里输出一个 report-json 代码块。
## 输入说明
用户消息里会给你三部分原料（可能有缺省）：
1. **任务意图摘要**：目标、约束、关键假设（来自 Step 1）；
2. **各席位最终交付**：每个席位的名称 + 它产出的完整正文；
3. **总控最终共识结论**。
## 工作流程（先分析，再产出）
1. **逐席位分析**：依次读每个有交付的席位，在思考里为它提炼出——该席位的一句话结论（summary）、2–4 个小节要点（sections）、0–3 个可量化指标（metrics，0–100 评分）、0–4 个风险点（risks，high/mid/low）、它承接了哪些其它席位的产物（dependsOn，只能引用本次清单里的席位 id）。
2. **抽取全局总览**：从跨席位信息里提炼总标题、一句话定位、3–6 个核心 KPI、1–3 个总览图表（风险分布 / 资源占比 / 方案评分 / 竞争力雷达等）、3–6 段执行路线（timeline），以及总控的最终结论 headline + 要点 points。
3. **缺数字时合理估算**：没有明确数字时，按研讨内容把定性的「高/中/低」映射成评分、把阶段描述排成时间线等，**合理估算而非凭空编造与原料矛盾的数据**；估算的指标在 label 里可带「(估)」字样。
4. **输出**：在回复最末尾输出一个 ```report-json 代码块。代码块之外只允许一两句简短说明，**不要**把数据重复贴进正文。
## report-json 契约（严格 JSON，全程简体中文）
```report-json
{"title":"大屏主标题（≤20字）","subtitle":"一句话定位",
"kpis":[{"label":"指标名","value":"数值","unit":"单位","trend":"up|down|flat"}],
"consensus":{"headline":"一句话最终结论","points":["结论要点，3-6 条"]},
"charts":[{"type":"line|bar|donut|radar","title":"图表标题","items":[{"label":"维度","value":数字}]}],
"timeline":[{"label":"阶段名","desc":"一句话说明"}],
"seats":[{"id":"seat-1","name":"席位名（与交付席位一致）","summary":"该席位结论一两句",
"sections":[{"heading":"小节标题","bullets":["要点，每条≤40字"]}],
"metrics":[{"label":"量化指标名","value":0到100}],
"risks":[{"label":"风险点","level":"high|mid|low"}],
"dependsOn":["它承接哪些席位的产物（席位id）"]}]}
```
### 字段硬性要求
- `seats` 覆盖**每一个有交付的席位**，一个席位一项；`id` 用 `seat-1`..`seat-N`，`name` 与交付席位名一致；
- `dependsOn` 只能引用本清单内出现过的席位 `id`，不能自指，不能编造不存在的 id；
- `charts` 的 `type` 只能是 `line` / `bar` / `donut` / `radar`；`kpis` 的 `trend` 只能是 `up` / `down` / `flat`；`risks` 的 `level` 只能是 `high` / `mid` / `low`；
- 数值字段写数字（不带引号、不带单位）；`kpis.value` 可带文字（如「约 30」），单位单独放 `unit`；
- **JSON 必须严格合法**：双引号、无注释、无尾逗号、无单引号、布尔/数字不加引号。
## 多轮修改（用户会在输入框里继续提要求）
数据生成后，用户可能继续发来修改要求（如「把风险分布换成柱状图」「新增一个执行路线席位」「第 2 个席位的结论改细些」）。处理规则：
1. 把新消息当作对**上一份 report-json 的增量修改**：只改用户提到的部分，其余字段尽量保持不变；
2. 仍在回复最末尾输出**完整的、严格合法的** ```report-json 代码块（不是只给 diff）；
3. 若用户只是提问而非要求改数据，就只用文字回答、不重出代码块。
## 禁止事项
- ❌ 调用任何工具（write_file / bash / web_search / 派活 / 看图 / 澄清 等一律不用）
- ❌ 写任何文件、输出任何 HTML / Markdown 表格作为「产物」（产物只有那个 report-json 代码块）
- ❌ 输出非法 JSON（注释、尾逗号、单引号、数字带引号、缺闭合括号）
- ❌ 代码块之外大段复述数据，或把 JSON 也贴一份到正文
- ❌ 编造与原料明显矛盾的数据，或编造不存在的席位 id
- ❌ 在数据里添加「AI 自动生成」之类的水印字样
offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-dashboard/config.yaml
+3
View File
@@ -0,0 +1,3 @@
description: 圆桌大屏多页数据智能体，读取各席位的最终交付与总控共识，逐席位分析并提炼成一份大屏展示用的结构化数据（report-json）。本身不调研、不联网、不派活、不写文件、不输出 HTML，只做分析归纳与结构化抽取。
id: roundtable-dashboard
name: 大屏多页数据
offline-backend-20260512/backend/app/gateway/routers/multi_agent.py
+26
View File
@@ -65,6 +65,7 @@ from langgraph.checkpoint.base import empty_checkpoint
from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
DASHBOARD_AGENT_ID,
REPORT_AGENT_ID,
SUMMARY_AGENT_ID,
ensure_roundtable_functional_agents,
@@ -945,6 +946,28 @@ async def init_summary_thread(request: Request) -> dict[str, str]:
return {"agent_name": SUMMARY_AGENT_ID, "thread_id": thread_id}
@router.post("/dashboard/init")
async def init_dashboard_thread(request: Request) -> dict[str, str]:
"""为 Step 3「大屏多页」创建一个供大屏多页数据智能体(``roundtable-dashboard``)
运行的线程,返回 ``{"agent_name", "thread_id"}``。
与 ``/report/init`` / ``/summary/init`` 同构,换成第三个内置单例
``roundtable-dashboard``(专职「逐席位分析 → report-json」,run policy 为 ``dashboard``,
**禁用一切工具**):
- ``roundtable-dashboard`` 是**内置单例**(``_roundtable_seed`` 种子化,启动即同步进 DB),
这里**只建一个轻量 thread**,绝不重建 agent;
- 线程打 ``_ROUNDTABLE_THREAD_METADATA`` 标记成系统会话,从「最近对话」里隐藏;
- 入口先 ``ensure_roundtable_functional_agents()`` 自愈(磁盘目录被手删也能补建);
- 建好后前端用现有 ``POST /run/stream``(``agent_type="special"``、
``agent_name="roundtable-dashboard"``、``d_agent_thread_id={DASHBOARD_AGENT_ID: thread_id}``)
跑这一轮(后续问答/增量改**复用同一线程**,不重新 init),智能体在对话里输出
```report-json,前端用 ``extractReportData`` 抽取后按当前主题渲染大屏。
"""
ensure_roundtable_functional_agents()
thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
return {"agent_name": DASHBOARD_AGENT_ID, "thread_id": thread_id}
# ---------------------------------------------------------------------------
# 上游 LangGraph SSE 透传（intent / recommend / leader / special 共用）
# ---------------------------------------------------------------------------
@@ -1613,6 +1636,9 @@ async def _special_run(
role = "report"
elif agent_name == SUMMARY_AGENT_ID:
role = "summary"
elif agent_name == DASHBOARD_AGENT_ID:
# 大屏多页数据智能体：纯「研讨成果 → report-json」提炼，禁用一切工具。
role = "dashboard"
elif no_file:
role = "seat_parallel"
else:
offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/report_json.py
+271
View File
@@ -0,0 +1,271 @@
"""Step3「大屏多页」report-json 的健壮抽取 + 归一化（schema coercion）。
后台与前端共用同一份契约：内置「大屏多页智能体」(``roundtable-dashboard``) 通读各席位
交付 + 总控共识，在回复末尾输出一个 ```report-json 代码块描述大屏数据。
模型实际输出常有偏差——用 ```json 而非 ```report-json、裸 JSON 无围栏、流式截断没闭合、
尾逗号 / 单引号等。旧版后台用单一 ``rfind("```report-json")`` + ``json.loads`` 抽取，
任何偏差都直接失败（即用户反馈的「大屏多页抽取 json 又失败」）。
本模块把前端 ``report-template.ts`` 里已验证的多策略抽取 + 字段归一化移植到后端：
依次尝试 report-json / json / 无标注 代码块、未闭合围栏尾串、夹在文字里的平衡 ``{…}``
对象；每个候选 ``json.loads`` 后过 :func:`coerce_report_data` 校验「像不像本契约」，
命中即返回**归一化后的 JSON 字符串**——保证可被前端 ``JSON.parse`` 且字段齐全、
``seats`` 的 ``dependsOn`` 只引用真实存在的席位 id。
纯逻辑、无 app 依赖，放在 harness 供进程内网关与测试复用。
"""
from __future__ import annotations
import json
import re
from typing import Any
_FENCE = "```report-json"
# 契约关键键：用来判断一段 JSON「像不像」本大屏契约，避免把无关 {} 误当数据。
_LOOKS_LIKE_RE = re.compile(r'"(seats|consensus|kpis|charts|timeline)"\s*:')
# 收集所有 ```lang\n...``` 代码块（lang 可空）。
_FENCE_RE = re.compile(r"```([^\n`]*)\r?\n(.*?)```", re.DOTALL)
# ── 标量/容器归一化小工具（对齐前端 str/num/arr） ───────────────────────────────
def _as_str(v: Any, fallback: str = "") -> str:
return v if isinstance(v, str) else fallback
def _as_num(v: Any) -> float | int:
if isinstance(v, bool): # bool 是 int 子类，单独挡掉
return 0
if isinstance(v, (int, float)):
n = v
else:
try:
n = float(v)
except (TypeError, ValueError):
return 0
if isinstance(n, float) and n.is_integer():
return int(n)
return n
def _as_list(v: Any) -> list[Any]:
return v if isinstance(v, list) else []
def _looks_like_report_json(s: str) -> bool:
return bool(_LOOKS_LIKE_RE.search(s))
def _collect_fenced_blocks(text: str) -> list[tuple[str, str]]:
return [(m.group(1).strip().lower(), m.group(2).strip()) for m in _FENCE_RE.finditer(text)]
def _balanced_json_objects(text: str) -> list[str]:
"""扫描出所有顶层平衡的 ``{…}`` 对象（字符串内的花括号不计），用于从夹杂文字里捞裸 JSON。"""
out: list[str] = []
depth = 0
start = -1
in_str = False
esc = False
for i, ch in enumerate(text):
if in_str:
if esc:
esc = False
elif ch == "\\":
esc = True
elif ch == '"':
in_str = False
continue
if ch == '"':
in_str = True
elif ch == "{":
if depth == 0:
start = i
depth += 1
elif ch == "}" and depth > 0:
depth -= 1
if depth == 0 and start >= 0:
out.append(text[start : i + 1])
start = -1
return out
# ── schema coercion（对齐前端 coerceReportData） ───────────────────────────────
def coerce_report_data(raw: Any) -> dict | None:
"""把已 ``json.loads`` 的对象归一化成大屏契约 dict（缺字段给安全默认；非法 → None）。
规则与前端 ``coerceReportData`` 一致：seats 至少要有 name 才保留；dependsOn 仅保留
真实存在且非自指的席位 id；kpis/charts/timeline 过滤空项并限量；整体既无 seats 又无
最终结论 headline 时判为「不像本契约」→ None。
"""
if not isinstance(raw, dict):
return None
seats: list[dict] = []
for i, s in enumerate(_as_list(raw.get("seats"))):
so = s if isinstance(s, dict) else {}
name = _as_str(so.get("name")).strip()
if not name:
continue
sid = _as_str(so.get("id")).strip() or f"seat-{i + 1}"
sections = []
for sec in _as_list(so.get("sections")):
co = sec if isinstance(sec, dict) else {}
heading = _as_str(co.get("heading"))
bullets = [b for b in (_as_str(x) for x in _as_list(co.get("bullets"))) if b]
if heading or bullets:
sections.append({"heading": heading, "bullets": bullets})
metrics = []
for m in _as_list(so.get("metrics")):
mo = m if isinstance(m, dict) else {}
label = _as_str(mo.get("label"))
if label:
metrics.append({"label": label, "value": min(100, max(0, _as_num(mo.get("value"))))})
risks = []
for r in _as_list(so.get("risks")):
ro = r if isinstance(r, dict) else {}
label = _as_str(ro.get("label"))
if not label:
continue
level = _as_str(ro.get("level"))
risks.append({"label": label, "level": level if level in ("high", "mid", "low") else "mid"})
depends_on = [d for d in (_as_str(x).strip() for x in _as_list(so.get("dependsOn"))) if d]
seats.append(
{
"id": sid,
"name": name,
"summary": _as_str(so.get("summary")),
"sections": sections,
"metrics": metrics,
"risks": risks,
"dependsOn": depends_on,
}
)
# dependsOn 只保留真实存在的席位 id（模型可能编 id），且不自指。
id_set = {s["id"] for s in seats}
for s in seats:
s["dependsOn"] = [d for d in s["dependsOn"] if d in id_set and d != s["id"]]
consensus_raw = raw.get("consensus") if isinstance(raw.get("consensus"), dict) else {}
consensus = {
"headline": _as_str(consensus_raw.get("headline")),
"points": [p for p in (_as_str(x) for x in _as_list(consensus_raw.get("points"))) if p],
}
kpis = []
for k in _as_list(raw.get("kpis")):
ko = k if isinstance(k, dict) else {}
label = _as_str(ko.get("label"))
value = _as_str(ko.get("value"), str(ko.get("value")) if ko.get("value") is not None else "")
if not (label and value):
continue
item: dict[str, Any] = {"label": label, "value": value}
unit = _as_str(ko.get("unit"))
if unit:
item["unit"] = unit
trend = _as_str(ko.get("trend"))
if trend in ("up", "down", "flat"):
item["trend"] = trend
kpis.append(item)
if len(kpis) >= 6:
break
charts = []
for c in _as_list(raw.get("charts")):
co = c if isinstance(c, dict) else {}
ctype = _as_str(co.get("type"))
items = []
for it in _as_list(co.get("items")):
io = it if isinstance(it, dict) else {}
label = _as_str(io.get("label"))
if label:
items.append({"label": label, "value": _as_num(io.get("value"))})
if not items:
continue
charts.append(
{
"type": ctype if ctype in ("bar", "donut", "radar", "line") else "bar",
"title": _as_str(co.get("title")),
"items": items,
}
)
if len(charts) >= 3:
break
timeline = []
for t in _as_list(raw.get("timeline")):
to = t if isinstance(t, dict) else {}
label = _as_str(to.get("label"))
if label:
timeline.append({"label": label, "desc": _as_str(to.get("desc"))})
data = {
"title": _as_str(raw.get("title"), "方案研讨大屏"),
"subtitle": _as_str(raw.get("subtitle")),
"kpis": kpis,
"consensus": consensus,
"charts": charts,
"timeline": timeline,
"seats": seats,
}
if not data["seats"] and not data["consensus"]["headline"]:
return None
return data
# ── 多策略抽取 ───────────────────────────────────────────────────────────────
def extract_report_json(text: str | None) -> str | None:
"""从模型回复里健壮抽取大屏 report-json，返回**归一化后的 JSON 字符串**（失败 → None）。
候选顺序（对齐前端 ``extractReportData``）：
1. 代码块，按 lang 优先级 report-json > json/jsonc > 无标注 > 其它，同级取靠后的；
2. 未闭合围栏（流式截断）：最后一个 ```report-json 之后到结尾；
3. 夹在文字里的所有平衡 ``{…}`` 对象，契约形状的优先、靠后的先试。
每个候选 ``json.loads`` 成功后过 :func:`coerce_report_data`，命中即返回归一化 JSON。
"""
if not text:
return None
def rank(lang: str) -> int:
if lang == "report-json":
return 0
if lang in ("json", "jsonc"):
return 1
if lang == "":
return 2
return 3
candidates: list[str] = []
blocks = list(enumerate(_collect_fenced_blocks(text)))
blocks.sort(key=lambda bi: (rank(bi[1][0]), -bi[0]))
candidates.extend(body for _, (_, body) in blocks)
# 未闭合围栏（流式截断）：取最后一个 report-json fence 之后到结尾。
f = text.rfind(_FENCE)
if f >= 0:
candidates.append(text[f + len(_FENCE) :].strip())
# 裸 JSON（无围栏 / 夹在文字里）：契约形状的先试，再试其余；都按靠后优先。
objs = _balanced_json_objects(text)
candidates.extend(o for o in reversed(objs) if _looks_like_report_json(o))
candidates.extend(reversed(objs))
for body in candidates:
if not body:
continue
try:
raw = json.loads(body)
except (ValueError, TypeError):
continue
data = coerce_report_data(raw)
if data is not None:
return json.dumps(data, ensure_ascii=False)
return None