feat(ai写作): 素材收集专家按真实智能体运行时跑技能 + 步骤条对齐

- 新增 run_skill_via_real_agent:复用 SubagentExecutor(与 lead agent / 子智能体
  同一套运行时),加载技能 SKILL.md + 全套工具 + 懒加载沙箱,让技能的真实检索逻辑
  (读文件 / 调内网接口 / MCP)真正执行,再按 JSON 契约收割结构化素材;researcher 的
  _execute_skill 在 researcher_skill_agent 开启时切到它,失败/空结果回退直连检索。
  解决内网技能(在别的智能体里好使)在 AI 写作里检不出东西的问题。
- 步骤条轻量对齐别的智能体:技能调用拆成「读取 SKILL.md → 执行检索 → 检索到 N 条」
  三段(前两段新增,结果段沿用 skill_result);前端 SkillStepsCard 零改动。
- 含 roundtable-planning(业务链选择 / Step2 编排)的进行中改动。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
fengxiangdebaoziandClaude Opus 4.8 committed 2026-06-24 22:19:30 +08:00
1 parent 1a127ef602
commit afb78a6943
6 files changed
+164 -17

No files matched your search

@@ -140,7 +140,7 @@ export function BusinessChainPicker({
const hasCoord = (plan.coordinatorPrompt?.trim() ?? "") !== "";
return hasParallel || hasGoal || hasCoord ? plan : null;
}, [selectedChain, survivingSeats]);
const [dagExecMode, setDagExecMode] = useState<"fast" | "sandbox">("fast");
const [dagExecMode, setDagExecMode] = useState<"fast" | "sandbox">("sandbox");
const handleUse = () => {
if (!canUse || !selectedChain) return;
@@ -460,7 +460,7 @@ export function useStep2Orchestration(
gatherFirst = false,
orchestrationMode,
orchestrationPlan,
dagExecMode = "fast",
dagExecMode = "sandbox",
buildInitialIntent,
triggerToast,
setRoles,
@@ -192,7 +192,7 @@ export function buildChainSeatSelection(
const agents = plan.stages
.flatMap((st) => st.agentIds)
.map((id) => ({ agent_id: id, name: nameOf(id), model: modelOf(id), seat_skill_directive: skillOf(id), seatMode: seatModeOf(id) }));
return { agents, mode: "dag", chain: chainMeta, plan, dagExecMode: "fast", seatMode: chain.seatMode, gatherFirst: chain.gatherFirst ?? false };
return { agents, mode: "dag", chain: chainMeta, plan, dagExecMode: "sandbox", seatMode: chain.seatMode, gatherFirst: chain.gatherFirst ?? false };
}
}
@@ -360,7 +360,7 @@ export default function RoundtablePlanningPage() {
const [orchestrationPlan, setOrchestrationPlan] = useState<OrchestrationPlan | null>(null);
// dag 执行子模式:fast(并行双列禁写)/ sandbox(并行单列可写、沙箱跟随)。
// 由推荐弹窗的「执行方式」选择决定(仅 dag 有意义)。
const [dagExecMode, setDagExecMode] = useState<"fast" | "sandbox">("fast");
const [dagExecMode, setDagExecMode] = useState<"fast" | "sandbox">("sandbox");
const [step2SidebarExpanded, setStep2SidebarExpanded] = useState(true);
const [step2Runs, setStep2Runs] = useState<DraftStep2RunSnapshot[]>([]);
const [activeStep2RunId, setActiveStep2RunId] = useState<string | null>(null);
@@ -684,7 +684,7 @@ export default function RoundtablePlanningPage() {
setOrchestrationMode(run.orchestrationMode ?? "recommend");
setChainMeta(run.chain ?? null);
setOrchestrationPlan(run.orchestrationPlan ?? null);
setDagExecMode(run.dagExecMode ?? "fast");
setDagExecMode(run.dagExecMode ?? "sandbox");
setStep3Report(run.step3 ?? null);
step2.hydrateFromDraft(run as unknown as Step2Snapshot);
markRestoredConsensusAsEmitted(Boolean(run.hasConsensus));
@@ -791,7 +791,7 @@ export default function RoundtablePlanningPage() {
setOrchestrationMode(snapshot.orchestrationMode ?? "recommend");
setChainMeta(snapshot.chain ?? null);
setOrchestrationPlan(snapshot.orchestrationPlan ?? null);
setDagExecMode(snapshot.dagExecMode ?? "fast");
setDagExecMode(snapshot.dagExecMode ?? "sandbox");
step2.hydrateFromDraft(snapshot as unknown as Step2Snapshot);
markRestoredConsensusAsEmitted(Boolean(snapshot.hasConsensus));
},
@@ -945,7 +945,7 @@ export default function RoundtablePlanningPage() {
// chain / dag 都来自业务链条,保留来源链名;recommend 无。
chain: result.mode === "recommend" ? null : result.chain ?? null,
orchestrationPlan: result.mode === "dag" ? result.plan ?? null : null,
dagExecMode: result.mode === "dag" ? result.dagExecMode ?? "fast" : undefined,
dagExecMode: result.mode === "dag" ? result.dagExecMode ?? "sandbox" : undefined,
step3: null,
};
setStep2Runs([...syncedRuns, newRun]);
@@ -998,7 +998,7 @@ export default function RoundtablePlanningPage() {
setGatherFirst(result.gatherFirst ?? false);
setChainMeta(result.mode === "recommend" ? null : result.chain ?? null);
setOrchestrationPlan(result.mode === "dag" ? result.plan ?? null : null);
setDagExecMode(result.mode === "dag" ? result.dagExecMode ?? "fast" : "fast");
setDagExecMode(result.mode === "dag" ? result.dagExecMode ?? "sandbox" : "sandbox");
setIsRecommendDialogOpen(false);
setCurrentStep(2);
triggerToast(
@@ -1527,7 +1527,7 @@ export default function RoundtablePlanningPage() {
setOrchestrationMode("recommend");
setChainMeta(null);
setOrchestrationPlan(null);
setDagExecMode("fast");
setDagExecMode("sandbox");
setStep3Report(null);
// 停掉页面层的 Step3 三个生成流(新任务不该带着旧任务的流跑)+ 复位进入闸门,
// 并停止后台作业实时跟随(作业本身继续在后台跑)。
@@ -41,6 +41,148 @@ def use_skill_search_agent() -> bool:
return bool(cfg.get("researcher_skill_agent", False)) if isinstance(cfg, dict) else False
# ── 用真实智能体运行时跑技能(和「别的智能体调用技能」完全一致)────────────────────
#
# 旧的 run_skill_search_agent 只给模型绑了一个写死路由到 DDG 的通用 search 工具,技能
# SKILL.md 里要用的真实工具(read_file / bash / 内网检索工具 / MCP)根本不存在,所以
# 内网技能跑不出东西。本函数改为复用 **SubagentExecutor**(与 lead agent / 子智能体同一套
# 运行时):加载技能 SKILL.md(skills 白名单)+ 全套工具 + 懒加载沙箱,让技能按其说明真正
# 检索,再让模型把检索到的真实素材按 JSON 契约输出,我们解析成结构化素材。
#
# 单次工具调用循环上限(防跑飞)。
_REAL_AGENT_MAX_TURNS = 16
_MATERIAL_OUTPUT_CONTRACT = (
"你的任务:严格按上面挂载的检索技能(SKILL.md)的说明,用相应工具检索与「写作意图」相关的**真实素材**。\n"
"- 必须按技能说明实际调用工具完成检索(读取知识库文件 / 调用内网检索接口 / MCP 等),**不得凭记忆编造**。\n"
"- 检索完成后,**只输出一个 JSON 对象**(不要解释、不要 markdown 代码块、用英文双引号),把检索到的真实素材列出:\n"
'{"materials": [{"title": "标题", "content": "素材正文摘录(尽量保留原文, <=500字)", "source": "来源", "url": "原文链接(可空)", "published_at": "日期(可空)"}]}\n'
"- content 必须来自检索到的真实内容;若确实没检索到任何素材,输出 {\"materials\": []}。"
)
async def run_skill_via_real_agent(
*,
skill,
intent: str,
keywords: list[str],
user_query: str,
config: RunnableConfig,
state: dict,
model_name_fallback: str | None,
max_results: int,
timeout_s: int,
emit_query=None,
) -> tuple[list, str]:
"""用真实智能体运行时(SubagentExecutor)跑一个检索技能,收割结构化素材。
与「别的智能体调用技能」一致:技能 SKILL.md + 全套工具 + 沙箱都齐全,技能的真实检索
逻辑(读文件 / 调内网接口 / MCP)能真正执行。模型最终按 JSON 契约吐出素材,解析成
SearchResult。返回 ``(results, error_detail)``:results 为空且 error 非空 = 该技能失败
(researcher 据此回退直连检索)。
"""
from deerflow.config import get_app_config
from deerflow.subagents.config import SubagentConfig
from deerflow.subagents.executor import SubagentExecutor, SubagentStatus
from deerflow.tools import get_available_tools
from ..search.base import SearchResult
from ._agent_runner import RESEARCHER_AGENT_ID, WritingAgentUnavailable, load_writing_agent
from ._utils import parse_json
skill_name = getattr(skill, "name", "") or "技能"
try:
agent_cfg, soul = load_writing_agent(RESEARCHER_AGENT_ID)
except WritingAgentUnavailable:
agent_cfg, soul = None, ""
model_name = resolve_agent_model_name(
agent_cfg, state, config, explicit_fallback=model_name_fallback, strict=False
)
try:
app_config = get_app_config()
tools = get_available_tools(
model_name=model_name, subagent_enabled=False, include_mcp=True, app_config=app_config
)
except Exception as e: # noqa: BLE001
logger.warning("skill real-agent[%s]: build tools failed: %s", skill_name, e)
return [], f"工具初始化失败: {e}"
system_prompt = (soul + "\n\n" if soul else "") + _MATERIAL_OUTPUT_CONTRACT
task_lines = [f"写作意图:{intent or '(无)'}"]
if user_query:
task_lines.append(f"用户补充检索意图:{user_query}")
if keywords:
task_lines.append(f"可参考的检索方向:{', '.join(keywords[:8])}")
task_lines.append(f"目标素材数量约 {max_results} 条。")
task = "\n".join(task_lines)
sub_cfg = SubagentConfig(
name=f"ai-writing-skill-{skill_name}",
description="AI 写作素材检索",
system_prompt=system_prompt,
tools=None, # 继承全套工具,技能按 SKILL.md 自行取用
disallowed_tools=["task"], # 素材检索不需要再派生子任务
skills=[skill_name], # 仅加载这个技能的 SKILL.md
model=model_name or "inherit",
max_turns=_REAL_AGENT_MAX_TURNS,
timeout_seconds=max(30, timeout_s),
)
thread_id = (config.get("configurable") or {}).get("thread_id")
try:
executor = SubagentExecutor(
config=sub_cfg, tools=tools, app_config=app_config,
parent_model=model_name, thread_id=thread_id,
)
except Exception as e: # noqa: BLE001
logger.warning("skill real-agent[%s]: executor init failed: %s", skill_name, e)
return [], f"运行时初始化失败: {e}"
# 步骤条「轻量对齐」别的智能体:读取 SKILL.md → 执行检索 两段(结果段由 researcher
# 的 _emit_skill_outcome 以 skill_result 收尾)。这两步在图的事件循环里同步发出,
# 写到时间线没问题;技能真正执行在下面的线程里跑。
if emit_query:
emit_query(f"读取技能「{skill_name}」的 SKILL.md…")
emit_query(f"技能「{skill_name}」按 SKILL.md 说明执行检索…")
# execute() 内部用「沙箱/MCP 友好」的隔离事件循环跑;放进线程,避免阻塞图的事件循环。
try:
result = await asyncio.to_thread(executor.execute, task)
except Exception as e: # noqa: BLE001
logger.warning("skill real-agent[%s]: execute failed: %s", skill_name, e, exc_info=True)
return [], f"技能检索失败: {e}"
if result.status != SubagentStatus.COMPLETED:
return [], result.error or f"技能检索未完成({result.status.value})"
parsed = parse_json(result.result or "", {"materials": []})
raw_materials = parsed.get("materials")
if not isinstance(raw_materials, list):
return [], ""
results: list = []
for m in raw_materials[:max_results]:
if not isinstance(m, dict):
continue
content = str(m.get("content") or "").strip()
title = str(m.get("title") or "").strip()
if not content and not title:
continue
results.append(SearchResult(
title=title or "(无标题)",
content=content[:500],
source=str(m.get("source") or skill_name).strip() or skill_name,
published_at=str(m.get("published_at") or "").strip(),
url=str(m.get("url") or "").strip(),
relevance_score=0.8,
source_type="skill",
))
return results, ""
def _read_skill_body(skill) -> str:
"""读技能 SKILL.md 全文;失败回退 name+description。"""
try:
@@ -40,7 +40,7 @@ from ._agent_runner import (
run_writing_agent,
use_builtin_agent,
)
from ._skill_agent import run_skill_search_agent, use_skill_search_agent
from ._skill_agent import run_skill_via_real_agent, use_skill_search_agent
from ._utils import get_ai_writing_config, keyword_outline_section, llm_json, safe_get_stream_writer
logger = logging.getLogger(__name__)
@@ -715,17 +715,20 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
error_detail = ";".join(dict.fromkeys(errs))[:200]
return added, error_detail
else:
# ── 智能体式技能调用(config ai_writing.researcher_skill_agent 打开时)──
# 把技能 SKILL.md 注入 + 绑定 search 工具,让模型按技能说明驱动检索,
# 收割工具返回的 SearchResult 作为素材(与「智能体调用技能」逻辑一致)。
# ── 用真实智能体运行时跑技能(config ai_writing.researcher_skill_agent 打开时)──
# 复用 SubagentExecutor:技能 SKILL.md + 全套工具 + 沙箱齐全,技能真实检索逻辑
# (读文件 / 调内网接口 / MCP)能真正执行,再按 JSON 契约收割结构化素材——和
# 「别的智能体调用技能」完全一致,解决内网技能检不出东西的问题。
if use_skill_search_agent():
def _emit_q(q: str, _name=sk.name) -> None:
def _emit_q(msg: str, _name=sk.name) -> None:
# run_skill_via_real_agent 传入的是完整步骤文案(读取 SKILL.md /
# 执行检索…),原样透传到步骤条,与别的智能体的技能调用步骤对齐。
_emit_skill_step({
"type": "invoke_skill",
"message": f"素材收集专家用技能「{_name}」检索:{q[:60]}",
"message": msg,
"skill_name": _name,
})
agent_results, agent_err = await run_skill_search_agent(
agent_results, agent_err = await run_skill_via_real_agent(
skill=sk,
intent=intent,
keywords=keywords,
@@ -739,7 +742,9 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
)
if agent_results:
return _collect(agent_results), ""
# 弱模型不会工具调用 / 智能体检索失败 → 回退直连检索,保证素材不空。
# 技能运行时失败 / 没检索到 → 回退直连检索,保证素材不空。
if agent_err:
logger.info("skill real-agent[%s] returned no materials: %s", sk.name, agent_err)
# 定向技能检索(默认路径 / 智能体模式回退):user_query 模式优先用用户原话;
# 非知识库技能再附上 description 让 query 贴合该技能的领域。