NathanAlex7
9ef970320b
五智能体流水线真实提交后返回结果序列化失败(WindowsPath 不可 JSON 序列化)。 将 file 字段转为 str, 确保 run() 返回值可被 API 序列化返回。
236 lines
9.6 KiB
Python
236 lines
9.6 KiB
Python
"""XingbangWorkflow — 五智能体审批流水线 + 自进化闭环。
|
|
|
|
执行流程:
|
|
data_curator → risk_scout → case_finder → approval_advisor → risk_monitor
|
|
(case_finder 与 risk_scout 可并行, 此处先串行保证可追溯)
|
|
|
|
每个 agent 的产出写入输出目录, 供下游读取; 同时全程记录
|
|
轨迹到 evolution 模块, 跑完后计算奖励并更新进化洞察。
|
|
|
|
API 风格: run() 返回完整结果; 也可用 run_stream() 逐步 yield 进度。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any, AsyncGenerator
|
|
|
|
from xingbang.config import agent_specs
|
|
from xingbang.agents import build_agents, XingbangAgent
|
|
from xingbang.evolution.trajectory import TrajectoryRecorder
|
|
from xingbang.evolution.reward import compute_reward_v1
|
|
from xingbang.evolution.optimizer import EvolutionEngine
|
|
|
|
|
|
class XingbangWorkflow:
|
|
"""兴邦五智能体审批工作流。"""
|
|
|
|
def __init__(
|
|
self,
|
|
llm,
|
|
base_dir: Path,
|
|
output_dir: Path | None = None,
|
|
evolution_enabled: bool = True,
|
|
):
|
|
self._base_dir = Path(base_dir)
|
|
self._output_dir = output_dir or (self._base_dir / "xingbang_output")
|
|
self._output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 自进化模块
|
|
self.evolution_enabled = evolution_enabled
|
|
self._recorder = TrajectoryRecorder(self._base_dir, enabled=evolution_enabled)
|
|
self._evolution = EvolutionEngine(self._base_dir, self._recorder, enabled=evolution_enabled)
|
|
|
|
# 五智能体(注入证据回调 → 轨迹)
|
|
self._agents: dict[str, XingbangAgent] = build_agents(
|
|
llm=llm,
|
|
base_dir=self._output_dir,
|
|
output_dir=self._output_dir,
|
|
on_evidence=self._on_evidence,
|
|
)
|
|
|
|
# 流水线顺序
|
|
self._pipeline = ["data_curator", "risk_scout", "case_finder",
|
|
"approval_advisor", "risk_monitor"]
|
|
self._results: dict[str, dict] = {}
|
|
|
|
def _on_evidence(self, kind: str, data: dict):
|
|
# 轨迹采集(保持简单: 在 workflow 层显式 add_step)
|
|
pass
|
|
|
|
@property
|
|
def recorder(self) -> TrajectoryRecorder:
|
|
return self._recorder
|
|
|
|
@property
|
|
def evolution(self) -> EvolutionEngine:
|
|
return self._evolution
|
|
|
|
# ── 单次评审 ────────────────────────────────────────
|
|
async def run(self, task: str, customer: str = "") -> dict[str, Any]:
|
|
"""运行完整五智能体审批流水线,返回各步产出 + 轨迹 + 奖励。"""
|
|
workflow_id = f"wf_{int(time.time())}_{uuid.uuid4().hex[:4]}"
|
|
traj_id = self._recorder.begin(workflow_id=workflow_id, customer=customer)
|
|
|
|
# 第一步任务描述
|
|
tasks = self._build_tasks(task)
|
|
final_output = ""
|
|
summary_parts = []
|
|
|
|
for key in self._pipeline:
|
|
agent = self._agents[key]
|
|
agent_task = tasks.get(key, task)
|
|
|
|
# 收集上游上下文(该 agent 依赖的产出)
|
|
ctx = self._collect_context(key, agent_task)
|
|
|
|
t0 = time.time()
|
|
try:
|
|
result = await agent.run(agent_task, context=ctx)
|
|
success = True
|
|
output = result["output"]
|
|
file = result.get("file")
|
|
except Exception as e: # noqa
|
|
success = False
|
|
output = f"[执行失败] {e}"
|
|
file = None
|
|
|
|
elapsed_ms = int((time.time() - t0) * 1000)
|
|
|
|
# 记录这一步到轨迹
|
|
step = {
|
|
"agent": key,
|
|
"name": agent.spec.name,
|
|
"input": agent_task[:300],
|
|
"tools": ["read_xingbang_data", "read_output"],
|
|
"output_file": str(file) if file else None,
|
|
"output_preview": output[:200],
|
|
"success": success,
|
|
"time_ms": elapsed_ms,
|
|
}
|
|
self._recorder.add_step(step)
|
|
self._results[key] = {"output": output, "file": file, "success": success}
|
|
summary_parts.append(f"[{key}] {'✅成功' if success else '❌失败'}")
|
|
final_output = output # 最后一环(风险哨兵)作为总结论
|
|
|
|
# 收尾: 计算奖励 + 保存进化洞察
|
|
trajectory = self._recorder.finalize(
|
|
success=all(self._results[k]["success"] for k in self._pipeline),
|
|
summary=" | ".join(summary_parts),
|
|
)
|
|
reward = compute_reward_v1(trajectory)
|
|
trajectory["cumulative_reward"] = reward
|
|
insights_path = self._evolution.save_insights()
|
|
|
|
return {
|
|
"workflow_id": workflow_id,
|
|
"trajectory_id": traj_id,
|
|
"customer": customer,
|
|
"reward": reward,
|
|
"results": {
|
|
k: {
|
|
kk: (str(vv) if kk == "file" and vv is not None else vv)
|
|
for kk, vv in v.items()
|
|
if kk in ("output", "file", "success")
|
|
}
|
|
for k, v in self._results.items()
|
|
},
|
|
"final_summary": summary_parts,
|
|
"evolution_insights_file": str(insights_path),
|
|
}
|
|
|
|
# ── 逐步流式(供前端展示进度) ───────────────────────
|
|
async def run_stream(self, task: str, customer: str = "") -> AsyncGenerator[dict, None]:
|
|
"""逐步 yield 每步进度与产出,便于前端 SSE 展示。"""
|
|
workflow_id = f"wf_{int(time.time())}_{uuid.uuid4().hex[:4]}"
|
|
traj_id = self._recorder.begin(workflow_id=workflow_id, customer=customer)
|
|
tasks = self._build_tasks(task)
|
|
|
|
for key in self._pipeline:
|
|
agent = self._agents[key]
|
|
agent_task = tasks.get(key, task)
|
|
ctx = self._collect_context(key, agent_task)
|
|
yield {"event": "agent_start", "agent": key, "name": agent.spec.name}
|
|
|
|
t0 = time.time()
|
|
try:
|
|
result = await agent.run(agent_task, context=ctx)
|
|
success = True
|
|
output = result["output"]
|
|
file = result.get("file")
|
|
except Exception as e: # noqa
|
|
success = False
|
|
output = f"[执行失败] {e}"
|
|
file = None
|
|
elapsed_ms = int((time.time() - t0) * 1000)
|
|
|
|
step = {
|
|
"agent": key, "name": agent.spec.name, "input": agent_task[:300],
|
|
"tools": ["read_xingbang_data", "read_output"],
|
|
"output_file": str(file) if file else None,
|
|
"output_preview": output[:200], "success": success, "time_ms": elapsed_ms,
|
|
}
|
|
self._recorder.add_step(step)
|
|
self._results[key] = {"output": output, "file": file, "success": success}
|
|
yield {"event": "agent_end", "agent": key, "name": agent.spec.name,
|
|
"success": success, "output": output, "file": str(file) if file else None}
|
|
|
|
trajectory = self._recorder.finalize(success=True)
|
|
reward = compute_reward_v1(trajectory)
|
|
trajectory["cumulative_reward"] = reward
|
|
insights_path = self._evolution.save_insights()
|
|
yield {"event": "done", "workflow_id": workflow_id,
|
|
"trajectory_id": traj_id, "reward": reward,
|
|
"insights_file": str(insights_path)}
|
|
|
|
# ── 辅助 ────────────────────────────────────────────
|
|
def _build_tasks(self, user_task: str) -> dict[str, str]:
|
|
"""为每个智能体构造针对性的任务描述。"""
|
|
return {
|
|
"data_curator": (
|
|
f"请读取兴邦数据包中的多源数据, 为客户『{user_task}』(申请评审) "
|
|
f"生成结构化客户情报卡, 保存到 customer_intel.md。"
|
|
),
|
|
"risk_scout": (
|
|
f"基于上游 customer_intel.md, 对客户『{user_task}』做尽调分析, "
|
|
f"生成尽调报告初稿, 保存到 due_diligence.md。"
|
|
),
|
|
"case_finder": (
|
|
f"检索历史案例库与制度规则, 为『{user_task}』这类项目找出最相关案例, "
|
|
f"生成历史案例参考, 保存到 case_reference.md。"
|
|
),
|
|
"approval_advisor": (
|
|
f"汇总情报卡/尽调/历史案例, 按三档分级制度为『{user_task}』"
|
|
f"输出审批参考意见, 保存到 approval_opinion.md。"
|
|
),
|
|
"risk_monitor": (
|
|
f"审批前动态复核最新风控数据, 为『{user_task}』输出风控复核结论, "
|
|
f"保存到 risk_recheck.md。"
|
|
),
|
|
}
|
|
|
|
def _collect_context(self, agent_key: str, task: str) -> str:
|
|
"""收集该 agent 依赖的上游产出作为上下文。"""
|
|
spec = agent_specs()[agent_key]
|
|
parts = []
|
|
for dep in spec.inputs_from:
|
|
res = self._results.get(self._dep_owner(dep))
|
|
if res and res.get("output"):
|
|
parts.append(f"[{dep}]\n{res['output'][:2000]}")
|
|
if parts:
|
|
return "\n\n".join(parts)
|
|
return ""
|
|
|
|
def _dep_owner(self, output_file: str) -> str:
|
|
"""根据产出文件名反查产生它的 agent key。"""
|
|
mapping = {
|
|
"customer_intel.md": "data_curator",
|
|
"due_diligence.md": "risk_scout",
|
|
"case_reference.md": "case_finder",
|
|
"approval_opinion.md": "approval_advisor",
|
|
"risk_recheck.md": "risk_monitor",
|
|
}
|
|
return mapping.get(output_file, "data_curator")
|