fengxiangdebaoziandClaude Opus 4.8
0890b3c375
- 新增专职「结构化抽取智能体」roundtable-structure（种子资产 SOUL+config，注册进 seeder 启动 seed+sync 进 DB，可在智能体管理页编辑、未来挂入库技能）。 - /api/flow-extract/stream 改用该 agent 的 SOUL(load_agent_soul，缺失回退内置默认) + 模型(前端所选 > agent 配置 > 默认)；抽取/校验/编辑统一。SOUL 先输出简要分析过程再出 flow-json，模型慢时也有实时反馈（含 <think> 思考流）。 - 新增 /api/flow-extract/intent 意图识别（参考 ai_writing 一次性 JSON 范式）：识别 重做报告/改流程图/重画流程图/提问/模糊；前端按意图分流，模糊则发起澄清让用户点选， 不再用关键词硬路由（修「说重做报告却出流程图」）。 - Phase 2 问答修改：edit 模式(当前图+指令→改图→校验)；Phase 3 画布手动改/删节点。 - 修：用户消息发送即上屏(不等模型/分类返回)；regenerate_report 不再清空对话历史； 全屏 portal 到 body 工具栏恢复；解析兼容独立 edges 数组。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
626 lines
28 KiB
Python
626 lines
28 KiB
Python
"""流程图结构化抽取 + 校验 的 SSE 端点。
|
||
|
||
第三步「方案总结」之后，要把报告 / 各席位交付的层级结论抽成一张**流程图**（节点 + 连线）。
|
||
之前让总结智能体在同一对话线程里顺手吐 ``flow-json`` 很不稳：业务链 ``coordinator_prompt``
|
||
常写着「输出结构化结论 / 按阶段产出报告」，模型被线程上下文锚定，要么不出 flow-json、要么
|
||
结构残缺（只剩两三个节点）。
|
||
|
||
这里改成**独立的一次性模型调用**（不走 lead_agent / 不带线程历史），两段式：
|
||
① 抽取（``model.astream``）：把报告/席位交付按层级抽成 flow-json，**流式**把过程吐给前端实时显示；
|
||
② 校验/修复（``model.ainvoke``）：把抽取结果连同源材料再过一遍模型，补全漏层、修正连边、
|
||
去掉阶段/汇总节点，产出干净的 flow-json；后端再做容错解析 + 结构校验，返回最终图数据。
|
||
|
||
SSE 帧（``data: {json}\n\n``）：
|
||
- ``{"event":"extract_delta","content":"…"}`` 抽取流式增量（前端把 JSON 实时显示出来）
|
||
- ``{"event":"extract_done"}`` 抽取结束，进入校验
|
||
- ``{"event":"validating"}`` 校验中
|
||
- ``{"event":"final","graph":{title,nodes,edges}}`` 校验后的最终图（前端据此绘制）
|
||
- ``{"event":"error","detail":"…"}``
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from collections.abc import AsyncIterator
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Body, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from langchain_core.messages import HumanMessage, SystemMessage
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.gateway.routers._roundtable_seed import (
|
||
STRUCTURE_AGENT_ID,
|
||
ensure_roundtable_functional_agents,
|
||
)
|
||
from deerflow.config.agents_config import load_agent_config, load_agent_soul
|
||
from deerflow.models.factory import create_chat_model
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/flow-extract", tags=["flow-extract"])
|
||
|
||
|
||
# ── schemas ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
class SeatDelivery(BaseModel):
|
||
sender: str = ""
|
||
content: str = ""
|
||
|
||
|
||
class FlowExtractRequest(BaseModel):
|
||
# 数据来源：有报告优先用报告，否则用各席位交付。
|
||
report_md: str | None = Field(default=None)
|
||
seat_deliveries: list[SeatDelivery] = Field(default_factory=list)
|
||
# 层级骨架：优先 coordinator_prompt 里写明的层级，否则用 chain_levels。
|
||
chain_levels: list[str] = Field(default_factory=list)
|
||
coordinator_prompt: str | None = Field(default=None)
|
||
model: str | None = Field(default=None)
|
||
# Phase 2 问答修改：带上当前图 + 用户指令 → 进 edit 模式（按指令改图，其余不变）。
|
||
current_graph: dict[str, Any] | None = Field(default=None)
|
||
instruction: str | None = Field(default=None)
|
||
|
||
|
||
class FlowIntentRequest(BaseModel):
|
||
"""第三步输入框意图识别：判断用户想 重做报告 / 改流程图 / 重画流程图 / 提问 / 模糊待澄清。"""
|
||
|
||
message: str = ""
|
||
has_report: bool = False
|
||
has_flow: bool = False
|
||
model: str | None = Field(default=None)
|
||
|
||
|
||
# 意图枚举（与前端路由一一对应）。
|
||
_INTENTS = {"regenerate_report", "modify_flow", "regenerate_flow", "qa", "unclear"}
|
||
|
||
|
||
# ── flow-json 容错解析（Python 版，对齐前端 lib/flow-graph.ts）──────────────
|
||
|
||
_FENCE = "```flow-json"
|
||
_VALID_TYPES = {"start", "end", "stage", "task", "decision", "risk", "deliverable", "role"}
|
||
|
||
|
||
def _as_str(v: Any) -> str:
|
||
return v.strip() if isinstance(v, str) else ""
|
||
|
||
|
||
def _loose_nodes(body: str) -> list[dict[str, Any]]:
|
||
"""正则逐字段抽取节点（strict json.loads 失败时回退）。容忍裸换行 / 尾逗号。"""
|
||
# 取 "nodes" 后的 [ ... ]（到与之配对的 ] 或文本结尾）。
|
||
key = body.find('"nodes"')
|
||
if key < 0:
|
||
return []
|
||
arr = body.find("[", key)
|
||
if arr < 0:
|
||
return []
|
||
# 平衡括号切出每个顶层 {…}
|
||
slices: list[str] = []
|
||
depth = 0
|
||
start = -1
|
||
in_str = False
|
||
esc = False
|
||
for i in range(arr + 1, len(body)):
|
||
ch = body[i]
|
||
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 == "}":
|
||
depth -= 1
|
||
if depth == 0 and start >= 0:
|
||
slices.append(body[start : i + 1])
|
||
start = -1
|
||
elif ch == "]" and depth == 0:
|
||
break
|
||
out: list[dict[str, Any]] = []
|
||
for sl in slices:
|
||
out.append(_loose_one(sl))
|
||
return out
|
||
|
||
|
||
def _loose_edges(body: str) -> list[dict[str, Any]]:
|
||
"""非法 JSON 回退路径下，从顶层 `"edges":[ {from,to,label} … ]` 容错抽边。"""
|
||
key = body.find('"edges"')
|
||
if key < 0:
|
||
return []
|
||
arr = body.find("[", key)
|
||
if arr < 0:
|
||
return []
|
||
depth = 0
|
||
start = -1
|
||
in_str = False
|
||
esc = False
|
||
slices: list[str] = []
|
||
for i in range(arr + 1, len(body)):
|
||
ch = body[i]
|
||
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 == "}":
|
||
depth -= 1
|
||
if depth == 0 and start >= 0:
|
||
slices.append(body[start : i + 1])
|
||
start = -1
|
||
elif ch == "]" and depth == 0:
|
||
break
|
||
|
||
def pick(slice_: str, *keys: str) -> str:
|
||
for k in keys:
|
||
m = re.search(r'"' + k + r'"\s*:\s*"([^"]*)"', slice_)
|
||
if m:
|
||
return m.group(1)
|
||
return ""
|
||
|
||
out: list[dict[str, Any]] = []
|
||
for sl in slices:
|
||
frm = pick(sl, "from", "source")
|
||
to = pick(sl, "to", "target")
|
||
if frm and to:
|
||
e: dict[str, Any] = {"from": frm, "to": to}
|
||
lbl = pick(sl, "label")
|
||
if lbl:
|
||
e["label"] = lbl
|
||
out.append(e)
|
||
return out
|
||
|
||
|
||
def _loose_one(slice_: str) -> dict[str, Any]:
|
||
def pick(*keys: str) -> str | None:
|
||
for k in keys:
|
||
m = re.search(r'"' + k + r'"\s*:\s*"((?:\\.|[^"\\])*)"', slice_)
|
||
if m:
|
||
raw = m.group(1)
|
||
try:
|
||
return json.loads('"' + raw.replace("\n", "\\n").replace("\r", "") + '"')
|
||
except Exception: # noqa: BLE001
|
||
return raw.replace("\\n", "\n").replace('\\"', '"')
|
||
return None
|
||
|
||
obj: dict[str, Any] = {}
|
||
for k, dst in (("id", "id"), ("type", "type"), ("detail", "detail")):
|
||
v = pick(k)
|
||
if v:
|
||
obj[dst] = v
|
||
label = pick("label", "name")
|
||
if label:
|
||
obj["label"] = label
|
||
group = pick("group", "level", "tier", "layer")
|
||
if group:
|
||
obj["group"] = group
|
||
fm = re.search(r'"(?:from|prev)"\s*:\s*\[([\s\S]*?)\]', slice_)
|
||
if fm:
|
||
inner = fm.group(1)
|
||
ids: list[Any] = [{"id": x} for x in re.findall(r'\{[^{}]*?"id"\s*:\s*"([^"]+)"[^{}]*?\}', inner)]
|
||
if not ids:
|
||
ids = re.findall(r'"([^"]+)"', inner)
|
||
if ids:
|
||
obj["from"] = ids
|
||
return obj
|
||
|
||
|
||
def _norm_node(raw: dict[str, Any]) -> dict[str, Any] | None:
|
||
nid = _as_str(raw.get("id"))
|
||
label = _as_str(raw.get("label")) or _as_str(raw.get("name"))
|
||
if not nid or not label:
|
||
return None
|
||
rtype = _as_str(raw.get("type")).lower()
|
||
ntype = rtype if rtype in _VALID_TYPES else "task"
|
||
node: dict[str, Any] = {"id": nid, "label": label, "type": ntype}
|
||
detail = _as_str(raw.get("detail")) or _as_str(raw.get("description"))
|
||
if detail:
|
||
node["detail"] = detail
|
||
group = _as_str(raw.get("group")) or _as_str(raw.get("level")) or _as_str(raw.get("tier")) or _as_str(raw.get("layer"))
|
||
if group:
|
||
node["group"] = group
|
||
return node
|
||
|
||
|
||
def _edges_from(raw: dict[str, Any], nid: str) -> list[dict[str, Any]]:
|
||
frm = raw.get("from") or raw.get("prev")
|
||
if not isinstance(frm, list):
|
||
return []
|
||
out: list[dict[str, Any]] = []
|
||
for e in frm:
|
||
if isinstance(e, str) and e.strip():
|
||
out.append({"from": e.strip(), "to": nid})
|
||
elif isinstance(e, dict):
|
||
fid = _as_str(e.get("id")) or _as_str(e.get("from"))
|
||
if fid:
|
||
edge = {"from": fid, "to": nid}
|
||
lbl = _as_str(e.get("label"))
|
||
if lbl:
|
||
edge["label"] = lbl
|
||
out.append(edge)
|
||
return out
|
||
|
||
|
||
def parse_flow_graph(text: str) -> dict[str, Any] | None:
|
||
"""从（可能带 fence / 非法 JSON 的）文本里解析出 ``{title, nodes, edges}``。无节点→None。"""
|
||
start = text.rfind(_FENCE)
|
||
body = text[start + len(_FENCE) :] if start >= 0 else text
|
||
close = body.find("```")
|
||
if close >= 0:
|
||
body = body[:close]
|
||
body = body.strip()
|
||
|
||
title = ""
|
||
nodes_raw: list[dict[str, Any]] = []
|
||
obj: Any = None
|
||
for candidate in (body, re.sub(r",(\s*[}\]])", r"\1", body)): # 原文 + 去尾逗号
|
||
try:
|
||
obj = json.loads(candidate)
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
obj = None
|
||
if isinstance(obj, dict):
|
||
title = _as_str(obj.get("title"))
|
||
raw_list = obj.get("nodes")
|
||
nodes_raw = raw_list if isinstance(raw_list, list) else []
|
||
else:
|
||
m = re.search(r'"title"\s*:\s*"([^"]*)"', body)
|
||
if m:
|
||
title = m.group(1)
|
||
nodes_raw = _loose_nodes(body)
|
||
|
||
nodes: list[dict[str, Any]] = []
|
||
edges: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for raw in nodes_raw:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
node = _norm_node(raw)
|
||
if not node or node["id"] in seen:
|
||
continue
|
||
seen.add(node["id"])
|
||
nodes.append(node)
|
||
edges.extend(_edges_from(raw, node["id"]))
|
||
if not nodes:
|
||
return None
|
||
|
||
# 兼容**独立 `edges` 数组**：模型可能把连线写成节点上的 `from`，也可能写成顶层
|
||
# `edges:[{from,to,label}]`（或 source/target 别名）。后者若不处理，节点全无连线（bug）。
|
||
raw_edges: Any = obj.get("edges") if isinstance(obj, dict) else None
|
||
if not isinstance(raw_edges, list):
|
||
raw_edges = _loose_edges(body)
|
||
for e in raw_edges or []:
|
||
if not isinstance(e, dict):
|
||
continue
|
||
frm = _as_str(e.get("from")) or _as_str(e.get("source"))
|
||
to = _as_str(e.get("to")) or _as_str(e.get("target"))
|
||
if not frm or not to:
|
||
continue
|
||
edge: dict[str, Any] = {"from": frm, "to": to}
|
||
lbl = _as_str(e.get("label"))
|
||
if lbl:
|
||
edge["label"] = lbl
|
||
edges.append(edge)
|
||
# 只保留两端都存在的边 + 去重。
|
||
valid_edges: list[dict[str, Any]] = []
|
||
keys: set[str] = set()
|
||
for e in edges:
|
||
if e["from"] not in seen or e["to"] not in seen or e["from"] == e["to"]:
|
||
continue
|
||
k = f'{e["from"]}->{e["to"]}'
|
||
if k in keys:
|
||
continue
|
||
keys.add(k)
|
||
valid_edges.append(e)
|
||
return {"title": title, "nodes": nodes, "edges": valid_edges}
|
||
|
||
|
||
# ── prompts ────────────────────────────────────────────────────────────────
|
||
|
||
_SHAPE = (
|
||
"flow-json 严格用下面这个形状（一个 JSON 对象，nodes 是节点数组；示例仅示意结构、勿照抄内容）：\n"
|
||
"```flow-json\n"
|
||
'{"title":"流程图标题","nodes":[\n'
|
||
' {"id":"n1","label":"任务简短名","type":"start","group":"任务","detail":"任务目标"},\n'
|
||
' {"id":"n2","label":"目的核心短语","type":"task","group":"目的","detail":"名称+理由要点(2-4句,单行)","from":["n1"]},\n'
|
||
' {"id":"n3","label":"行为体核心短语","type":"role","group":"行为体","detail":"名称+理由要点","from":["n2"]}\n'
|
||
"]}\n"
|
||
"```\n"
|
||
"字段：id(n1/n2…唯一)、label(≤12字,节点标题)、type(start/end/stage/task/decision/risk/deliverable/role 之一,仅决定配色)、"
|
||
"group(该节点所在层级名称,如 任务/目的/行为体/预期行为/驱动因素/叙事,必填)、"
|
||
"detail(该条目名称+理由要点,约2-4句,**写成一行**不要内含换行)、"
|
||
"from(前驱节点 id 列表,只引用前面已出现的节点)。"
|
||
)
|
||
|
||
_STRUCTURE = (
|
||
"【画法·严格遵守】"
|
||
"① 根节点 n1=本次任务本身(报告/任务总览里的任务名称),label 用任务简短名,group 填「任务」,不要用「任务」泛称当 label;"
|
||
"② 从第二层起按层级逐层展开:每层在报告/交付里通常列了多条具体条目(如 目的层有 目的1/2/3、行为体层有各自对应行为体,后面还有 预期行为·驱动因素、叙事…),"
|
||
"把每一条具体条目各画成一个节点,不要把整层合并成一个笼统节点;"
|
||
"③ 每个节点 group 填它所在层级名称(取自报告里该层级真实名称);"
|
||
"④ 连边严格按报告里的对应关系串成分支:根任务→各目的;每个目的→它对应的行为体;每个行为体→对应的预期行为·驱动因素→对应的叙事……每组各成一条自上而下的分支(有几组就几条,组间不混连);"
|
||
"⑤ 务必把每层每一条都输出成节点、不要漏层、不要只输出开头一两个节点就停(典型约13-16个节点);"
|
||
"⑥ 只画这些层级实体节点,不要把『综合分析/核心洞察/执行路线/建议/结论』及『阶段①②③④』等汇总段落/阶段名画成节点。"
|
||
)
|
||
|
||
|
||
def _material(req: FlowExtractRequest) -> str:
|
||
report = (req.report_md or "").strip()
|
||
if report:
|
||
return "===== 方案总结报告全文（数据来源，逐条抽取，勿改写）=====\n" + report + "\n===== 报告结束 ====="
|
||
seats = [s for s in req.seat_deliveries if (s.content or "").strip()]
|
||
if seats:
|
||
body = "\n\n".join(f"## 席位{i + 1}：{s.sender or '（未具名席位）'}\n{s.content.strip()}" for i, s in enumerate(seats))
|
||
return "===== 各席位交付产物（数据来源，逐条抽取，勿改写）=====\n" + body + "\n===== 交付结束 ====="
|
||
return ""
|
||
|
||
|
||
def _hierarchy_hint(req: FlowExtractRequest) -> str:
|
||
if (req.coordinator_prompt or "").strip():
|
||
return "层级顺序优先采用下面「业务链总控编排提示」里写明的层级链(如它写的 任务→目的→行为体→…):\n" + req.coordinator_prompt.strip()
|
||
levels = [s.strip() for s in req.chain_levels if s.strip()]
|
||
if levels:
|
||
return "层级顺序(自上而下)：任务 → " + " → ".join(levels) + "。"
|
||
return "层级顺序：以「任务」为根，按报告/交付的逻辑结构合理分层展开。"
|
||
|
||
|
||
# 兜底系统提示：当 roundtable-structure 的 SOUL 缺失时用它（与 SOUL 同款契约）。
|
||
_DEFAULT_SYSTEM = (
|
||
"你是「结构化抽取智能体」。职责：读懂《方案总结报告》/ 各席位交付内容，分析其层级结构，"
|
||
"抽取成可交互流程图的结构化数据(flow-json)；也按指令改图、校验修复。"
|
||
"请**先用简短文字(2-4句)说明分析过程**(识别的层级链、每层条目数、分支对应)让用户看到进展，"
|
||
"**再**在回复末尾输出一个完整的 ```flow-json 代码块（除这段简短说明外不写长篇正文）。全程简体中文。\n"
|
||
+ _SHAPE
|
||
+ "\n"
|
||
+ _STRUCTURE
|
||
)
|
||
|
||
|
||
def _system_prompt() -> str:
|
||
"""结构化抽取的系统提示 = roundtable-structure 智能体的 SOUL（可在智能体管理页编辑、
|
||
未来挂入库技能）；SOUL 缺失则回退内置 _DEFAULT_SYSTEM。"""
|
||
soul = load_agent_soul(STRUCTURE_AGENT_ID)
|
||
return soul.strip() if soul and soul.strip() else _DEFAULT_SYSTEM
|
||
|
||
|
||
def _resolve_model(payload_model: str | None) -> str | None:
|
||
"""模型优先级：前端所选(payload) > 智能体配置的 model > None(用 config 默认模型)。"""
|
||
if payload_model:
|
||
return payload_model
|
||
try:
|
||
cfg = load_agent_config(STRUCTURE_AGENT_ID)
|
||
if cfg and cfg.model:
|
||
return cfg.model
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return None
|
||
|
||
|
||
def _extract_messages(req: FlowExtractRequest) -> list[Any]:
|
||
user = (
|
||
_material(req)
|
||
+ "\n\n"
|
||
+ _hierarchy_hint(req)
|
||
+ "\n\n请分析以上数据来源，按你的职责设定输出一个完整的 ```flow-json 代码块。"
|
||
)
|
||
return [SystemMessage(content=_system_prompt()), HumanMessage(content=user)]
|
||
|
||
|
||
def _edit_messages(req: FlowExtractRequest) -> list[Any]:
|
||
"""Phase 2 问答修改：按用户指令改当前流程图（增/删/改节点或连线），其余保持不变。"""
|
||
cur = json.dumps(req.current_graph or {}, ensure_ascii=False)
|
||
user = (
|
||
_material(req)
|
||
+ "\n\n===== 当前流程图（flow-json，待按指令修改）=====\n"
|
||
+ cur
|
||
+ "\n===== 当前流程图结束 =====\n\n用户修改指令：\n"
|
||
+ (req.instruction or "").strip()
|
||
+ "\n\n请按指令修改这张流程图：未被指令涉及的部分保持原样、节点 id 尽量沿用、删节点连带删边，"
|
||
+ "输出**修改后的完整** ```flow-json 代码块。"
|
||
)
|
||
return [SystemMessage(content=_system_prompt()), HumanMessage(content=user)]
|
||
|
||
|
||
def _validate_messages(req: FlowExtractRequest, extracted: str) -> list[Any]:
|
||
user = (
|
||
_material(req)
|
||
+ "\n\n===== 已抽取的 flow-json（待校验修复）=====\n"
|
||
+ extracted
|
||
+ "\n===== 待校验结束 =====\n\n请对照数据来源校验并修复这份 flow-json："
|
||
+ "根节点是任务本身、每层每条都在没漏、连边按对应关系成分支无悬空边、每个节点都有 group、"
|
||
+ "没有汇总/阶段编号节点；发现问题就改对、没问题就原样规范化，输出修复后的完整 ```flow-json 代码块。"
|
||
)
|
||
return [SystemMessage(content=_system_prompt()), HumanMessage(content=user)]
|
||
|
||
|
||
# ── endpoint ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _frame(data: dict[str, Any]) -> str:
|
||
return f"data: {json.dumps(data, ensure_ascii=False, default=str)}\n\n"
|
||
|
||
|
||
def _chunk_parts(chunk: Any) -> tuple[str, str]:
|
||
"""从流式 chunk 取 (正文增量, 思考增量)。思考走 additional_kwargs.reasoning_content
|
||
（DeepSeek/兼容 OpenAI 推理模型的单列字段），正文走 content。"""
|
||
content = chunk.content if isinstance(getattr(chunk, "content", None), str) else ""
|
||
ak = getattr(chunk, "additional_kwargs", None) or {}
|
||
reasoning = ""
|
||
if isinstance(ak, dict):
|
||
r = ak.get("reasoning_content") or ak.get("reasoning")
|
||
if isinstance(r, str):
|
||
reasoning = r
|
||
return content, reasoning
|
||
|
||
|
||
@router.post("/stream")
|
||
async def extract_flow_stream(request: Request, payload: FlowExtractRequest = Body(...)) -> StreamingResponse:
|
||
"""流式抽取流程图 JSON + 流式校验修复（抽取/校验过程含思考都实时吐前端）→ 返回最终结构化图。"""
|
||
|
||
ensure_roundtable_functional_agents() # 自愈：确保 roundtable-structure 的 SOUL/config 存在
|
||
resolved_model = _resolve_model(payload.model)
|
||
|
||
async def gen() -> AsyncIterator[str]:
|
||
try:
|
||
# 带 instruction + current_graph → Phase 2 问答修改（按指令改当前图）；否则首次抽取。
|
||
is_edit = bool((payload.instruction or "").strip()) and payload.current_graph is not None
|
||
if not _material(payload).strip() and not is_edit:
|
||
yield _frame({"event": "error", "detail": "缺少数据来源（报告或各席位交付均为空）"})
|
||
return
|
||
|
||
# ① 抽取 / 编辑（正文 + 思考(reasoning) 拼成 `<think>…</think>正文` 增量逐帧吐，前端实时显示）
|
||
round1_messages = _edit_messages(payload) if is_edit else _extract_messages(payload)
|
||
model = create_chat_model(name=resolved_model, thinking_enabled=True)
|
||
extracted = ""
|
||
think_open = False
|
||
async for chunk in model.astream(round1_messages):
|
||
c, r = _chunk_parts(chunk)
|
||
out = ""
|
||
if r:
|
||
if not think_open:
|
||
out += "<think>"
|
||
think_open = True
|
||
out += r
|
||
if c:
|
||
if think_open:
|
||
out += "</think>"
|
||
think_open = False
|
||
out += c
|
||
extracted += c
|
||
if out:
|
||
yield _frame({"event": "extract_delta", "content": out})
|
||
if think_open:
|
||
yield _frame({"event": "extract_delta", "content": "</think>"})
|
||
yield _frame({"event": "extract_done"})
|
||
|
||
# ② 校验 / 修复（同样流式 + 思考实时显示）
|
||
yield _frame({"event": "validating"})
|
||
validated = ""
|
||
graph: dict[str, Any] | None = None
|
||
try:
|
||
vmodel = create_chat_model(name=resolved_model, thinking_enabled=True)
|
||
think_open = False
|
||
async for chunk in vmodel.astream(_validate_messages(payload, extracted)):
|
||
c, r = _chunk_parts(chunk)
|
||
out = ""
|
||
if r:
|
||
if not think_open:
|
||
out += "<think>"
|
||
think_open = True
|
||
out += r
|
||
if c:
|
||
if think_open:
|
||
out += "</think>"
|
||
think_open = False
|
||
out += c
|
||
validated += c
|
||
if out:
|
||
yield _frame({"event": "validate_delta", "content": out})
|
||
if think_open:
|
||
yield _frame({"event": "validate_delta", "content": "</think>"})
|
||
graph = parse_flow_graph(validated)
|
||
except Exception: # noqa: BLE001 — 校验失败不致命，回退抽取结果
|
||
logger.warning("flow-extract validation failed, fall back to extraction", exc_info=True)
|
||
graph = None
|
||
if graph is None:
|
||
graph = parse_flow_graph(extracted)
|
||
if graph is None:
|
||
yield _frame({"event": "error", "detail": "未能从模型输出解析出流程图节点"})
|
||
return
|
||
yield _frame({"event": "final", "graph": graph})
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception("flow-extract stream failed")
|
||
yield _frame({"event": "error", "detail": str(exc)})
|
||
|
||
return StreamingResponse(
|
||
gen(),
|
||
media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"},
|
||
)
|
||
|
||
|
||
# ── 意图识别（一次性 JSON，参考 ai_writing/intent_parser 范式）─────────────────
|
||
|
||
|
||
def _parse_intent_json(text: str) -> dict[str, Any] | None:
|
||
"""从模型输出里容错解析 {intent, reason}。strict → 去围栏/去尾逗号 → 平衡括号。"""
|
||
candidates: list[str] = []
|
||
t = text.strip()
|
||
candidates.append(t)
|
||
m = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", t)
|
||
if m:
|
||
candidates.append(m.group(1).strip())
|
||
s = t.find("{")
|
||
e = t.rfind("}")
|
||
if s >= 0 and e > s:
|
||
candidates.append(t[s : e + 1])
|
||
for c in candidates:
|
||
for cand in (c, re.sub(r",(\s*[}\]])", r"\1", c)):
|
||
try:
|
||
obj = json.loads(cand)
|
||
if isinstance(obj, dict):
|
||
return obj
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
return None
|
||
|
||
|
||
_INTENT_SYS = (
|
||
"你是圆桌「方案总结」第三步输入框的**意图识别器**。第三步有两类产物：方案总结报告(Markdown) 和 结构化流程图。"
|
||
"用户在输入框发了一条消息，请判断他的**真实意图**，只输出一个 JSON 对象（不要解释、不要 markdown 代码块、用英文双引号）：\n"
|
||
'{"intent":"<枚举>","reason":"<一句话依据>"}\n'
|
||
"intent 只能取其一：\n"
|
||
"- regenerate_report：想重新分析 / 重写【方案总结报告】（如「重新生成报告」「报告再写一版」「重新分析一下」）；\n"
|
||
"- modify_flow：想修改**当前流程图**的某些节点/连线（如「把目的1改成X」「删掉第3个行为体」「加个验收节点」「流程图这里调整下」）；\n"
|
||
"- regenerate_flow：想**重新生成整张**流程图（如「重新画流程图」「流程图重来」「重新结构化输出」）；\n"
|
||
"- qa：只是提问 / 咨询，不需要改任何产物（如「为什么选 OTA 平台」「这个目的什么意思」）；\n"
|
||
"- unclear：意图不明确、无法判断要改报告还是改流程图还是提问（如只说「改一下」「不对」「重来」而没指明对象）。\n"
|
||
"判断要点：结合下面给出的上下文（是否已有报告 / 是否已有流程图）；当前已有流程图时，含「节点/连线/这一层/这个目的/行为体…改/删/加」之类多半是 modify_flow；"
|
||
"明确提到「报告」才是 regenerate_report；只有完全无法判断对象时才用 unclear。\n"
|
||
"示例：\n"
|
||
'「再帮我生成一次报告」→ {"intent":"regenerate_report","reason":"明确要求重新生成报告"}\n'
|
||
'「把目的1的名称改短点」→ {"intent":"modify_flow","reason":"要求修改流程图某节点"}\n'
|
||
'「流程图重新画一张」→ {"intent":"regenerate_flow","reason":"要求重画整张流程图"}\n'
|
||
'「OTA 平台为什么是直接行为体」→ {"intent":"qa","reason":"纯提问不改产物"}\n'
|
||
'「改一下」→ {"intent":"unclear","reason":"没说改报告还是流程图"}'
|
||
)
|
||
|
||
|
||
@router.post("/intent")
|
||
async def classify_flow_intent(request: Request, payload: FlowIntentRequest = Body(...)) -> dict[str, str]:
|
||
"""识别第三步输入框消息的意图（一次模型调用）。失败 / 解析不出 → unclear，让前端发起澄清。"""
|
||
msg = (payload.message or "").strip()
|
||
if not msg:
|
||
return {"intent": "unclear", "reason": "空消息"}
|
||
ctx = f"当前上下文：{'已生成报告' if payload.has_report else '尚无报告'}；{'已生成流程图' if payload.has_flow else '尚无流程图'}。"
|
||
try:
|
||
model = create_chat_model(name=_resolve_model(payload.model), thinking_enabled=False)
|
||
res = await model.ainvoke(
|
||
[SystemMessage(content=_INTENT_SYS), HumanMessage(content=ctx + "\n\n用户消息：" + msg + "\n\n请输出 JSON：")]
|
||
)
|
||
text = res.content if isinstance(res.content, str) else _as_str(getattr(res, "text", ""))
|
||
obj = _parse_intent_json(text) or {}
|
||
intent = _as_str(obj.get("intent")).lower()
|
||
if intent not in _INTENTS:
|
||
intent = "unclear"
|
||
return {"intent": intent, "reason": _as_str(obj.get("reason"))}
|
||
except Exception: # noqa: BLE001 — 分类失败不致命，回退澄清让用户自己定
|
||
logger.warning("flow intent classify failed", exc_info=True)
|
||
return {"intent": "unclear", "reason": "意图识别失败"}
|