JOJO
4d9b709a9e
工作流运行时： - 状态机与编排：modules/workflow_state_manager.py + server/workflow_flow.py （激活快照/阶段推进/审核节点/分支决策/柔性通知/max_stage_rounds 撞限询问） - 五个工具（activate/report_stage/choose_branch/get_status/deactivate） 与 REST API（server/workflow_runtime_api.py） - 前端：QuickDock 工作流窗口（三段式进度，推进/驳回/完成/退出动画）、 slash 菜单激活与退出、轮询事件消费、进入对话状态回填 - 审核：modules/workflow_review_agent.py（pass/reject 把关节点） 审核智能体统一配置： - 个人空间新增「审核智能体」标签页：自动审批/目标/工作流三个审核智能体 统一选择模型+思考模式+超时/轮次参数 - modules/review_agent_config.py 统一解析（复用子智能体模型库）， 废除独立 json 配置（auto_approval/goal_review/workflow_review） - goal 审核接入 max_rounds 上限（原常量未接线）；workflow 审核硬编码 6 轮改为可配 联调修复： - /new 空对话激活：后端自动创建对话并完整继承模式参数 （work_mode/permission/execution/reasoning_effort，修复思考模式丢失） - 激活/通知消息 starts_work=True，恢复智能体回复头部与工作计时 - 节点目录改为从开始节点拓扑遍历（修复按保存顺序显示错乱） - QuickDock 乐观掩码不再掩盖工作流实时状态（修复 /new 激活窗口瞬关+延迟瞬开）； /new 路由不套用全局内容缓存（修复空对话展开空白数秒后收回） - 工作流完成先广播完成态快照再摘牌，窗口播完落定+退出动画再收起 - 激活提示中的工具名修正为 report_workflow_stage
349 lines
17 KiB
Python
349 lines
17 KiB
Python
"""工作流阶段审核智能体（Workflow Review Agent）。
|
||
|
||
由 modules/goal_review_agent.py fork 而来。职责不同：
|
||
- goal_review_agent 判断"长期目标是否真正达成"（done/continue）。
|
||
- workflow_review_agent 判断"工作流当前阶段的产出是否达到进入下一阶段的门槛"（pass/reject）。
|
||
|
||
两种审核模式（与 goal_review 同构）：
|
||
- readonly：只给 report_workflow_review 一个工具，不注入 run_command。
|
||
- active：额外注入只读 run_command 工具，允许其取证后再下结论。
|
||
|
||
返回结构：{"decision": "pass"|"reject", "message": str, "source": "workflow_review_agent"}。
|
||
兜底（异常/超时/未产出结论/配置缺失）一律返回 reject，message 说明审核服务异常
|
||
并提示主智能体「请告知用户」——定稿决策：审核异常视为驳回（计入 reject_counts），
|
||
用户看到消息后能发现是审核服务问题而非真实驳回。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Dict, List, Optional
|
||
|
||
import httpx
|
||
|
||
from config import PROMPTS_DIR, LOGS_DIR
|
||
from modules.goal_state_manager import REVIEW_MODE_ACTIVE, REVIEW_MODE_READONLY
|
||
from modules.review_agent_config import resolve_review_agent_config
|
||
|
||
PROMPT_NAME = "workflow_review_agent.txt"
|
||
DEFAULT_TIMEOUT_SECONDS = 120
|
||
DEFAULT_MAX_ROUNDS = 6
|
||
DEFAULT_MAX_COMMAND_TIMEOUT = 60
|
||
DEBUG_SAVE_WORKFLOW_REVIEW_TRANSCRIPT = True
|
||
DEBUG_TRANSCRIPT_DIR = Path(LOGS_DIR) / "workflow_review_agent"
|
||
|
||
# 兜底驳回消息（审核智能体未能产出明确结论时使用）
|
||
FALLBACK_REJECT_MESSAGE = (
|
||
"本次审核未能正常完成（审核服务异常或未产出结论），按驳回处理。"
|
||
"请告知用户审核服务可能异常；若属偶发，可稍后重新汇报本阶段。"
|
||
)
|
||
|
||
|
||
def load_workflow_review_agent_config() -> Dict[str, Any]:
|
||
"""加载工作流审核智能体配置：统一走个人空间「审核智能体」设置 + 子智能体模型库。"""
|
||
cfg = resolve_review_agent_config("workflow_review")
|
||
cfg["name"] = "workflow-review-agent"
|
||
return cfg
|
||
|
||
|
||
class WorkflowReviewAgent:
|
||
def __init__(self, *, web_terminal: Any):
|
||
self.web_terminal = web_terminal
|
||
self.cfg = load_workflow_review_agent_config()
|
||
|
||
@staticmethod
|
||
def _system_prompt(review_mode: str) -> str:
|
||
try:
|
||
template = (Path(PROMPTS_DIR) / PROMPT_NAME).read_text(encoding="utf-8")
|
||
except Exception:
|
||
template = "你是工作流阶段审核智能体。你必须调用 report_workflow_review 返回 pass 或 reject。"
|
||
|
||
active_start = "{{ACTIVE_REVIEW_ONLY}}"
|
||
active_end = "{{/ACTIVE_REVIEW_ONLY}}"
|
||
if review_mode == REVIEW_MODE_ACTIVE:
|
||
return template.replace(active_start, "").replace(active_end, "").strip()
|
||
|
||
while active_start in template and active_end in template:
|
||
start = template.find(active_start)
|
||
end = template.find(active_end, start)
|
||
if start < 0 or end < 0:
|
||
break
|
||
template = template[:start] + template[end + len(active_end):]
|
||
return template.strip()
|
||
|
||
@staticmethod
|
||
def _report_tool() -> Dict[str, Any]:
|
||
return {
|
||
"type": "function",
|
||
"function": {
|
||
"name": "report_workflow_review",
|
||
"description": (
|
||
"提交工作流阶段审核结论并结束本次审核。pass=阶段产出合格，进入下一阶段；"
|
||
"reject=阶段产出不合格，打回整改。"
|
||
),
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"decision": {
|
||
"type": "string",
|
||
"enum": ["pass", "reject"],
|
||
"description": "pass=通过；reject=驳回。",
|
||
},
|
||
"message": {
|
||
"type": "string",
|
||
"description": (
|
||
"decision=pass 时简述通过理由（给主智能体与用户看）；"
|
||
"decision=reject 时写给主执行模型的整改意见，需具体可执行，"
|
||
"指出缺什么、怎么补、重新审核时要看到什么证据。"
|
||
),
|
||
},
|
||
},
|
||
"required": ["decision", "message"],
|
||
},
|
||
},
|
||
}
|
||
|
||
@staticmethod
|
||
def _run_command_tool() -> Dict[str, Any]:
|
||
return {
|
||
"type": "function",
|
||
"function": {
|
||
"name": "run_command",
|
||
"description": (
|
||
"在终端中执行只读指令。可用于查看文件内容、搜索代码/文本、检查目录结构、"
|
||
"查看 git 状态与差异、运行测试以核实阶段产出是否合格。"
|
||
"禁止用于写入、删除、改权限或其他修改性操作。"
|
||
),
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"command": {
|
||
"type": "string",
|
||
"description": (
|
||
"要执行的终端命令字符串。应优先使用只读命令，例如 ls、cat、grep、find、"
|
||
"rg、git status、git diff、pwd、stat、以及运行测试的命令等。"
|
||
),
|
||
}
|
||
},
|
||
"required": ["command"],
|
||
},
|
||
},
|
||
}
|
||
|
||
def _build_tools(self, review_mode: str) -> List[Dict[str, Any]]:
|
||
if review_mode == REVIEW_MODE_ACTIVE:
|
||
return [self._run_command_tool(), self._report_tool()]
|
||
return [self._report_tool()]
|
||
|
||
async def _run_readonly_command(self, command: str) -> Dict[str, Any]:
|
||
work_path = Path(getattr(self.web_terminal.context_manager, "project_path", "."))
|
||
timeout = int(self.cfg.get("max_command_timeout") or DEFAULT_MAX_COMMAND_TIMEOUT)
|
||
try:
|
||
result = await self.web_terminal.terminal_ops.run_command(
|
||
command=command,
|
||
working_dir=str(work_path),
|
||
timeout=timeout,
|
||
sandbox_write_access=False,
|
||
)
|
||
return result if isinstance(result, dict) else {"success": False, "error": "run_command 返回格式异常"}
|
||
except Exception as exc:
|
||
return {"success": False, "error": str(exc)}
|
||
|
||
async def review(
|
||
self,
|
||
*,
|
||
payload_text: str,
|
||
review_mode: str = REVIEW_MODE_READONLY,
|
||
progress_cb: Optional[Callable[[Dict[str, Any]], None]] = None,
|
||
cancel_check: Optional[Callable[[], Optional[Dict[str, Any]]]] = None,
|
||
) -> Dict[str, Any]:
|
||
review_mode = review_mode if review_mode in (REVIEW_MODE_READONLY, REVIEW_MODE_ACTIVE) else REVIEW_MODE_READONLY
|
||
debug_transcript: List[Dict[str, Any]] = []
|
||
|
||
def _trace(event: str, payload: Dict[str, Any]) -> None:
|
||
if not DEBUG_SAVE_WORKFLOW_REVIEW_TRANSCRIPT:
|
||
return
|
||
debug_transcript.append({"ts": int(time.time() * 1000), "event": event, "payload": payload})
|
||
|
||
def _flush_trace(final_result: Dict[str, Any]) -> None:
|
||
if not DEBUG_SAVE_WORKFLOW_REVIEW_TRANSCRIPT:
|
||
return
|
||
try:
|
||
DEBUG_TRANSCRIPT_DIR.mkdir(parents=True, exist_ok=True)
|
||
row = {
|
||
"created_at": int(time.time() * 1000),
|
||
"review_mode": review_mode,
|
||
"messages": messages,
|
||
"trace": debug_transcript,
|
||
"final_result": final_result,
|
||
}
|
||
file = DEBUG_TRANSCRIPT_DIR / f"workflow_review_{int(time.time() * 1000)}.json"
|
||
file.write_text(json.dumps(row, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
def _reject(message: str) -> Dict[str, Any]:
|
||
return {"decision": "reject", "message": message or FALLBACK_REJECT_MESSAGE, "source": "workflow_review_agent"}
|
||
|
||
url = str(self.cfg.get("url") or "").strip()
|
||
key = str(self.cfg.get("key") or "").strip()
|
||
model = str(self.cfg.get("model") or "").strip()
|
||
if not url or not key or not model:
|
||
out = _reject(f"工作流审核智能体配置缺失，无法完成审核。{FALLBACK_REJECT_MESSAGE}")
|
||
_flush_trace(out)
|
||
return out
|
||
endpoint = f"{url.rstrip('/')}/chat/completions"
|
||
headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
|
||
timeout_seconds = int(self.cfg.get("timeout_seconds") or DEFAULT_TIMEOUT_SECONDS)
|
||
max_rounds = max(1, int(self.cfg.get("max_rounds") or DEFAULT_MAX_ROUNDS))
|
||
extra_params = dict(self.cfg.get("extra_params") or {})
|
||
tools = self._build_tools(review_mode)
|
||
|
||
messages: List[Dict[str, Any]] = [
|
||
{"role": "system", "content": self._system_prompt(review_mode)},
|
||
{"role": "user", "content": payload_text},
|
||
]
|
||
|
||
async with httpx.AsyncClient(timeout=timeout_seconds) as client:
|
||
rounds = 0
|
||
forced_tool_retry_used = False
|
||
while True:
|
||
rounds += 1
|
||
if cancel_check:
|
||
external = cancel_check()
|
||
if external:
|
||
return external
|
||
if rounds > max_rounds:
|
||
out = _reject(f"审核超过最大轮次仍未产出结论。{FALLBACK_REJECT_MESSAGE}")
|
||
_flush_trace(out)
|
||
return out
|
||
if progress_cb:
|
||
progress_cb({"stage": "model_call", "round": rounds, "message": f"审核轮次 {rounds}"})
|
||
req = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"tools": tools,
|
||
"tool_choice": "auto",
|
||
"temperature": 0.0,
|
||
**extra_params,
|
||
}
|
||
_trace("request", {"round": rounds, "request": req})
|
||
try:
|
||
resp = await client.post(endpoint, headers=headers, json=req)
|
||
resp.raise_for_status()
|
||
except httpx.HTTPStatusError as exc:
|
||
body = ""
|
||
try:
|
||
body = exc.response.text
|
||
except Exception:
|
||
body = str(exc)
|
||
_trace("http_error", {"round": rounds, "status_code": exc.response.status_code if exc.response else None, "body": body})
|
||
# 兼容某些模型/网关不接受额外参数：自动降级重试一次（去掉 extra_params）
|
||
if extra_params:
|
||
retry_req = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"tools": tools,
|
||
"tool_choice": "auto",
|
||
"temperature": 0.0,
|
||
}
|
||
_trace("retry_request_without_extra_params", {"round": rounds, "request": retry_req})
|
||
retry_resp = await client.post(endpoint, headers=headers, json=retry_req)
|
||
try:
|
||
retry_resp.raise_for_status()
|
||
resp = retry_resp
|
||
except httpx.HTTPStatusError:
|
||
out = _reject(f"审核请求失败({retry_resp.status_code})。{FALLBACK_REJECT_MESSAGE}")
|
||
_flush_trace(out)
|
||
return out
|
||
else:
|
||
out = _reject(
|
||
f"审核请求失败({exc.response.status_code if exc.response else 'unknown'})。{FALLBACK_REJECT_MESSAGE}"
|
||
)
|
||
_flush_trace(out)
|
||
return out
|
||
except Exception as exc:
|
||
_trace("request_exception", {"round": rounds, "error": str(exc)})
|
||
out = _reject(f"审核请求异常（{exc}）。{FALLBACK_REJECT_MESSAGE}")
|
||
_flush_trace(out)
|
||
return out
|
||
|
||
choice = ((resp.json().get("choices") or [{}])[0] or {}).get("message") or {}
|
||
reasoning_content = (
|
||
choice.get("reasoning_content") or choice.get("reasoning") or choice.get("thinking") or ""
|
||
)
|
||
_trace("response", {"round": rounds, "message": choice, "reasoning_content": reasoning_content})
|
||
tool_calls = choice.get("tool_calls") or []
|
||
content = choice.get("content")
|
||
|
||
if not tool_calls:
|
||
if content or reasoning_content:
|
||
messages.append(
|
||
{
|
||
"role": "assistant",
|
||
"content": str(content or ""),
|
||
"reasoning_content": str(reasoning_content or ""),
|
||
}
|
||
)
|
||
reminder = (
|
||
"禁止直接输出结论文本。必须调用工具：必要时先调用 run_command 核实，"
|
||
"再调用 report_workflow_review 给出最终结论。"
|
||
)
|
||
if not forced_tool_retry_used:
|
||
forced_tool_retry_used = True
|
||
reminder = "你刚刚未通过工具给出明确结果。必须调用工具，禁止直接输出内容。请立即调用 report_workflow_review 返回最终结论。"
|
||
messages.append({"role": "user", "content": reminder})
|
||
continue
|
||
|
||
# 有 tool_calls：追加完整 assistant 消息（与主智能体结构对齐）
|
||
assistant_message = {
|
||
"role": "assistant",
|
||
"content": str(content or ""),
|
||
"reasoning_content": str(reasoning_content or ""),
|
||
"tool_calls": tool_calls,
|
||
}
|
||
messages.append(assistant_message)
|
||
for call in tool_calls:
|
||
fn = (call.get("function") or {}).get("name")
|
||
raw_args = (call.get("function") or {}).get("arguments") or "{}"
|
||
try:
|
||
args = json.loads(raw_args) if isinstance(raw_args, str) else (raw_args or {})
|
||
except Exception:
|
||
args = {}
|
||
if fn == "report_workflow_review":
|
||
decision = str(args.get("decision") or "").strip().lower()
|
||
message = str(args.get("message") or "").strip()
|
||
if decision == "pass":
|
||
out = {"decision": "pass", "message": message or "审核通过。", "source": "workflow_review_agent"}
|
||
_flush_trace(out)
|
||
return out
|
||
if decision == "reject":
|
||
out = _reject(message or "审核未通过，请按整改意见补充后重新汇报。")
|
||
_flush_trace(out)
|
||
return out
|
||
# 非法 decision：保守驳回
|
||
out = _reject(message or f"审核返回了无法识别的结论。{FALLBACK_REJECT_MESSAGE}")
|
||
_flush_trace(out)
|
||
return out
|
||
if fn == "run_command":
|
||
cmd = str(args.get("command") or "").strip()
|
||
if progress_cb:
|
||
progress_cb({"stage": "run_command", "round": rounds, "command": cmd})
|
||
tool_result = await self._run_readonly_command(cmd) if cmd else {"success": False, "error": "command 不能为空"}
|
||
_trace("tool_result", {"round": rounds, "tool": "run_command", "command": cmd, "result": tool_result})
|
||
tool_content = json.dumps(tool_result, ensure_ascii=False)
|
||
messages.append(
|
||
{
|
||
"role": "tool",
|
||
"content": tool_content,
|
||
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime())
|
||
+ f".{int((time.time() % 1) * 1000000):06d}",
|
||
"message_id": f"msg_{uuid.uuid4().hex}",
|
||
"tool_call_id": call.get("id"),
|
||
"name": "run_command",
|
||
}
|
||
)
|