Merge remote-tracking branch 'origin/master' into ji

# Conflicts:
#	frontend-web/src/components/workspace/messages/message-list-item.tsx
#	offline-backend-20260512/backend/app/gateway/app.py
#	offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/html_page.py
#	offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py
This commit is contained in:
吉鑫升 committed 2026-06-03 19:01:45 +08:00
commit a1d460e220
133 files changed
+12413 -1239

No files matched your search

+1 -1
View File
@@ -17,7 +17,7 @@ SERPER_API_KEY=your-serper-api-key
# AUTH_JWT_SECRET must stay stable across restarts, or existing logins expire.
# GATEWAY_CORS_ORIGINS must include the exact frontend origins, separated by commas.
AUTH_JWT_SECRET=i3_BiBK44sYHiFIkChhDs46y57ZewJKO3diQDw01anI
GATEWAY_CORS_ORIGINS=http://localhost:5173,http://127.0.0.1:5173
GATEWAY_CORS_ORIGINS=http://localhost:5173,http://127.0.0.1:5173,http://localhost:5174,http://127.0.0.1:5174
GATEWAY_CORS_ALLOW_ALL=0
GATEWAY_ENABLE_DOCS=true
GATEWAY_WORKERS=2
@@ -40,9 +40,11 @@ from app.gateway.routers import (
models,
multi_agent,
notifications,
public_embed,
public_scheduled_tasks,
recommend,
recommended_questions,
roundtable_chains,
roundtable_drafts,
runs,
scheduled_tasks,
@@ -806,6 +808,10 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
dev_origins = [
"http://localhost:5173",
"http://127.0.0.1:5173",
"http://localhost:5174",
"http://127.0.0.1:5174",
"http://localhost:3000",
"http://127.0.0.1:3000",
]
logger.warning(
"No CORS configuration found; enabling local dev origins only: %s. "
@@ -854,6 +860,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Roundtable-planning drafts + recommendation history at /api/roundtable-drafts
app.include_router(roundtable_drafts.router)
# Roundtable-planning business chains at /api/roundtable-chains
app.include_router(roundtable_chains.router)
# Suggestions API is mounted at /api/threads/{thread_id}/suggestions
app.include_router(suggestions.router)
@@ -871,6 +880,8 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# surface results without a DeerFlow login.
app.include_router(public_scheduled_tasks.router)
# Public embed Q&A status polling (/api/public/embed/*)
app.include_router(public_embed.router)
app.include_router(html_page_favorites.router)
# System settings API is mounted at /api/system-settings
@@ -134,6 +134,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.thread_share_store = make_thread_share_store(sf)
from deerflow.persistence.embed_sessions import make_embed_session_store
app.state.embed_session_store = make_embed_session_store(sf)
from deerflow.persistence.ai_writing_sessions import (
make_ai_writing_session_store,
register_default_store as _register_ai_writing_default_store,
@@ -160,6 +164,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.roundtable_draft_store = make_roundtable_draft_store(sf)
from deerflow.persistence.roundtable_chains import make_roundtable_chain_store
app.state.roundtable_chain_store = make_roundtable_chain_store(sf)
# Run event store (has its own factory with config-driven backend selection)
run_events_config = getattr(config, "run_events", None)
app.state.run_event_store = make_run_event_store(run_events_config)
@@ -273,6 +281,14 @@ def get_thread_share_store(request: Request):
return val
def get_embed_session_store(request: Request):
"""Return the iframe embed session → thread mapping store."""
val = getattr(request.app.state, "embed_session_store", None)
if val is None:
raise HTTPException(status_code=503, detail="Embed session store not available")
return val
def get_ai_writing_session_store(request: Request):
"""Return the AI writing session store, or None when not configured."""
return getattr(request.app.state, "ai_writing_session_store", None)
@@ -60,16 +60,21 @@ logger = logging.getLogger(__name__)
INTENT_AGENT_ID = "roundtable-intent"
RECOMMENDER_AGENT_ID = "roundtable-recommender"
COORDINATOR_AGENT_ID = "roundtable-coordinator"
# Step 3「结果绘制」用的方案可视化总结智能体:读各席位交付 + 共识,画自包含图表 HTML。
# 与上面三个一样是内置单例,种子化一次常驻磁盘+DB,绝不每次调用重建。
REPORT_AGENT_ID = "roundtable-report"
# 验证版 SOUL.md + config.yaml 的源码资源目录(随 git 发布)。每个子目录名即 agent id,
# 内含该 agent 的 config.yaml 与 SOUL.md,是 seeder 落盘时字节级复制的来源。
_ASSETS_DIR = Path(__file__).parent / "_roundtable_seed_assets"
# 这三个功能型 agent 任何部署都必须存在;顺序无关紧要(逐个独立补建)。
# 这几个功能型/内置 agent 任何部署都必须存在;顺序无关紧要(逐个独立补建)。
# 前三个支撑 Step 1 澄清 / 推荐 / Step 2 总控;roundtable-report 支撑 Step 3 结果绘制。
_FUNCTIONAL_AGENT_IDS: tuple[str, ...] = (
INTENT_AGENT_ID,
RECOMMENDER_AGENT_ID,
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
)
# seeder 会复制资源目录里的这些文件(存在即复制,缺失则跳过)。
@@ -0,0 +1,76 @@
# 圆桌方案可视化研讨汇报智能体
## 核心定位
你是多智能体圆桌研讨 **Step 3** 的「方案可视化研讨汇报智能体」。各席位(情报收集 / 环境评估 / 方案设计 / 风险审查 / 执行规划 / 总结陈述等)已经完成研讨并产出交付,总控也给出了最终共识。
你的**唯一职责**:把这些文字成果归纳、提炼,**撰写成一份图文并茂、以文字叙述为主线、图表点睛的 HTML 研讨汇报**,用 `write_file` 工具写入到 `/mnt/user-data/outputs/方案总览.html`。
这份汇报是给**用户阅读**的——目的是让用户能清晰、轻松地看懂「这群智能体讨论出了什么方案、为什么是它」。所以**可读性第一**:用通顺的中文把研讨脉络和方案讲清楚,再用图表辅助理解。
你**不**调研、**不**联网搜索、**不**派活、**不**追问用户——只做归纳、撰写与可视化。
## 输入说明
用户消息里会给你三部分原料(可能有缺省):
1. **任务意图摘要**:目标、约束、关键假设(来自 Step 1);
2. **各席位最终交付**:每个席位的名称 + 它产出的完整 markdown 正文;
3. **总控最终共识结论**。
请通读全部原料:一方面**提炼每个席位的核心观点/交付**用于文字叙述;另一方面**抽取可量化、可结构化的信息**(指标、阶段、里程碑、风险项、预算/资源占比、优劣势等)来驱动图表。
## 产出要求(最重要)
### 1. 必须是「自包含、离线可用」的单文件 HTML
- 所有 CSS 内联在 `<style>`,所有图表用**纯 inline SVG** 直接画(坐标、路径、文字都由你计算好写死);
- **绝对禁止**引用任何外部资源:不要 `<script src=...>` 引 CDN(Chart.js/ECharts/D3 等一律不行)、不要外链字体/图片/CSS。本系统离线部署,外链一律加载失败;
- 如确需交互/动画,只能用**内联的原生 JS**(`<script>...</script>` 里自己写),但默认优先纯 SVG 静态图,稳定优先;
- 字体用系统字体栈(`system-ui, -apple-system, "Microsoft YaHei", sans-serif`)。
### 2. 报告结构(**研讨叙事型**,文字为主线、图表点睛)
按下面的章节顺序组织,**文字叙述是骨架**,图表穿插在最契合的章节里辅助理解。缺料的章节可省略或合并:
| # | 章节 | 内容(文字) | 配图建议 |
|---|------|------|----------|
| ① | 报告头 | 标题 + 生成时间占位 + 「AI 自动生成」标签 | — |
| ② | 一句话结论 + 关键指标卡 | 用一句话点明最终方案/结论 | 4–6 个 KPI 数字卡(共识度、参与席位数、识别风险数、里程碑数、预估周期、预算规模等) |
| ③ | 任务背景与目标 | 2–4 句交代要解决什么、目标与关键约束 | 可选:目标/约束要点列表 |
| ④ | 各席位观点与交付 | **每个席位一段提炼文字**(这位专家关注什么、给出了什么关键结论/建议),用小标题+正文,让读者读懂每方的贡献 | 关键席位视情况配小图(如情报席位的对比、环境席位的承载评估) |
| ⑤ | 综合分析 | 把跨席位的信息串起来分析(风险全景、资源权衡、阶段安排等) | **此处是图表主战场**:风险概率×影响矩阵、资源/预算饼图、方案对比柱状、各席位贡献雷达等 |
| ⑥ | 总控最终共识方案 | **重点章节**,详写最终拍板的方案是什么、核心要点、为什么这么定 | 可选:共识达成度环形仪表 |
| ⑦ | 执行路线 | 落地步骤/阶段说明 | 甘特图 / 阶段时间线(阶段条 + 里程碑点) |
每个图表都要有**标题、图例、坐标轴或数值标注**,并在图旁配**一两句文字解读**,让图不孤立。
### 3. 图表分量:按需点睛,但不要"全是文字"
- **详略得当**:文字承载论证与叙述,图表承载可量化/可结构化的信息。不要为了凑数硬画无意义的图,也**不要因为缺数据就完全不画**。
- **没有明确数字时也要给图**:基于文字内容**合理提炼/估算**出图(如把定性的高/中/低映射成评分、把阶段描述排成时间线、把优劣势排成 SWOT 四象限),并**显式标注「定性评估 / 据研讨内容估算」**,不要凭空编造与原料矛盾的数据。
- 目标手感:全篇**穿插 3–5 个**有信息量的图表,分布在 ②⑤⑥⑦ 等章节,使页面图文并茂、读起来不枯燥。
常用图表素材库(按章节按需取用,能画哪些画哪些):
共识达成度 → 环形进度仪表;各席位贡献画像 → 雷达图(完整性/可行性/风险覆盖/创新性/落地性);风险评估 → 概率×影响气泡矩阵;执行路线 → 甘特/时间线;资源预算 → 饼图/环形图;方案权衡 → SWOT 四象限或对比柱状;决策链路 → 流程图(链条模式尤其适用)。
### 4. 视觉风格
- 主色对齐圆桌的**浅红 / 绛红**主题:主色 `#c25555`,浅底 `#fef4f3`,描边 `#ede9e9`,文字 `#1f2937`,辅助灰 `#717b88`;图表配色在主色基础上取同色系/邻近色,保持克制专业;
- 整体是**研讨汇报**版式:报告头 + 分章节卡片化分区、清晰的章节标题层级、正文行距舒适、留白充足、响应式(`max-width` 容器 + 网格自适应),打印友好;
- 文字排版讲究可读性:段落不要过长,关键结论可用强调色/加粗,列表化呈现要点。
## 工作流程
1. 通读原料,在思考里列两份清单:**各席位要点**(用于文字叙述)+ **可量化信息**(用于图表,含需要估算的项);
2. **用一次 `write_file` 把完整 HTML 文本直接写入** `/mnt/user-data/outputs/方案总览.html`(`description` 写「方案可视化研讨汇报」)。
- `write_file` **支持很长的内容,不会截断**——请放心一次性把整份 HTML 写进去;
- **严禁**写 Python / shell / 任何脚本去"生成"HTML,也**严禁**用 bash 跑脚本产出文件——你必须自己把 HTML 文本直接写进 `write_file`;
- **严禁**把 HTML 拆成多个文件或分多次拼接,就一个 `方案总览.html`;
- 控制好体量:聚焦叙述 + 3–5 个图表,别堆砌到写不完。
3. 写完后,用**一两句话**向用户说明「已生成研讨汇报,含 X 个章节、Y 个图表」即可——**不要**把 HTML 源码贴进对话,也**不要**再去读取/校验文件。
> **修改/返工规则**:如果写完后校验发现问题需要修正,请**用 `write_file` 整文件重写**到同一路径,**不要用 `str_replace` 做局部替换**。整文件重写能让右侧预览实时刷新;局部替换不会反映到预览里。
## 禁止事项
- ❌ 引用任何外部 CDN / 外链资源(离线必然失败)
- ❌ 把大段 HTML 源码输出到对话正文(只写文件 + 一句话总结)
- ❌ 联网搜索、派活给其他智能体、反问用户
- ❌ 通篇只有图表、没有像样的文字叙述(这是给用户读的汇报,不是纯看板)
- ❌ 通篇只有文字、完全不配图(缺数据也要合理估算出图并标注)
- ❌ 写 Python / shell / 任何脚本去生成 HTML,或用 bash 跑脚本产出文件(必须自己把 HTML 文本直接写进 `write_file`)
- ❌ 把 HTML 拆成多文件或分多次拼接、或写完后再去读取/校验文件
- ❌ 用 `str_replace` 局部修改 HTML(返工请整文件 `write_file` 重写)
- ❌ 编造与原料明显矛盾的数据
@@ -0,0 +1,3 @@
description: 圆桌方案可视化总结智能体,读取各席位的最终交付与总控共识,绘制一个自包含、含多种图表的 HTML 可视化看板(写入 outputs/方案总览.html)。本身不调研、不联网、不派活,只做归纳与可视化。
id: roundtable-report
name: 方案总结
@@ -59,6 +59,7 @@ from typing import Any, AsyncIterator
import httpx
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
ensure_roundtable_functional_agents,
)
from deerflow.config import get_app_config
@@ -822,6 +823,25 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
)
@router.post("/report/init")
async def init_report_thread(request: Request) -> dict[str, str]:
"""为 Step 3「结果绘制」创建一个供方案可视化总结智能体(``roundtable-report``)
运行的线程,返回 ``{"agent_name", "thread_id"}``。
设计要点:
- ``roundtable-report`` 是**内置单例**(``_roundtable_seed`` 种子化,启动即同步进
DB),这里**只建一个轻量 thread**,绝不重建 agent;
- 线程打 ``_ROUNDTABLE_THREAD_METADATA`` 标记成系统会话,从「最近对话」里隐藏;
- 入口先 ``ensure_roundtable_functional_agents()`` 自愈(磁盘目录被手删也能补建);
- 建好后前端用现有 ``POST /run/stream``(``agent_type="special"``、
``agent_name="roundtable-report"``、``d_agent_thread_id={REPORT_AGENT_ID: thread_id}``)
跑这一轮,智能体把图表 HTML 写进该线程的 ``outputs/``,再由沙箱预览。
"""
ensure_roundtable_functional_agents()
thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
return {"agent_name": REPORT_AGENT_ID, "thread_id": thread_id}
# ---------------------------------------------------------------------------
# 上游 LangGraph SSE 透传(intent / recommend / leader / special 共用)
# ---------------------------------------------------------------------------
@@ -1143,12 +1163,22 @@ async def _special_run(
"""
t0 = time.perf_counter()
thread_id = agent_threads[agent_name]
excluded_tools = ["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"]
if agent_name == REPORT_AGENT_ID:
# Step 3「结果绘制」的报告智能体只需**直接 write_file 写出完整 HTML**。
# 额外禁用 bash / str_replace:
# - 禁 bash → 杜绝它写 Python/脚本再跑出 HTML(脚本产物落在 outputs 里,
# 但前端虚拟预览只抓 write_file 的 args.content,抓不到脚本产物 → 预览停在
# 那个半截 write_file 上,正是用户看到的「没写完就去生成 py 文件」);
# - 禁 str_replace → 强制返工用整文件 write_file 重写(str_replace 没有 content
# 字段,会让前端虚拟预览读到 undefined 而清空沙箱)。
excluded_tools = [*excluded_tools, "bash", "str_replace"]
body = _run_payload(
agent_name=agent_name,
model_name=model_name,
thread_id=thread_id,
new_message=new_message,
excluded_tools=["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"],
excluded_tools=excluded_tools,
skill_stop_names=skill_stop_names,
)
t_payload_built = time.perf_counter()
@@ -0,0 +1,318 @@
"""Public (unauthenticated) API for embedded workspace chat integrations.
Parent pages embed DeerFlow in a plain iframe and cannot read ``thread_id``
from the child frame. They pass a ``session_id`` (their own correlation id)
in the iframe URL; the embed page registers ``session_id → thread_id`` here,
then the parent polls ``GET /sessions/{session_id}/status``.
"""
from __future__ import annotations
import logging
import re
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field, field_validator
from app.gateway.deps import get_checkpointer, get_embed_session_store, get_run_manager, get_thread_store
from app.gateway.routers.thread_shares import _load_thread_snapshot
from app.gateway.routers.threads import _derive_thread_status
from deerflow.persistence.embed_sessions.base import EmbedSessionStore
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/public/embed", tags=["public-embed"])
_SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]{8,128}$")
_UNFINISHED_THREAD_STATUSES = frozenset({"running", "busy"})
def compute_embed_finished(
*,
has_active_run: bool,
meta_status: str,
derived_status: str,
) -> bool:
"""Return whether an embed parent may treat the Q&A session as complete."""
if has_active_run or meta_status in _UNFINISHED_THREAD_STATUSES:
return False
if derived_status == "interrupted":
return False
return True
def _validate_session_id(session_id: str) -> str:
value = session_id.strip()
if not _SESSION_ID_RE.fullmatch(value):
raise HTTPException(
status_code=422,
detail="session_id must be 8-128 characters (letters, digits, underscore, hyphen)",
)
return value
class RegisterEmbedSessionRequest(BaseModel):
session_id: str = Field(..., description="Correlation id from the parent page (iframe query param)")
thread_id: str = Field(..., description="DeerFlow conversation thread id")
@field_validator("session_id")
@classmethod
def _check_session_id(cls, value: str) -> str:
if not _SESSION_ID_RE.fullmatch(value.strip()):
raise ValueError("session_id must be 8-128 characters (letters, digits, underscore, hyphen)")
return value.strip()
@field_validator("thread_id")
@classmethod
def _check_thread_id(cls, value: str) -> str:
trimmed = value.strip()
if not trimmed:
raise ValueError("thread_id is required")
return trimmed
class RegisterEmbedSessionResponse(BaseModel):
session_id: str
thread_id: str
registered: bool = True
class EmbedThreadStatusFields(BaseModel):
finished: bool
thread_status: str
has_active_run: bool
latest_run_status: str | None = None
class EmbedThreadStatusResponse(EmbedThreadStatusFields):
thread_id: str
class EmbedSessionStatusResponse(EmbedThreadStatusFields):
session_id: str
registered: bool = Field(description="False until the iframe has registered this session_id")
thread_id: str | None = Field(default=None, description="Resolved thread id when registered")
class EmbedSessionResultResponse(EmbedThreadStatusFields):
session_id: str
registered: bool = Field(description="False until the iframe has registered this session_id")
thread_id: str | None = Field(default=None, description="Resolved thread id when registered")
messages: list[dict] = Field(default_factory=list, description="Full serialized conversation messages")
generated_text: str = Field(default="", description="Concatenated assistant/AI message text")
generated_files: list[str] = Field(
default_factory=list,
description="Virtual artifact paths produced during the conversation (e.g. /mnt/user-data/outputs/...)",
)
class EmbedThreadResultResponse(EmbedThreadStatusFields):
thread_id: str
messages: list[dict] = Field(default_factory=list)
generated_text: str = Field(default="")
generated_files: list[str] = Field(default_factory=list)
def _message_text(message: dict) -> str:
content = message.get("content")
if isinstance(content, str):
return content.strip()
if isinstance(content, list):
parts: list[str] = []
for part in content:
if isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str):
text = part["text"].strip()
if text:
parts.append(text)
return "\n".join(parts)
return ""
def extract_generated_text(messages: list[dict]) -> str:
"""Join all assistant/AI message bodies in conversation order."""
parts: list[str] = []
for message in messages:
if not isinstance(message, dict):
continue
message_type = message.get("type") or message.get("role")
if message_type not in {"ai", "assistant"}:
continue
text = _message_text(message)
if text:
parts.append(text)
return "\n\n".join(parts)
def build_embed_result_payload(
*,
messages: list[dict],
artifacts: list[str],
) -> tuple[list[dict], str, list[str]]:
return messages, extract_generated_text(messages), artifacts
async def _result_for_thread_id(thread_id: str, request: Request) -> EmbedThreadResultResponse:
checkpointer = get_checkpointer(request)
messages, artifacts = await _load_thread_snapshot(checkpointer, thread_id)
serialized_messages, generated_text, generated_files = build_embed_result_payload(
messages=messages,
artifacts=artifacts,
)
status = await _status_for_thread_id(thread_id, request)
return EmbedThreadResultResponse(
thread_id=thread_id,
messages=serialized_messages,
generated_text=generated_text,
generated_files=generated_files,
**status.model_dump(),
)
async def _status_for_thread_id(thread_id: str, request: Request) -> EmbedThreadStatusFields:
thread_store = get_thread_store(request)
run_mgr = get_run_manager(request)
checkpointer = get_checkpointer(request)
record = await thread_store.get(thread_id, user_id=None)
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
try:
checkpoint_tuple = await checkpointer.aget_tuple(config)
except Exception:
logger.exception("Failed to read checkpoint for embed status on %s", thread_id)
raise HTTPException(status_code=500, detail="Failed to read thread state") from None
if record is None and checkpoint_tuple is None:
raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
derived_status = (
_derive_thread_status(checkpoint_tuple)
if checkpoint_tuple is not None
else (record or {}).get("status", "idle")
)
meta_status = str((record or {}).get("status") or derived_status)
has_active_run = await run_mgr.has_inflight(thread_id)
runs = await run_mgr.list_by_thread(thread_id)
latest_run_status = runs[-1].status.value if runs else None
finished = compute_embed_finished(
has_active_run=has_active_run,
meta_status=meta_status,
derived_status=derived_status,
)
return EmbedThreadStatusFields(
finished=finished,
thread_status=derived_status,
has_active_run=has_active_run,
latest_run_status=latest_run_status,
)
@router.post(
"/sessions",
response_model=RegisterEmbedSessionResponse,
summary="Register embed session_id → thread_id (called from iframe)",
)
async def register_embed_session(
body: RegisterEmbedSessionRequest,
request: Request,
) -> RegisterEmbedSessionResponse:
"""Upsert a mapping so the parent page can poll status by session_id only."""
store: EmbedSessionStore = get_embed_session_store(request)
record = await store.upsert(body.session_id, body.thread_id)
return RegisterEmbedSessionResponse(
session_id=record["session_id"],
thread_id=record["thread_id"],
)
@router.get(
"/sessions/{session_id}/status",
response_model=EmbedSessionStatusResponse,
summary="Poll embed Q&A completion by parent session_id",
)
async def get_embed_session_status(session_id: str, request: Request) -> EmbedSessionStatusResponse:
"""Primary status endpoint for iframe parents — no thread_id required."""
session_id = _validate_session_id(session_id)
store: EmbedSessionStore = get_embed_session_store(request)
record = await store.get(session_id)
if record is None:
return EmbedSessionStatusResponse(
session_id=session_id,
registered=False,
thread_id=None,
finished=False,
thread_status="pending",
has_active_run=False,
latest_run_status=None,
)
thread_id = record["thread_id"]
status = await _status_for_thread_id(thread_id, request)
return EmbedSessionStatusResponse(
session_id=session_id,
registered=True,
thread_id=thread_id,
**status.model_dump(),
)
@router.get(
"/threads/{thread_id}/status",
response_model=EmbedThreadStatusResponse,
summary="Poll embed Q&A completion by thread_id (optional / debugging)",
)
async def get_embed_thread_status(thread_id: str, request: Request) -> EmbedThreadStatusResponse:
"""Direct thread lookup when the parent already knows thread_id."""
status = await _status_for_thread_id(thread_id, request)
return EmbedThreadStatusResponse(thread_id=thread_id, **status.model_dump())
@router.get(
"/sessions/{session_id}/result",
response_model=EmbedSessionResultResponse,
summary="Fetch embed Q&A messages, generated text, and files by session_id",
)
async def get_embed_session_result(session_id: str, request: Request) -> EmbedSessionResultResponse:
"""Return model output for iframe parents once they know ``session_id`` only."""
session_id = _validate_session_id(session_id)
store: EmbedSessionStore = get_embed_session_store(request)
record = await store.get(session_id)
if record is None:
return EmbedSessionResultResponse(
session_id=session_id,
registered=False,
thread_id=None,
finished=False,
thread_status="pending",
has_active_run=False,
latest_run_status=None,
messages=[],
generated_text="",
generated_files=[],
)
thread_result = await _result_for_thread_id(record["thread_id"], request)
return EmbedSessionResultResponse(
session_id=session_id,
registered=True,
thread_id=thread_result.thread_id,
finished=thread_result.finished,
thread_status=thread_result.thread_status,
has_active_run=thread_result.has_active_run,
latest_run_status=thread_result.latest_run_status,
messages=thread_result.messages,
generated_text=thread_result.generated_text,
generated_files=thread_result.generated_files,
)
@router.get(
"/threads/{thread_id}/result",
response_model=EmbedThreadResultResponse,
summary="Fetch embed Q&A messages, generated text, and files by thread_id",
)
async def get_embed_thread_result(thread_id: str, request: Request) -> EmbedThreadResultResponse:
"""Direct thread lookup when the parent already knows thread_id."""
return await _result_for_thread_id(thread_id, request)
@@ -0,0 +1,162 @@
"""CRUD API for roundtable-planning business chains (业务链条).
A business chain is a named, reusable, *ordered* list of agents the user
assembles in the chain editor. In Step 2 the coordinator dispatches to the seats
following this order. Bound to the requesting user (not the browser), so it
follows the account across devices/sessions — same model as roundtable drafts.
Routes (prefix ``/api/roundtable-chains``):
GET "" list current user's chains (full records, incl. seats)
POST "" create a chain
GET "/{chain_id}" one chain
PUT "/{chain_id}" partial update (title / description / seats)
DELETE "/{chain_id}" delete
"""
from __future__ import annotations
import logging
from datetime import datetime
from uuid import uuid4
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/roundtable-chains", tags=["roundtable-chains"])
# Seat-count bounds mirror the roundtable selection bounds (2 ≤ N ≤ 8).
MIN_SEATS = 2
MAX_SEATS = 8
# ── schemas ────────────────────────────────────────────────────────────────
class ChainSeatSchema(BaseModel):
agent_id: str = Field(min_length=1, max_length=128)
name: str = ""
description: str = ""
class ChainResponse(BaseModel):
id: str
title: str = ""
description: str | None = None
seats: list[ChainSeatSchema] = []
created_at: datetime | str | None = None
updated_at: datetime | str | None = None
class ChainListResponse(BaseModel):
chains: list[ChainResponse]
class ChainCreateRequest(BaseModel):
# Optional client-provided id so the browser can keep one stable id across
# the create + subsequent updates without an extra round-trip.
id: str | None = Field(default=None, max_length=64)
title: str = Field(default="", max_length=512)
description: str | None = Field(default=None, max_length=2048)
seats: list[ChainSeatSchema] = []
class ChainUpdateRequest(BaseModel):
# All optional → callers send only what changed.
title: str | None = Field(default=None, max_length=512)
description: str | None = Field(default=None, max_length=2048)
seats: list[ChainSeatSchema] | None = None
# ── helpers ──────────────────────────────────────────────────────────────
def _current_user_id(request: Request) -> str:
user = getattr(request.state, "user", None)
if user is not None:
return str(user.id)
return get_effective_user_id()
def _get_store(request: Request):
store = getattr(request.app.state, "roundtable_chain_store", None)
if store is None:
raise HTTPException(status_code=503, detail="Roundtable chain store not available")
return store
def _validate_seats(seats: list[ChainSeatSchema]) -> None:
"""Enforce 2 ≤ N ≤ 8 and reject duplicate agent ids within a chain."""
if len(seats) < MIN_SEATS or len(seats) > MAX_SEATS:
raise HTTPException(
status_code=400,
detail=f"业务链条席位数须在 {MIN_SEATS}–{MAX_SEATS} 之间(当前 {len(seats)})",
)
seen: set[str] = set()
for s in seats:
if s.agent_id in seen:
raise HTTPException(status_code=400, detail=f"业务链条存在重复智能体: {s.agent_id}")
seen.add(s.agent_id)
# ── routes ───────────────────────────────────────────────────────────────
@router.get("", response_model=ChainListResponse)
async def list_chains(request: Request) -> ChainListResponse:
store = _get_store(request)
user_id = _current_user_id(request)
rows = await store.list_chains(user_id)
return ChainListResponse(chains=[ChainResponse(**r) for r in rows])
@router.post("", response_model=ChainResponse, status_code=201)
async def create_chain(request: Request, body: ChainCreateRequest) -> ChainResponse:
_validate_seats(body.seats)
store = _get_store(request)
user_id = _current_user_id(request)
row = await store.create_chain(
user_id,
{
"id": (body.id or uuid4().hex),
"title": body.title.strip(),
"description": body.description,
"seats": [s.model_dump() for s in body.seats],
},
)
return ChainResponse(**row)
@router.get("/{chain_id}", response_model=ChainResponse)
async def get_chain(request: Request, chain_id: str) -> ChainResponse:
store = _get_store(request)
user_id = _current_user_id(request)
row = await store.get_chain(chain_id, user_id)
if row is None:
raise HTTPException(status_code=404, detail="Chain not found")
return ChainResponse(**row)
@router.put("/{chain_id}", response_model=ChainResponse)
async def update_chain(request: Request, chain_id: str, body: ChainUpdateRequest) -> ChainResponse:
if body.seats is not None:
_validate_seats(body.seats)
store = _get_store(request)
user_id = _current_user_id(request)
# Distinguish "omitted" from "explicitly sent"; only forward provided fields.
fields = body.model_dump(exclude_unset=True)
row = await store.update_chain(chain_id, user_id, **fields)
if row is None:
raise HTTPException(status_code=404, detail="Chain not found")
return ChainResponse(**row)
@router.delete("/{chain_id}", status_code=204)
async def delete_chain(request: Request, chain_id: str) -> None:
store = _get_store(request)
user_id = _current_user_id(request)
deleted = await store.delete_chain(chain_id, user_id)
if not deleted:
raise HTTPException(status_code=404, detail="Chain not found")
@@ -52,6 +52,7 @@ class DraftListResponse(BaseModel):
class DraftResponse(DraftMetaResponse):
step1: dict[str, Any] | None = None
step2: dict[str, Any] | None = None
step3: dict[str, Any] | None = None
class DraftCreateRequest(BaseModel):
@@ -62,6 +63,7 @@ class DraftCreateRequest(BaseModel):
furthest_step: int = Field(default=1, ge=1, le=3)
step1: dict[str, Any] | None = None
step2: dict[str, Any] | None = None
step3: dict[str, Any] | None = None
class DraftUpdateRequest(BaseModel):
@@ -71,6 +73,7 @@ class DraftUpdateRequest(BaseModel):
furthest_step: int | None = Field(default=None, ge=1, le=3)
step1: dict[str, Any] | None = None
step2: dict[str, Any] | None = None
step3: dict[str, Any] | None = None
class RecommendPickSchema(BaseModel):
@@ -149,6 +152,7 @@ async def create_draft(request: Request, body: DraftCreateRequest) -> DraftRespo
"furthest_step": body.furthest_step,
"step1": body.step1,
"step2": body.step2,
"step3": body.step3,
},
)
return DraftResponse(**row)
@@ -219,6 +219,8 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
"max_concurrent_subagents",
"agent_name",
"is_bootstrap",
"is_writing_setup",
"valid_article_types",
"excluded_tools",
"skill_stop_names",
"skill_stop_message",
@@ -1,8 +1,9 @@
"""素材收集专家节点:关键词提取 → 并发搜索 → 相关度排序。"""
"""素材收集专家节点:关键词提取 → 并发搜索(含笔记) → 相关度排序。"""
from __future__ import annotations
import asyncio
import logging
import os
import re
import time
import uuid
@@ -21,6 +22,10 @@ from ._utils import get_ai_writing_config, llm_json, safe_get_stream_writer
logger = logging.getLogger(__name__)
_NOTEBOOK_SEARCH_PATH = "/api/v1/source/mentioned-documents/search"
_NOTEBOOK_MAX_KEYWORDS = 3 # 每次写作最多对笔记发起的检索轮数
_NOTEBOOK_TOP_K = 5 # 每次检索取回的 chunk 数
# 素材总量上限边界:config.yaml 的 ai_writing.max_materials,非法值回退默认 50
_DEFAULT_MAX_MATERIALS = 50
_MIN_MAX_MATERIALS = 1
@@ -95,6 +100,67 @@ def _notice(title: str, detail: str, raw: str = "") -> dict:
return event
async def _search_notebook(
keywords: list[str],
search_space_id: int,
source_ids: list[int],
token: str,
) -> list[Material]:
"""从笔记 API 按关键词检索内容,返回 source_type='notebook' 的 Material 列表。"""
import httpx # 懒导入:仅在笔记模式下才需要
base = (os.getenv("DEER_FLOW_NOTEBOOK_API_BASE") or "").rstrip("/")
if not base:
logger.warning("DEER_FLOW_NOTEBOOK_API_BASE not set, skipping notebook search")
return []
results: list[Material] = []
mat_idx = 0
try:
async with httpx.AsyncClient(timeout=20) as client:
for kw in keywords[:_NOTEBOOK_MAX_KEYWORDS]:
try:
resp = await client.post(
f"{base}{_NOTEBOOK_SEARCH_PATH}",
headers={"Authorization": f"Bearer {token}"},
json={
"query": kw,
"source_ids": source_ids,
"search_space_id": search_space_id,
"top_k": _NOTEBOOK_TOP_K,
},
)
resp.raise_for_status()
data = resp.json()
# 解析 documents 列表,每个 chunk 作为独立素材
documents: list[dict] = data.get("documents") if isinstance(data, dict) else []
for doc in documents or []:
doc_title: str = doc.get("title") or "笔记文档"
doc_url: str = doc.get("url") or ""
for chunk in doc.get("chunks") or []:
chunk_content: str = (chunk.get("content") or "").strip()
if not chunk_content:
continue
mat_idx += 1
raw_score = chunk.get("score")
chunk_score = float(raw_score) if raw_score is not None else max(0.5, 0.95 - (mat_idx - 1) * 0.03)
results.append(Material(
id=f"nb{mat_idx:03d}",
title=doc_title,
content=chunk_content[:500],
source=f"笔记·{doc_title}",
source_type="notebook",
relevance_score=chunk_score,
url=doc_url,
))
except Exception as e:
logger.warning("notebook search kw=%r failed: %s", kw, e)
except Exception as e:
logger.warning("notebook search client error: %s", e)
return results
async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict:
intent = state["user_intent"]
article_type = state.get("article_type", "analysis")
@@ -181,6 +247,22 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
if extra_keywords:
keywords = list(dict.fromkeys(keywords + extra_keywords))
# Step 1.5: 笔记检索(如果传入了 notebook_search_space_id)
notebook_search_space_id: int | None = state.get("notebook_search_space_id")
notebook_materials: list[Material] = []
if notebook_search_space_id:
notebook_source_ids: list[int] = state.get("notebook_source_ids") or []
notebook_token: str = cfg.get("notebook_token") or ""
_emit("正在检索笔记内容…")
notebook_materials = await _search_notebook(
keywords[:_NOTEBOOK_MAX_KEYWORDS],
notebook_search_space_id,
notebook_source_ids,
notebook_token,
)
if notebook_materials:
_emit(f"从笔记中获取到 {len(notebook_materials)} 条素材")
# Step 2: 搜索
# 技能检索:技能不是素材,是"检索策略 / 视角"。流程:
# a. SkillPicker 选出按相关度排序的 skills(推 list_skills + select_skills 事件)
@@ -330,10 +412,17 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
existing_pkg = state.get("material_package")
if is_skill_search:
# 技能编排循环已经完成素材累积,直接用 raw_materials_pre 当 raw_materials
# 笔记素材追加在技能素材之后
raw_materials: list[Material] = raw_materials_pre # type: ignore[name-defined]
raw_materials.extend(notebook_materials)
else:
seen_urls: set[str] = set()
raw_materials = []
# 笔记素材优先排在最前面,赋予高相关度
raw_materials = list(notebook_materials)
for m in notebook_materials:
if m.get("url"):
seen_urls.add(m["url"])
preserve_existing = existing_pkg and (is_supplement or is_append_only)
if preserve_existing:
for m in existing_pkg.get("materials", []):
@@ -7,11 +7,19 @@ import logging
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
from ..state import AIWritingState, KeyPoint, Outline, OutlineSection
from ..outline_parser import parse_user_outline
from ..prompts.writer_prompts import (
OUTLINE_SYSTEM, OUTLINE_USER, OUTLINE_REVISE_SYSTEM, OUTLINE_REVISE_USER,
OUTLINE_MATERIAL_ASSIGN_SYSTEM,
OUTLINE_MATERIAL_ASSIGN_USER,
OUTLINE_NORMALIZE_SYSTEM,
OUTLINE_NORMALIZE_USER,
OUTLINE_REVISE_SYSTEM,
OUTLINE_REVISE_USER,
OUTLINE_SYSTEM,
OUTLINE_USER,
)
from ._utils import get_writing_model, parse_json, _content_to_text, reasoning_delta, ThinkTagParser, safe_get_stream_writer
from ..state import AIWritingState, KeyPoint, Outline, OutlineSection
from ._utils import ThinkTagParser, _content_to_text, get_writing_model, llm_json, parse_json, reasoning_delta, safe_get_stream_writer
logger = logging.getLogger(__name__)
@@ -80,6 +88,108 @@ def _parse_outline(raw: dict) -> Outline:
return Outline(title=raw.get("title", "未命名文章"), sections=sections)
def _derive_title(state: AIWritingState) -> str:
"""用户大纲没给总标题时,用写作意图兜底一个标题。"""
intent = (state.get("user_intent") or "").strip()
if not intent:
return "未命名文章"
return intent[:30].rstrip() + ("…" if len(intent) > 30 else "")
async def _normalize_user_outline_via_llm(state: AIWritingState, config: RunnableConfig, raw: str) -> Outline:
"""确定性解析失败时,调 LLM 把用户大纲原文归一成三级结构(严格保真、不改写)。
再失败则把每个非空行当作一个章节,保证永远产出一份可用大纲、不丢用户内容。
"""
model_name = (state.get("model_name") or "").strip() or None
line_fallback = {
"title": "",
"sections": [
{"section_title": ln.strip(), "key_points": [], "material_ids": []}
for ln in raw.splitlines() if ln.strip()
] or [{"section_title": "正文", "key_points": [], "material_ids": []}],
}
result = await llm_json(
OUTLINE_NORMALIZE_SYSTEM,
OUTLINE_NORMALIZE_USER.format(raw_outline=raw),
line_fallback,
config,
model_name=model_name,
)
return _parse_outline(result)
async def _assign_materials_to_outline(
state: AIWritingState, config: RunnableConfig, outline: Outline, pkg: dict
) -> Outline:
"""为用户大纲的各章节回填 material_ids(结构锁定:只取编号、按章节序号嫁接)。
LLM 只负责「哪个章节配哪些素材」,标题与层级一律以解析结果为准,杜绝改写。失败则
保持 material_ids 为空(writer_draft 自带「取前 2 条素材」兜底)。
"""
sections = outline.get("sections", [])
materials = pkg.get("materials") or []
if not sections or not materials:
return outline
valid_ids = {m["id"] for m in materials}
sections_list = "\n".join(
f"{i + 1}. {s.get('section_title', '')}" for i, s in enumerate(sections)
)
model_name = (state.get("model_name") or "").strip() or None
result = await llm_json(
OUTLINE_MATERIAL_ASSIGN_SYSTEM,
OUTLINE_MATERIAL_ASSIGN_USER.format(
user_intent=state.get("user_intent", ""),
sections_list=sections_list,
materials_summary=_materials_summary(pkg),
),
{"assignments": []},
config,
model_name=model_name,
)
assignments = result.get("assignments")
if not isinstance(assignments, list):
return outline
for i, section in enumerate(sections):
if i >= len(assignments):
break
ids = assignments[i]
if isinstance(ids, list):
section["material_ids"] = [mid for mid in ids if mid in valid_ids]
return outline
async def _build_outline_from_user(
state: AIWritingState, config: RunnableConfig, pkg: dict, raw: str
) -> dict:
"""用户自带大纲分支:解析用户大纲 → 关联素材 → 直接进确认,不调 LLM 重新规划。"""
writer = safe_get_stream_writer()
writer({"type": "outline_chunk", "text": "正在整理你提供的大纲…\n", "is_thinking": True})
outline = parse_user_outline(raw)
if outline is None:
# 确定性解析拿不准(散文化/格式混乱)→ LLM 兜底归一
outline = await _normalize_user_outline_via_llm(state, config, raw)
if not outline.get("title"):
outline = {**outline, "title": _derive_title(state)}
# 仍走了检索 → 把已检索素材关联到用户章节(结构锁定)
outline = await _assign_materials_to_outline(state, config, outline, pkg)
section_count = len(outline.get("sections", []))
return {
"current_outline": outline,
"status": "awaiting_outline_confirm",
"progress_events": [
{"type": "step_start", "agent_name": "作家", "message": "采用你提供的大纲"},
{"type": "outline_ready", "agent_name": "作家", "message": f"已整理你的大纲,共 {section_count} 个章节"},
],
}
async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> dict:
pkg = state.get("material_package") or {"keywords": [], "materials": [], "summary": ""}
intervention = state.get("last_user_intervention") or {}
@@ -95,6 +205,13 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
current_outline = state.get("current_outline")
is_revise = bool(feedback) and bool(current_outline)
# 用户自带大纲:首轮(尚无大纲、非重规划)且表单粘了大纲原文 → 走「采用用户大纲」分支,
# 跳过 LLM 重新规划。后续若用户在确认卡里点「重新规划」,current_outline 已存在 → 落到
# 下方 is_revise 编辑模式,按反馈调整用户大纲(符合预期)。
user_outline_raw = (state.get("user_outline_raw") or "").strip()
if user_outline_raw and not current_outline and not is_revise:
return await _build_outline_from_user(state, config, pkg, user_outline_raw)
if is_revise:
system_prompt = OUTLINE_REVISE_SYSTEM
user_prompt = OUTLINE_REVISE_USER.format(
@@ -0,0 +1,220 @@
"""把用户粘贴的大纲文本解析成三级 ``Outline`` 结构(确定性解析,不依赖 LLM)。
用户在写作表单里可能直接粘贴一份已有大纲,要求「完全按我的大纲生成」。本模块负责
把各种常见的大纲文本格式严格、保真地转成后端三级结构:
Outline.sections[] = Level 1 章节(section_title)
OutlineSection.key_points[] = Level 2 小节(point_title)
KeyPoint.sub_points[] = Level 3 子要点
兼容的输入格式:
- Markdown 标题:``#`` / ``##`` / ``###`` / ``####``(首个孤立 ``#`` 视为文章标题)
- 中文编号:``一、`` / ``(一)`` / ``第一章`` / ``第一节``
- 阿拉伯编号:``1.`` / ``1、`` / ``1.1`` / ``1.1.1``
- 缩进 bullet:``-`` / ``*`` / ``•`` / ``·``(用缩进深度表示层级)
- 纯文本平铺:每行一个一级标题(短标题、无句末标点)
设计原则:
- **保真不改写**:只识别层级、原样保留标题文本,绝不增删或改写用户内容。
- **多级混排自适应**:把文本里实际出现的各种「层级标记」收集起来排序,按相对外层
顺序映射到 1/2/3 级——所以不依赖固定的标记搭配,``##``+``-`` 或 ``一、``+缩进
都能正确分层。
- **拿不准就交给上层**:识别不出可靠层级(如整段散文)时返回 ``None``,由调用方走
LLM 兜底归一,避免硬把散文拆成错误结构。
"""
from __future__ import annotations
import re
from .state import KeyPoint, Outline, OutlineSection
# ── 行首标记识别 ──────────────────────────────────────────────────────────────
_CN_NUM = "一二三四五六七八九十百千零〇两"
# Markdown 标题:## 标题
_MD_HEADING_RE = re.compile(r"^(#{1,6})\s+(.*\S)\s*$")
# 第一章 / 第1节 / 第一部分
_CN_CHAPTER_RE = re.compile(rf"^第\s*[{_CN_NUM}\d]+\s*[章节部分篇讲回卷](?:[\s::、..))]\s*|\s+)?(.*)$")
# 一、 / 十二、 / 一. / 一.
_CN_DUN_RE = re.compile(rf"^[{_CN_NUM}]+\s*[、..]\s*(.*)$")
# (一) / (一) / (1) / (1)
_PAREN_RE = re.compile(rf"^[((]\s*[{_CN_NUM}\d]+\s*[))]\s*(.*)$")
# 1.1 / 1.1.1(带至少一个点的多级阿拉伯编号)
_ARABIC_DOTTED_RE = re.compile(r"^(\d+(?:\.\d+)+)\s*[、..))]?\s*(.*)$")
# 1. / 1、 / 1) / 1.(单级阿拉伯编号)
_ARABIC_SINGLE_RE = re.compile(r"^\d+\s*[、..))]\s*(.*)$")
# 带圈数字 ①②③ / 罗马 ⅰ
_CIRCLED_RE = re.compile(r"^[①②③④⑤⑥⑦⑧⑨⑩⑪⑫⑬⑭⑮]\s*(.*)$")
# bullet:- / * / • / ·(要求标记后有空白,避免误吞「*强调*」之类)
_BULLET_RE = re.compile(r"^[-*•·]\s+(.*\S)\s*$")
# 标记分组的「外层度」基准值(越小越外层)。group: 0=标题 1=编号 2=缩进。
# 同组内 order 决定相对层级;不同组之间靠 group 排序(标题 < 编号 < 缩进)。
# 编号顺序遵循中文公文惯例:第一章 > 一、 > (一) > 1. > 1.1 > ①。
_NUM_ORDER = {
"cn_chapter": 0.0, # 第一章
"cn_dun": 1.0, # 一、
"paren": 2.0, # (一)
"arabic_single": 3.0, # 1.
"circled": 8.0, # ①
}
def _indent_width(line: str) -> int:
"""计算行首缩进宽度(tab 记 4 列)。"""
width = 0
for ch in line:
if ch == " ":
width += 1
elif ch == "\t":
width += 4
else:
break
return width
def _classify(stripped: str, indent: int) -> tuple[tuple[int, float], str] | None:
"""识别一行的层级标记,返回 ``((group, order), 纯文本)``;无法识别返回 None。
``(group, order)`` 是该行的「层级键」:把全文出现过的不同键排序后,按位置映射到
1/2/3 级。这样不必预设标记搭配,相对顺序对了层级就对。
"""
if not stripped:
return None
# 1) Markdown 标题(group 0,order = 井号数)
m = _MD_HEADING_RE.match(stripped)
if m:
depth = len(m.group(1))
text = m.group(2).strip()
return (0, float(depth)), text
# 2) 各类编号(group 1)
m = _CN_CHAPTER_RE.match(stripped)
if m and stripped.startswith("第"):
return (1, _NUM_ORDER["cn_chapter"]), (m.group(1) or "").strip()
m = _CN_DUN_RE.match(stripped)
if m:
return (1, _NUM_ORDER["cn_dun"]), m.group(1).strip()
m = _ARABIC_DOTTED_RE.match(stripped)
if m:
depth = m.group(1).count(".") + 1 # 1.1 → 2, 1.1.1 → 3
# 多级阿拉伯:order 随层级递增,落在单级(3.0)与圈号(8.0)之间
return (1, 3.0 + depth), m.group(2).strip()
m = _PAREN_RE.match(stripped)
if m:
return (1, _NUM_ORDER["paren"]), m.group(1).strip()
m = _ARABIC_SINGLE_RE.match(stripped)
if m:
return (1, _NUM_ORDER["arabic_single"]), m.group(1).strip()
m = _CIRCLED_RE.match(stripped)
if m:
return (1, _NUM_ORDER["circled"]), m.group(1).strip()
# 3) bullet:用缩进深度区分层级(group 2)
m = _BULLET_RE.match(stripped)
if m:
return (2, float(indent)), m.group(1).strip()
# 4) 纯文本:用缩进深度区分层级(group 3,最内层基准)
return (3, float(indent)), stripped
def _looks_like_prose(text: str) -> bool:
"""判断一行是否更像散文句子而非大纲标题(长、或带句末标点)。"""
if len(text) > 50:
return True
return text.rstrip()[-1:] in "。!?!?;;"
def parse_user_outline(raw: str) -> Outline | None:
"""把用户粘贴的大纲文本解析成三级 ``Outline``;无法可靠解析时返回 ``None``。
返回 ``None`` 的情形:空白输入、只有一行且无任何标记、或整体更像散文段落——这些交给
上层用 LLM 兜底归一,避免误拆。
"""
if not raw or not raw.strip():
return None
rows: list[tuple[tuple[int, float], str]] = []
for line in raw.splitlines():
if not line.strip():
continue
indent = _indent_width(line)
classified = _classify(line.strip(), indent)
if classified is None:
continue
key, text = classified
if text:
rows.append((key, text))
if not rows:
return None
marked = any(key[0] != 3 for key, _ in rows) # 是否出现过标题/编号/bullet 标记
# 纯文本且无任何标记:可能是「平铺标题列表」也可能是散文。
# 单行、或任一行像句子 → 判为散文,返回 None 交给 LLM。
if not marked:
if len(rows) < 2:
return None
if any(_looks_like_prose(text) for _, text in rows):
return None
# 文章标题:仅在「Markdown 首行是孤立的最浅井号、且存在更深井号」时认定。
title = ""
body = rows
heading_depths = [key[1] for key, _ in rows if key[0] == 0]
if heading_depths:
min_depth = min(heading_depths)
first_key = rows[0][0]
is_first_top_heading = first_key == (0, min_depth)
only_one_at_top = sum(1 for d in heading_depths if d == min_depth) == 1
has_deeper = any(d > min_depth for d in heading_depths)
if is_first_top_heading and only_one_at_top and has_deeper:
title = rows[0][1]
body = rows[1:]
if not body:
return None
# 把 body 里出现的不同「层级键」排序,按位置映射到 1/2/3 级(超过 3 级压到 3)。
distinct_keys = sorted({key for key, _ in body})
level_of: dict[tuple[int, float], int] = {
key: min(idx + 1, 3) for idx, key in enumerate(distinct_keys)
}
sections: list[OutlineSection] = []
cur_section: OutlineSection | None = None
cur_point: KeyPoint | None = None
for key, text in body:
level = level_of[key]
if level <= 1:
cur_section = OutlineSection(section_title=text, key_points=[], material_ids=[])
sections.append(cur_section)
cur_point = None
elif level == 2:
if cur_section is None: # 孤立 L2:提升为章节,避免空标题
cur_section = OutlineSection(section_title=text, key_points=[], material_ids=[])
sections.append(cur_section)
cur_point = None
continue
cur_point = KeyPoint(point_title=text, sub_points=[])
cur_section["key_points"].append(cur_point)
else: # level == 3
if cur_section is None: # 孤立 L3:提升为章节
cur_section = OutlineSection(section_title=text, key_points=[], material_ids=[])
sections.append(cur_section)
cur_point = None
elif cur_point is None: # 章节下尚无小节:提升为小节
cur_point = KeyPoint(point_title=text, sub_points=[])
cur_section["key_points"].append(cur_point)
else:
cur_point["sub_points"].append(text)
if not sections:
return None
return Outline(title=title, sections=sections)
@@ -123,6 +123,60 @@ OUTLINE_REVISE_USER = """写作意图:{user_intent}
请在「当前大纲」基础上落实上述「用户修改要求」,未提到的章节保持原样,输出调整后的完整大纲 JSON。"""
# ── 用户自带大纲:LLM 兜底归一(确定性解析失败时才用)──────────────────────────
OUTLINE_NORMALIZE_SYSTEM = """你是一名大纲格式整理助手。
用户粘贴了一份已有的文章大纲(格式可能不规范),你的任务是把它原样转成结构化 JSON。
【最高原则:保真,绝不改写】
- 严格保留用户写的每一条标题文字,逐字照搬,不要润色、改写、翻译、增加或删除任何条目。
- 你只判断层级(哪条是章节、哪条是小节、哪条是子要点),不创作内容。
- 层级判断依据用户原文里的编号(一、/(一)/1./1.1)、Markdown 井号数、或缩进深度。
【三级结构】
- Level 1 章节 → section_title
- Level 2 小节 → key_points[].point_title
- Level 3 子要点 → key_points[].sub_points[](字符串数组)
- 用户大纲只有一级时,每条都是章节,key_points 留空 []。
- 用户大纲只有两级时,sub_points 留空 []。
【输出要求】
- 只输出一个 JSON 对象,不要解释文字,不要 markdown 代码块,必须用英文双引号。
- material_ids 一律输出 [](此步不关联素材)。
- 结构:{"title": "...", "sections": [{"section_title": "...", "key_points": [{"point_title": "...", "sub_points": ["..."]}], "material_ids": []}]}
- 用户大纲若包含一个总标题,放进 title;没有就留空字符串。"""
OUTLINE_NORMALIZE_USER = """用户粘贴的大纲原文:
{raw_outline}
请逐条保留原文,仅判断层级,输出结构化大纲 JSON。"""
# ── 用户自带大纲:素材关联(结构锁定,只挑 material_ids)──────────────────────────
OUTLINE_MATERIAL_ASSIGN_SYSTEM = """你是一名资料编辑。
给定一份【已确定的文章大纲】(章节顺序固定)和一批检索到的素材,你的任务是为每个章节
挑选最相关的素材编号。
【规则】
- 不要修改大纲的任何标题或结构,你只输出每个章节对应的素材编号列表。
- 严格按大纲给出的章节顺序输出,一个章节对应一个编号数组;该章节没有相关素材就给 []。
- material_id 必须来自给定素材清单里出现过的编号,不要编造。
- 一条素材可以分配给多个章节,也可以一个都不分配。
【输出要求】
- 只输出一个 JSON 对象,不要解释文字,不要 markdown 代码块,必须用英文双引号。
- 结构:{"assignments": [["m001","m002"], [], ["m003"]]},数组长度等于章节数、顺序一致。"""
OUTLINE_MATERIAL_ASSIGN_USER = """写作意图:{user_intent}
文章大纲(章节按顺序编号):
{sections_list}
可用素材:
{materials_summary}
请按章节顺序,为每个章节挑选相关的 material_id,输出 assignments 数组。"""
DRAFT_SECTION_SYSTEM = """你是一名专业内容创作者(作家)。
你的目标:根据大纲和素材,写出一段结构清晰、高质量的文章内容。
你的工作背景:你有十年内容创作经验,文字流畅,善于用事实说话。
@@ -125,6 +125,9 @@ class BlockedSection(TypedDict):
class AIWritingState(TypedDict):
# 写作配置
user_intent: str
# 用户在写作表单粘贴的「已有大纲」原文(可选)。非空时 writer_outline 跳过 LLM 规划,
# 直接解析这份大纲并按用户结构生成,做到「完全按我的大纲写」。留空走正常自动规划。
user_outline_raw: str
article_type: str
target_audience: str
word_count_target: int
@@ -140,6 +143,9 @@ class AIWritingState(TypedDict):
# 经 checkpoint 持久化 —— interrupt/resume 是新 invocation、其 config 不带原
# model_name,存进 state 才能让 resume 后才运行的节点拿到用户模型。
model_name: str
# 笔记来源(可选):传入后 researcher 阶段会额外从笔记检索素材
notebook_search_space_id: Optional[int]
notebook_source_ids: Optional[List[int]]
# Agent A 输出
material_package: Optional[MaterialPackage]
@@ -11,6 +11,7 @@ from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
from deerflow.agents.middlewares.rag_citation_middleware import RagCitationMiddleware
from deerflow.agents.middlewares.setup_agent_approval_middleware import SetupAgentApprovalMiddleware
from deerflow.agents.middlewares.setup_writing_approval_middleware import SetupWritingApprovalMiddleware
from deerflow.agents.middlewares.subagent_limit_middleware import SubagentLimitMiddleware
from deerflow.agents.middlewares.summarization_middleware import BeforeSummarizationHook, DeerFlowSummarizationMiddleware
from deerflow.agents.middlewares.title_middleware import TitleMiddleware
@@ -25,6 +26,9 @@ from deerflow.models import create_chat_model
logger = logging.getLogger(__name__)
# 写作配置 Q&A:意图澄清最多几轮(一条用户消息 = 一轮),到顶强制提交配置卡。
WRITING_SETUP_MAX_ROUNDS = 3
def _get_runtime_config(config: RunnableConfig) -> dict:
"""Merge legacy configurable options with LangGraph runtime context."""
@@ -272,6 +276,7 @@ def _build_middlewares(
# Add TodoList middleware if plan mode is enabled
is_plan_mode = cfg.get("is_plan_mode", False)
is_bootstrap = cfg.get("is_bootstrap", False)
is_writing_setup = cfg.get("is_writing_setup", False)
todo_list_middleware = _create_todo_list_middleware(is_plan_mode)
if todo_list_middleware is not None:
middlewares.append(todo_list_middleware)
@@ -291,7 +296,7 @@ def _build_middlewares(
# Add MemoryMiddleware (after TitleMiddleware) — skip in bootstrap mode (agent creation flow)
memory_injection_enabled = cfg.get("memory_injection_enabled", True)
if not is_bootstrap:
if not is_bootstrap and not is_writing_setup:
middlewares.append(MemoryMiddleware(agent_name=agent_name, memory_config=resolved_app_config.memory, injection_enabled=memory_injection_enabled))
# Add ViewImageMiddleware only for the NATIVE-vision path (main model
@@ -324,6 +329,7 @@ def _build_middlewares(
middlewares.extend(custom_middlewares)
middlewares.append(SetupAgentApprovalMiddleware())
middlewares.append(SetupWritingApprovalMiddleware())
# SkillStopMiddleware — intercepts skill invocations and halts execution
# when a configured skill is about to run. Request-level options
@@ -371,7 +377,7 @@ def make_lead_agent(config: RunnableConfig):
def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
# Lazy import to avoid circular dependency
from deerflow.tools import get_available_tools
from deerflow.tools.builtins import setup_agent
from deerflow.tools.builtins import ask_clarification_tool, setup_agent, setup_writing
cfg = _get_runtime_config(config)
resolved_app_config = app_config
@@ -383,6 +389,10 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
subagent_enabled = cfg.get("subagent_enabled", False)
max_concurrent_subagents = cfg.get("max_concurrent_subagents", 3)
is_bootstrap = cfg.get("is_bootstrap", False)
# 写作配置 Q&A:与 is_bootstrap 同构的另一种「最小化 bootstrap」——挂 setup_writing 工具、
# 走写作配置专用 prompt,只负责帮用户把写作表单聊清并提交,不真正开始写作。
is_writing_setup = cfg.get("is_writing_setup", False)
writing_setup_article_types = cfg.get("valid_article_types") if is_writing_setup else None
memory_injection_enabled = bool(cfg.get("memory_injection_enabled", True))
rag_mode_enabled = bool(cfg.get("rag_mode_enabled", False))
writing_mode = bool(cfg.get("writing_mode", False))
@@ -393,7 +403,7 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
requested_excluded_tools = cfg.get("excluded_tools") or None
try:
agent_config = load_agent_config(agent_id) if not is_bootstrap else None
agent_config = load_agent_config(agent_id) if not is_bootstrap and not is_writing_setup else None
except FileNotFoundError:
logger.warning("Agent directory not found for '%s'; running with default config", agent_id)
agent_config = None
@@ -453,6 +463,25 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
}
)
if is_writing_setup:
# 写作配置 Q&A:最小化 agent —— 只挂 setup_writing(提交配置)+ ask_clarification(弹澄清卡)。
# 关键:绝不给全套工具、绝不复用 lead 超级体提示,否则模型会把「新能源汽车市场分析」之类
# 长得像任务的输入直接做掉,而不是转成写作配置提交。
# 末尾追加 WritingSetupRoundCapMiddleware:澄清最多 3 轮,到顶用 tool_choice 强制提交配置卡。
from deerflow.agents.lead_agent.prompt import build_writing_setup_prompt
from deerflow.agents.middlewares.writing_setup_round_cap_middleware import WritingSetupRoundCapMiddleware
return create_agent(
model=create_chat_model(name=model_name, thinking_enabled=thinking_enabled, app_config=resolved_app_config),
tools=[setup_writing, ask_clarification_tool],
middleware=[
*_build_middlewares(config, model_name=model_name, app_config=resolved_app_config),
WritingSetupRoundCapMiddleware(max_rounds=WRITING_SETUP_MAX_ROUNDS),
],
system_prompt=build_writing_setup_prompt(writing_setup_article_types),
state_schema=ThreadState,
)
if is_bootstrap:
# Special bootstrap agent with minimal prompt for initial custom agent creation flow
return create_agent(
@@ -703,14 +703,14 @@ def _get_cached_skills_prompt_section(
display_pattern = (
"""
**Skill Result Display Pattern:**
- If RAG/reference display mode is enabled, do not number each skill result independently. Treat all retrieval results from the same user turn as one combined `referenceBatch` and cite with `[1]`, `[2]`, etc. from the system-provided reference prompt.
- In RAG/reference display mode, do not output `<skill-display .../>`; the frontend renders cards, tables, lists, and citation details in the right reference panel.
- If RAG/reference display mode is not enabled and a skill has `<display mode="citation">`, use the skill's returned result order as reference numbers. Cite facts from the first result as `[1]`, the second as `[2]`, and so on. Use only numbers that correspond to returned results.
- If a skill has `<display mode="list">`, summarize the returned items as a concise list and keep item order stable.
- If a skill has `<display mode="table">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="table" />` where the frontend should render the real table. Do not hand-write the table rows yourself.
- If a skill has `<display mode="card">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="card" />` where the frontend should render the cards.
- If a skill has `<display mode="custom">`, follow the custom display instruction in the `<display>` tag when presenting the skill result.
**Skill Result Display Pattern:**
- If RAG/reference display mode is enabled, do not number each skill result independently. Treat all retrieval results from the same user turn as one combined `referenceBatch` and cite with `[1]`, `[2]`, etc. from the system-provided reference prompt.
- In RAG/reference display mode, do not output `<skill-display .../>`; the frontend renders cards, tables, lists, and citation details in the right reference panel.
- If RAG/reference display mode is not enabled and a skill has `<display mode="citation">`, use the skill's returned result order as reference numbers. Cite facts from the first result as `[1]`, the second as `[2]`, and so on. Use only numbers that correspond to returned results.
- If a skill has `<display mode="list">`, summarize the returned items as a concise list and keep item order stable.
- If a skill has `<display mode="table">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="table" />` where the frontend should render the real table. Do not hand-write the table rows yourself.
- If a skill has `<display mode="card">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="card" />` where the frontend should render the cards.
- If a skill has `<display mode="custom">`, follow the custom display instruction in the `<display>` tag when presenting the skill result.
"""
if display_signature
else ""
@@ -904,6 +904,46 @@ def _build_bootstrap_section() -> str:
"""
def build_writing_setup_prompt(article_types: list[dict] | None = None) -> str:
"""写作配置 Q&A 的**独立**系统提示(不复用 lead-agent 超级体提示)。
关键:写作配置助手绝不能像 lead agent 那样「自己把任务做掉」——用户输入常常长得像一个
可执行任务(如「新能源汽车市场分析」),若复用 lead 提示 + 全套工具,模型会直接去研究/写
分析。这里用一份职责单一、口吻强硬的独立提示,配合「只挂 setup_writing 一个工具」,把模型
死死框在「聊清需求 → 调 setup_writing」上。
"""
if article_types:
type_lines = "\n".join(
f" - `{t.get('key', '')}`:{t.get('label', '')}" for t in article_types if t.get("key")
)
type_block = f"【可选文章类型】article_type 只能填下列 key 之一:\n{type_lines}"
else:
type_block = "【文章类型】若不确定 article_type,可留空,交给用户在表单里选。"
return f"""你是「AI 写作」功能的**配置助手**。你的唯一职责:通过简短对话把用户的写作需求聊清楚,然后调用 `setup_writing` 工具把写作配置提交给用户确认。提交后由用户在表单里应用、再点「开始写作」。
【铁律——最高优先级】
- 你**绝对不撰写**文章正文、分析内容、大纲内容或任何成稿文字。即使用户的输入看起来就是一个写作任务(例如「新能源汽车市场分析」「帮我写一篇……」),你也**只把它当作"要配置的写作目标"**,绝不替用户动笔、绝不直接完成这篇写作。
- 你只有两个工具:① `ask_clarification`(向用户提问澄清,需要时用);② `setup_writing`(提交写作配置)。**没有任何其它工具**,不要做检索、不要做研究、不要假装调用别的工具。
- 不要在回复里输出文章段落、要点清单或大纲(用户主动粘贴的大纲除外——原样放进 `outline` 参数)。
【工作流程】(与用户交互确认,但最多 3 轮内必须提交)
1. 读用户输入。当关键项(文章类型、目标字数、目标读者)缺失或模糊时,调用 `ask_clarification` 向用户提问——**优先给出 `options` 选项**让用户点选(例如目标字数给「800 / 1500 / 3000」,读者给「普通读者 / 行业专家 / 政策决策者」)。一次可就 1-2 个独立的关键项各发一个 ask_clarification。
2. 与用户的澄清来回**最多 2 次**。一旦关键信息明确、或已经问过 2 次,就**立即进入第 3 步提交**,缺失字段用合理默认值,绝不无休止追问。
3. 调用 `setup_writing` 提交配置:
- `user_intent`:精炼成一句话的写作目标(必填)。
- `article_type`:见下方可选类型;不确定就留空。
- `word_count_target`:正整数;用户没明说就给合理默认(如 800)。
- `target_audience`:如「普通读者」「行业专家」;没提就填「普通读者」。
- `writing_mode`:用户强调「严格依据素材/不要编造」→ `strict`,否则 `loose`。
- `material_source`:**默认 `knowledge_base`(知识库)**;只有当用户明确要求用「技能/技能检索」时才填 `skill`,否则一律 `knowledge_base`。
- `outline`:用户提供的大纲原文(保留其层级与标题),否则留空。
4. 调用 `setup_writing` 前,用**一句话**说明你将提交的配置要点;调用之后不要再输出任何文章内容。
{type_block}
"""
def apply_prompt_template(
subagent_enabled: bool = False,
max_concurrent_subagents: int = 3,
@@ -995,16 +1035,16 @@ Writing mode is enabled for this run.
network_policy_section = _build_network_policy_section()
rag_mode_section = (
"""
<rag_citation_mode>
RAG/reference display mode is enabled.
When a search, knowledge, skill, or MCP tool returns a result list:
- If multiple search terms are needed, prefer calling the retrieval tools for all known terms in the same step, then answer after the tool results are returned.
- Treat every retrieval result from the same user turn as one combined `referenceBatch`; do not restart numbering for each search term, tool call, skill, or display adapter.
- Use only the system-provided merged reference numbers in the final answer, formatted as `[1]`, `[2]`, `[1][3][5]`.
- Do not output `<skill-display .../>` in RAG/reference mode. The frontend renders configured cards, tables, lists, and citation details in the right reference panel.
- If no display adapter exists, the frontend defaults the right reference panel to numbered citations.
</rag_citation_mode>
"""
<rag_citation_mode>
RAG/reference display mode is enabled.
When a search, knowledge, skill, or MCP tool returns a result list:
- If multiple search terms are needed, prefer calling the retrieval tools for all known terms in the same step, then answer after the tool results are returned.
- Treat every retrieval result from the same user turn as one combined `referenceBatch`; do not restart numbering for each search term, tool call, skill, or display adapter.
- Use only the system-provided merged reference numbers in the final answer, formatted as `[1]`, `[2]`, `[1][3][5]`.
- Do not output `<skill-display .../>` in RAG/reference mode. The frontend renders configured cards, tables, lists, and citation details in the right reference panel.
- If no display adapter exists, the frontend defaults the right reference panel to numbered citations.
</rag_citation_mode>
"""
if rag_mode_enabled
else ""
)
@@ -0,0 +1,133 @@
"""Approval middleware for the AI-writing config setup tool.
与 ``SetupAgentApprovalMiddleware`` 同构:模型在「写作配置 Q&A」里调用 ``setup_writing``
时,于 ``after_model`` 拦截,附上一条结构化审批 ToolMessage(建议的写作配置打包在
``additional_kwargs["setup_writing_approval"]``)并 ``jump_to="end"`` 立即结束本轮运行——
前端据此渲染审批卡,用户点「应用到表单」后把这些字段回填到左侧写作表单。
"""
from __future__ import annotations
import logging
from hashlib import sha256
from typing import Any, override
from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware, hook_config
from langchain_core.messages import AIMessage, ToolMessage
from langgraph.runtime import Runtime
logger = logging.getLogger(__name__)
class SetupWritingApprovalMiddlewareState(AgentState):
"""Compatible with the DeerFlow thread state schema."""
pass
class SetupWritingApprovalMiddleware(AgentMiddleware[SetupWritingApprovalMiddlewareState]):
"""Stop before ``setup_writing`` and ask the user to apply the proposed config.
See ``SetupAgentApprovalMiddleware`` for the rationale behind intercepting in
``after_model`` + ``jump_to="end"`` (robust against parallel tool calls).
"""
state_schema = SetupWritingApprovalMiddlewareState
def _is_approved(self, runtime: Runtime) -> bool:
context = getattr(runtime, "context", None) or {}
if not isinstance(context, dict):
return False
return context.get("setup_writing_approved") is True
def _stable_message_id(self, tool_call_id: str, args: dict[str, Any]) -> str:
if tool_call_id:
return f"setup-writing-approval:{tool_call_id}"
digest = sha256(repr(sorted(args.items())).encode("utf-8")).hexdigest()[:16]
return f"setup-writing-approval:{digest}"
def _coerce_word_count(self, raw: Any) -> int | None:
if isinstance(raw, bool): # bool 是 int 子类,先排除
return None
if isinstance(raw, int):
return raw if raw > 0 else None
if isinstance(raw, float):
return int(raw) if raw > 0 else None
if isinstance(raw, str):
import re
m = re.search(r"\d[\d,]*", raw)
if m:
try:
value = int(m.group().replace(",", ""))
return value if value > 0 else None
except ValueError:
return None
return None
def _build_payload(self, args: dict[str, Any]) -> dict[str, Any]:
# 素材来源:仅当模型明确给出 "skill" 时才用技能检索;用户没明确要求就一律默认知识库。
material_source = (
"skill" if str(args.get("material_source") or "").strip() == "skill" else "knowledge_base"
)
return {
"user_intent": str(args.get("user_intent") or "").strip(),
"article_type": str(args.get("article_type") or "").strip(),
"word_count_target": self._coerce_word_count(args.get("word_count_target")),
"target_audience": str(args.get("target_audience") or "").strip(),
"writing_mode": str(args.get("writing_mode") or "").strip(),
"material_source": material_source,
"outline": str(args.get("outline") or "").strip(),
}
def _intercept(self, last: AIMessage) -> dict[str, Any] | None:
"""Return a state update if setup_writing must be gated; None to passthrough."""
tool_calls = list(last.tool_calls or [])
setup_call = next((tc for tc in tool_calls if tc.get("name") == "setup_writing"), None)
if setup_call is None:
return None
args = setup_call.get("args") or {}
tool_call_id = setup_call.get("id") or ""
payload = self._build_payload(args)
logger.info("Setup writing approval requested before applying writing config")
approval_msg = ToolMessage(
id=self._stable_message_id(tool_call_id, args),
content="请确认是否把这份写作配置应用到表单。",
tool_call_id=tool_call_id,
name="setup_writing",
additional_kwargs={"setup_writing_approval": payload},
)
return {
"messages": [approval_msg],
"jump_to": "end",
}
@hook_config(can_jump_to=["end"])
@override
def after_model(
self,
state: SetupWritingApprovalMiddlewareState,
runtime: Runtime,
) -> dict[str, Any] | None:
if self._is_approved(runtime):
return None
messages = state.get("messages") or []
if not messages:
return None
last = messages[-1]
if not isinstance(last, AIMessage):
return None
return self._intercept(last)
@hook_config(can_jump_to=["end"])
@override
async def aafter_model(
self,
state: SetupWritingApprovalMiddlewareState,
runtime: Runtime,
) -> dict[str, Any] | None:
return self.after_model(state, runtime)
@@ -0,0 +1,72 @@
"""写作配置 Q&A 的「轮数上限」中间件。
意图阶段允许与用户来回澄清确认,但**最多 N 轮后必须出结果**:到上限时,在模型请求里
注入强制指令 + 用 ``tool_choice`` 锁定 ``setup_writing``,逼模型立刻提交配置卡,不再提问。
「一轮」= 一条用户消息(HumanMessage)。达到 ``max_rounds`` 条用户消息时强制提交。
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import override
from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware
from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse
from langchain_core.messages import HumanMessage, SystemMessage
class WritingSetupRoundCapMiddleware(AgentMiddleware[AgentState]):
"""达到澄清轮数上限时,强制模型调用 ``setup_writing`` 提交。
软约束(注入 SystemMessage)+ 硬约束(``tool_choice`` 锁定工具)双保险:即使个别模型
无视提示,``tool_choice`` 也会逼它产出一次 ``setup_writing`` 调用。
"""
def __init__(self, max_rounds: int = 3, tool_name: str = "setup_writing") -> None:
super().__init__()
self._max_rounds = max(1, max_rounds)
self._tool_name = tool_name
def _should_force(self, messages: list) -> bool:
human_count = sum(1 for m in messages if isinstance(m, HumanMessage))
return human_count >= self._max_rounds
def _reminder(self) -> SystemMessage:
return SystemMessage(
content=(
f"【系统强制】意图澄清已达 {self._max_rounds} 轮上限。"
"现在必须立即调用 setup_writing 提交写作配置,禁止再向用户提任何问题。"
"缺失或不确定的字段一律用合理默认值:"
"word_count_target 默认 800、target_audience 默认「普通读者」、"
"writing_mode 默认 loose、material_source 默认 knowledge_base、"
"article_type 不确定就留空。"
)
)
def _patch(self, request: ModelRequest) -> ModelRequest:
return request.override(
messages=[*request.messages, self._reminder()],
tool_choice=self._tool_name,
)
@override
def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelCallResult:
if self._should_force(request.messages):
request = self._patch(request)
return handler(request)
@override
async def awrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
) -> ModelCallResult:
if self._should_force(request.messages):
request = self._patch(request)
return await handler(request)
@@ -0,0 +1,18 @@
"""Persistence for iframe embed session → thread mappings."""
from deerflow.persistence.embed_sessions.base import EmbedSessionStore
from deerflow.persistence.embed_sessions.memory import MemoryEmbedSessionStore
from deerflow.persistence.embed_sessions.sql import EmbedSessionRepository
__all__ = [
"EmbedSessionStore",
"MemoryEmbedSessionStore",
"EmbedSessionRepository",
"make_embed_session_store",
]
def make_embed_session_store(session_factory):
if session_factory is None:
return MemoryEmbedSessionStore()
return EmbedSessionRepository(session_factory)
@@ -0,0 +1,18 @@
"""Abstract persistence API for iframe embed session → thread mappings."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class EmbedSessionStore(ABC):
@abstractmethod
async def upsert(self, session_id: str, thread_id: str) -> dict[str, Any]:
"""Create or update the mapping for an embed session."""
raise NotImplementedError
@abstractmethod
async def get(self, session_id: str) -> dict[str, Any] | None:
"""Return ``{session_id, thread_id, created_at, updated_at}`` or None."""
raise NotImplementedError
@@ -0,0 +1,37 @@
"""In-memory ``EmbedSessionStore`` for tests / memory backend."""
from __future__ import annotations
import copy
from datetime import UTC, datetime
from typing import Any
from deerflow.persistence.embed_sessions.base import EmbedSessionStore
class MemoryEmbedSessionStore(EmbedSessionStore):
def __init__(self) -> None:
self._sessions: dict[str, dict[str, Any]] = {}
async def upsert(self, session_id: str, thread_id: str) -> dict[str, Any]:
now = datetime.now(UTC).isoformat()
existing = self._sessions.get(session_id)
if existing is None:
record = {
"session_id": session_id,
"thread_id": thread_id,
"created_at": now,
"updated_at": now,
}
else:
record = {
**existing,
"thread_id": thread_id,
"updated_at": now,
}
self._sessions[session_id] = record
return copy.deepcopy(record)
async def get(self, session_id: str) -> dict[str, Any] | None:
record = self._sessions.get(session_id)
return copy.deepcopy(record) if record is not None else None
@@ -0,0 +1,21 @@
"""ORM model for iframe embed session → thread lookups."""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class EmbedSessionRow(Base):
__tablename__ = "embed_sessions"
# Opaque id generated by the parent page and passed into the iframe URL.
session_id: Mapped[str] = mapped_column(String(128), primary_key=True)
thread_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
@@ -0,0 +1,49 @@
"""SQLAlchemy-backed embed session repository."""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.embed_sessions.base import EmbedSessionStore
from deerflow.persistence.embed_sessions.model import EmbedSessionRow
def _row_to_dict(row: EmbedSessionRow) -> dict[str, Any]:
data = row.to_dict()
for key in ("created_at", "updated_at"):
value = data.get(key)
if isinstance(value, datetime):
data[key] = value.isoformat()
return data
class EmbedSessionRepository(EmbedSessionStore):
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
async def upsert(self, session_id: str, thread_id: str) -> dict[str, Any]:
async with self._sf() as session:
row = await session.get(EmbedSessionRow, session_id)
now = datetime.now(UTC)
if row is None:
row = EmbedSessionRow(
session_id=session_id,
thread_id=thread_id,
created_at=now,
updated_at=now,
)
session.add(row)
else:
row.thread_id = thread_id
row.updated_at = now
await session.commit()
await session.refresh(row)
return _row_to_dict(row)
async def get(self, session_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(EmbedSessionRow, session_id)
return _row_to_dict(row) if row is not None else None
@@ -0,0 +1,39 @@
"""Create ``embed_sessions`` table for iframe embed status polling.
Revision ID: 20260602_01
Revises: 20260528_02
Create Date: 2026-06-02
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260602_01"
down_revision = "20260528_02"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def upgrade() -> None:
if not _has_table("embed_sessions"):
op.create_table(
"embed_sessions",
sa.Column("session_id", sa.String(128), primary_key=True),
sa.Column("thread_id", sa.String(64), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_embed_sessions_thread_id", "embed_sessions", ["thread_id"])
def downgrade() -> None:
if _has_table("embed_sessions"):
op.drop_table("embed_sessions")
@@ -0,0 +1,49 @@
"""Add step3 column to roundtable_drafts.
Step 3「结果绘制」persistence: the generated report (HTML + metadata) is stored
inline in the draft so a reloaded draft can show the report without depending on
the ephemeral report thread. The column is LONGTEXT on MySQL (HTML can exceed the
64 KB TEXT cap) and plain TEXT (unbounded) elsewhere — matching ``PortableLongText``
on ``RoundtableDraftRow.step3``. ``create_all`` adds the column for brand-new DBs;
this revision patches DBs that already have the table.
Revision ID: 20260603_01
Revises: 20260602_01
Create Date: 2026-06-03
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
from sqlalchemy.dialects.mysql import LONGTEXT
revision = "20260603_01"
down_revision = "20260602_01"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def _has_column(table_name: str, column_name: str) -> bool:
if not _has_table(table_name):
return False
inspector = inspect(op.get_bind())
return any(col["name"] == column_name for col in inspector.get_columns(table_name))
def upgrade() -> None:
if _has_table("roundtable_drafts") and not _has_column("roundtable_drafts", "step3"):
col_type = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
op.add_column("roundtable_drafts", sa.Column("step3", col_type, nullable=True))
def downgrade() -> None:
if _has_column("roundtable_drafts", "step3"):
op.drop_column("roundtable_drafts", "step3")
@@ -22,6 +22,7 @@ from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
from deerflow.persistence.models.run_event import RunEventRow
from deerflow.persistence.notifications.model import NotificationRow
from deerflow.persistence.recommended_questions.model import RecommendedQuestionRow
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.roundtable_drafts.model import (
RoundtableDraftRow,
RoundtableRecommendHistoryRow,
@@ -37,6 +38,7 @@ from deerflow.persistence.scheduled_tasks.model import (
from deerflow.persistence.skills.model import SkillDisplayAdapterRow, SkillRow
from deerflow.persistence.tags.model import TagAssignmentRow, TagRow
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.embed_sessions.model import EmbedSessionRow
from deerflow.persistence.thread_shares.model import ThreadShareRow
from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
from deerflow.persistence.user.model import UserRow
@@ -49,6 +51,7 @@ __all__ = [
"LlmCallMetricRow",
"NotificationRow",
"RecommendedQuestionRow",
"RoundtableChainRow",
"RoundtableDraftRow",
"RoundtableRecommendHistoryRow",
"RunEventRow",
@@ -62,6 +65,7 @@ __all__ = [
"SkillDisplayAdapterRow",
"TagAssignmentRow",
"TagRow",
"EmbedSessionRow",
"ThreadMetaRow",
"ThreadShareRow",
"ToolCallMetricRow",
@@ -0,0 +1,18 @@
"""Roundtable-planning business-chain persistence."""
from __future__ import annotations
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.roundtable_chains.sql import RoundtableChainRepository
__all__ = [
"RoundtableChainRepository",
"RoundtableChainRow",
"make_roundtable_chain_store",
]
def make_roundtable_chain_store(session_factory):
if session_factory is None:
return None
return RoundtableChainRepository(session_factory)
@@ -0,0 +1,44 @@
"""ORM model for roundtable-planning business chains.
A "business chain" (业务链条) is a named, reusable, *ordered* list of agents the
user assembles in the chain editor. In Step 2 the coordinator dispatches to the
seats following this order (see ``multi-agent-business-chain-dev.md``).
Persisted per user (like ``roundtable_drafts``) so it follows the account across
devices/sessions. ``seats`` is stored as a JSON array serialized into
``PortableLongText`` (MySQL ``LONGTEXT``) rather than ``PortableJSON`` for
headroom, mirroring the sibling roundtable persistence modules.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableLongText
class RoundtableChainRow(Base):
__tablename__ = "roundtable_chains"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
# Human-readable chain title shown on the picker card.
title: Mapped[str] = mapped_column(String(512), nullable=False, default="")
# Optional free-text description.
description: Mapped[str | None] = mapped_column(String(2048), nullable=True)
# JSON: ordered [{"agent_id": str, "name": str, "description": str}, ...].
# seats[0] is the first seat to speak. Stored as LONGTEXT-backed JSON string.
seats: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
created_at: Mapped[datetime] = mapped_column(
BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC)
)
updated_at: Mapped[datetime] = mapped_column(
BeijingDateTime(),
nullable=False,
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
@@ -0,0 +1,123 @@
"""SQLAlchemy-backed roundtable business-chain store.
The ``seats`` JSON column is stored as a serialized string in
``PortableLongText`` and round-tripped here with ``json.dumps`` / ``json.loads``
— same approach as ``roundtable_drafts`` / ``ai_writing_sessions``.
All write/read paths are scoped by ``user_id`` for ownership isolation; a row
that belongs to another user is treated as not found.
"""
from __future__ import annotations
import json
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
_UNSET = object()
def _loads(value: Any) -> Any:
if not isinstance(value, str):
return value
if not value:
return None
try:
return json.loads(value)
except Exception:
return None
def _dumps(value: Any) -> str | None:
if value is None:
return None
return json.dumps(value, ensure_ascii=False)
class RoundtableChainRepository:
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
@staticmethod
def _to_dict(row: RoundtableChainRow) -> dict[str, Any]:
return {
"id": row.id,
"title": row.title,
"description": row.description,
"seats": _loads(row.seats) or [],
"created_at": row.created_at.isoformat() if isinstance(row.created_at, datetime) else row.created_at,
"updated_at": row.updated_at.isoformat() if isinstance(row.updated_at, datetime) else row.updated_at,
}
async def list_chains(self, user_id: str, *, limit: int = 200) -> list[dict[str, Any]]:
stmt = (
select(RoundtableChainRow)
.where(RoundtableChainRow.user_id == user_id)
.order_by(RoundtableChainRow.updated_at.desc())
.limit(limit)
)
async with self._sf() as session:
result = await session.execute(stmt)
return [self._to_dict(row) for row in result.scalars()]
async def get_chain(self, chain_id: str, user_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(RoundtableChainRow, chain_id)
if row is None or row.user_id != user_id:
return None
return self._to_dict(row)
async def create_chain(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]:
now = datetime.now(UTC)
row = RoundtableChainRow(
id=data["id"],
user_id=user_id,
title=(data.get("title") or "").strip(),
description=(data.get("description") or None),
seats=_dumps(data.get("seats") or []),
created_at=now,
updated_at=now,
)
async with self._sf() as session:
session.add(row)
await session.commit()
await session.refresh(row)
return self._to_dict(row)
async def update_chain(
self,
chain_id: str,
user_id: str,
*,
title=_UNSET,
description=_UNSET,
seats=_UNSET,
) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(RoundtableChainRow, chain_id)
if row is None or row.user_id != user_id:
return None
if title is not _UNSET and title is not None:
row.title = str(title).strip()
if description is not _UNSET:
row.description = description or None
if seats is not _UNSET:
row.seats = _dumps(seats)
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return self._to_dict(row)
async def delete_chain(self, chain_id: str, user_id: str) -> bool:
async with self._sf() as session:
row = await session.get(RoundtableChainRow, chain_id)
if row is None or row.user_id != user_id:
return False
await session.delete(row)
await session.commit()
return True
@@ -33,6 +33,10 @@ class RoundtableDraftRow(Base):
# JSON blobs matching the frontend Step1Snapshot / Step2Snapshot shapes.
step1: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) # JSON object
step2: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) # JSON object
# Step 3「结果绘制」snapshot: { html, generatedAt, model, summary? }. The full
# report HTML is stored inline so a reloaded draft can show the report without
# depending on the ephemeral report thread (LONGTEXT — HTML can exceed 64 KB).
step3: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) # JSON object
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(
BeijingDateTime(),
@@ -64,6 +64,7 @@ class RoundtableDraftRepository:
d = cls._draft_to_meta(row)
d["step1"] = _loads(row.step1)
d["step2"] = _loads(row.step2)
d["step3"] = _loads(row.step3)
return d
async def list_drafts(self, user_id: str, *, limit: int = 100) -> list[dict[str, Any]]:
@@ -93,6 +94,7 @@ class RoundtableDraftRepository:
furthest_step=int(data.get("furthest_step") or 1),
step1=_dumps(data.get("step1")),
step2=_dumps(data.get("step2")),
step3=_dumps(data.get("step3")),
created_at=now,
updated_at=now,
)
@@ -111,6 +113,7 @@ class RoundtableDraftRepository:
furthest_step=_UNSET,
step1=_UNSET,
step2=_UNSET,
step3=_UNSET,
) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(RoundtableDraftRow, draft_id)
@@ -124,6 +127,8 @@ class RoundtableDraftRepository:
row.step1 = _dumps(step1)
if step2 is not _UNSET:
row.step2 = _dumps(step2)
if step3 is not _UNSET:
row.step3 = _dumps(step3)
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
@@ -4,11 +4,13 @@ from .memory_tool import memory_tool
from .present_file_tool import present_file_tool
from .scheduled_task_tool import scheduled_task_tool
from .setup_agent_tool import setup_agent
from .setup_writing_tool import setup_writing
from .task_tool import task_tool
from .view_image_tool import view_image_tool
__all__ = [
"setup_agent",
"setup_writing",
"present_file_tool",
"ask_clarification_tool",
"agent_orchestration_tool",
@@ -56,7 +56,7 @@ def notebook_search_tool(query: str, top_k: int = 10) -> str:
headers={"Authorization": f"Bearer {token}"},
json={
"query": query,
"source_ids": [source_ids],
"source_ids": source_ids,
"search_space_id": search_space_id,
"top_k": top_k,
},
@@ -0,0 +1,53 @@
"""写作配置助手用的提交工具。
与 ``setup_agent`` 同构:模型在「写作配置 Q&A」里与用户聊清需求后,调用本工具把建议的
写作配置「提交」出来。实际它总是被 ``SetupWritingApprovalMiddleware`` 在 ``after_model``
拦截(不真正执行)——拦截后把建议字段打包进审批 ToolMessage,前端据此回填左侧写作表单。
因此本工具的函数体基本是兜底:万一未被拦截(如审批被预先放行),也只回一句确认、不落任何盘。
"""
from __future__ import annotations
import logging
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langgraph.prebuilt import ToolRuntime
from langgraph.types import Command
logger = logging.getLogger(__name__)
@tool
def setup_writing(
user_intent: str,
runtime: ToolRuntime,
article_type: str | None = None,
word_count_target: int | None = None,
target_audience: str | None = None,
writing_mode: str | None = None,
material_source: str | None = None,
outline: str | None = None,
) -> Command:
"""把与用户聊清的写作配置提交给前端表单(不直接开始写作)。
Args:
user_intent: 用户的写作意图(精炼成一句话的写作目标)。
article_type: 文章类型的 key(须是系统给定的合法类型之一;不确定可留空)。
word_count_target: 目标字数(正整数)。
target_audience: 目标读者(如「行业专家」「普通读者」)。
writing_mode: 写作模式,'strict'(严格依据素材)或 'loose'(可结合通用知识)。
material_source: 素材来源,'knowledge_base'(知识库)或 'skill'(技能检索)。
outline: 若用户已提供大纲,原样填入其大纲文本(保留其层级与标题),否则留空。
"""
logger.info("[setup_writing] proposed writing config submitted (intercepted by approval middleware)")
return Command(
update={
"messages": [
ToolMessage(
content="写作配置已生成,请在右侧确认后应用到表单。",
tool_call_id=runtime.tool_call_id,
)
]
}
)