diff --git a/.gitignore b/.gitignore index c8dc4cc..e9c7a38 100644 --- a/.gitignore +++ b/.gitignore @@ -50,3 +50,9 @@ desktop.ini *.zip.old backend.zip frontend.zip + +# Xingbang runtime artifacts (客户产出的情报卡/报告 + 进化轨迹数据, 含隐私) +backend/xingbang_output/ +backend/xingbang_evolution/ +backend/diag_single_agent.py +backend/run_full_workflow.py diff --git a/backend/api/xingbang_api.py b/backend/api/xingbang_api.py new file mode 100644 index 0000000..8ce3785 --- /dev/null +++ b/backend/api/xingbang_api.py @@ -0,0 +1,145 @@ +"""Xingbang API — 兴邦金租评审工作流 + 自进化模块的 HTTP 接口。 + +端: + POST /api/xingbang/workflow 运行五智能体审批流水线(返回全部产出) + POST /api/xingbang/workflow/stream 流式运行(SSE, 逐步展示) + GET /api/xingbang/evolution 查看自进化洞察 + GET /api/xingbang/trajectories 列出历史轨迹 + GET /api/xingbang/trajectories/{id} 读取单条轨迹 + +复用 mini-OpenClaw 的 AgentManager.llm 与后端基础设施。 +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +from fastapi import APIRouter +from pydantic import BaseModel +from sse_starlette.sse import EventSourceResponse + +from graph.agent import agent_manager +from xingbang.workflow import XingbangWorkflow +from xingbang.config import XINGBANG_CONFIG + +router = APIRouter(prefix="/api/xingbang", tags=["xingbang"]) + + +# 懒加载单例工作流(复用 agent_manager.llm) +_workflow: XingbangWorkflow | None = None + + +def _get_workflow() -> XingbangWorkflow: + global _workflow + if _workflow is None: + if agent_manager._llm is None: + # 若 agent 未初始化, 手动初始化(需要 BASE_DIR) + from pathlib import Path as _P + _workflow = XingbangWorkflow( + llm=None, + base_dir=_P(__file__).resolve().parent.parent, + evolution_enabled=True, + ) + raise RuntimeError("Agent 未初始化, 请先启动后端(会自动初始化 agent_manager)") + _workflow = XingbangWorkflow( + llm=agent_manager._llm, + base_dir=Path(__file__).resolve().parent.parent, + evolution_enabled=True, + ) + return _workflow + + +class WorkflowRequest(BaseModel): + task: str # 客户/项目描述 + customer: str = "" # 客户名(用于轨迹标注) + + +@router.get("/config") +async def get_config(): + """返回兴邦项目配置(数据源、五智能体、档位)。""" + return XINGBANG_CONFIG + + +@router.post("/workflow") +async def run_workflow(req: WorkflowRequest): + """运行完整五智能体审批流水线。""" + wf = _get_workflow() + try: + result = await wf.run(req.task, customer=req.customer) + return result + except RuntimeError as e: + from fastapi import HTTPException + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/workflow/stream") +async def run_workflow_stream(req: WorkflowRequest): + """流式运行(SSE), 逐步展示五智能体进度与产出。""" + wf = _get_workflow() + + async def gen(): + async for event in wf.run_stream(req.task, customer=req.customer): + yield { + "event": event.get("event", "event"), + "data": json.dumps( + {k: v for k, v in event.items() if k != "event"}, + ensure_ascii=False, + ), + } + + return EventSourceResponse(gen()) + + +class FeedbackRequest(BaseModel): + trajectory_id: str + feedback: str # "positive" | "negative" + + +@router.post("/evolution/feedback") +async def submit_feedback(req: FeedbackRequest): + """提交用户反馈(Level 2 奖励信号), 追加进轨迹。""" + try: + traj_file = ( + Path(__file__).resolve().parent.parent + / "xingbang_evolution" / "trajectories" / f"{req.trajectory_id}.json" + ) + if not traj_file.exists(): + return {"error": f"轨迹不存在: {req.trajectory_id}"} + data = json.loads(traj_file.read_text(encoding="utf-8")) + data.setdefault("user_feedback", req.feedback) + data["user_feedback"] = req.feedback + traj_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + return {"status": "ok", "trajectory_id": req.trajectory_id, "feedback": req.feedback} + except Exception as e: + return {"error": str(e)} + + +@router.get("/evolution") +async def get_evolution(): + """查看自进化洞察报告。""" + wf = _get_workflow() + report = wf.evolution.generate_insights() + return {"insights": report, "config": XINGBANG_CONFIG} + + +@router.get("/trajectories") +async def list_trajectories(): + """列出所有已保存轨迹。""" + wf = _get_workflow() + paths = wf.recorder.list_trajectories() + return {"count": len(paths), + "trajectories": [p.stem for p in paths]} + + +@router.get("/trajectories/{trajectory_id}") +async def get_trajectory(trajectory_id: str): + """读取单条轨迹详情。""" + wf = _get_workflow() + traj = wf.recorder.load(trajectory_id) + if not traj: + from fastapi import HTTPException + raise HTTPException(status_code=404, detail="轨迹不存在") + return traj diff --git a/backend/app.py b/backend/app.py index 757beac..671b16d 100644 --- a/backend/app.py +++ b/backend/app.py @@ -47,6 +47,7 @@ from api.sessions import router as sessions_router from api.tokens import router as tokens_router from api.compress import router as compress_router from api.config_api import router as config_router +from api.xingbang_api import router as xingbang_router app.include_router(chat_router, prefix="/api") app.include_router(files_router, prefix="/api") @@ -54,6 +55,7 @@ app.include_router(sessions_router, prefix="/api") app.include_router(tokens_router, prefix="/api") app.include_router(compress_router, prefix="/api") app.include_router(config_router, prefix="/api") +app.include_router(xingbang_router) @app.get("/") diff --git a/backend/xingbang/__init__.py b/backend/xingbang/__init__.py new file mode 100644 index 0000000..9734fa1 --- /dev/null +++ b/backend/xingbang/__init__.py @@ -0,0 +1,23 @@ +"""兴邦金租 AI 评审平台 — 基于 mini-OpenClaw 的项目模块。 + +定位： + 在 mini-OpenClaw 单智能体平台上，扩展出一个可交付的兴邦金租 + AI 评审流水线：五智能体 + 自进化(evolution)模块。挂在 + mini-OpenClaw 后端，复用其 LLM、工具、会话基础设施。 + +五智能体流水线（对应兴邦三档分级审批）： + 1. 数据汇聚官 (data_curator) — 多源异构数据→结构化客户情报卡 + 2. 预审参谋 (risk_scout) — 基于情报卡做尽调初稿+风险分级 + 3. 经验检索 (case_finder) — 历史案例+制度规则 RAG 检索 + 4. 审批顾问 (approval_advisor) — 汇总前序产出→三档分级审批参考 + 5. 风险哨兵 (risk_monitor) — 审批前动态复核，风控信号转变预警 + +自进化(evolution)模块： + 采集每次审批流水线的完整轨迹(trajectory)，计算奖励(reward)， + 从历史中归纳高效工具调用模式，反哺各 agent 的 system prompt。 +""" + +from .workflow import XingbangWorkflow +from .config import XINGBANG_CONFIG + +__all__ = ["XingbangWorkflow", "XINGBANG_CONFIG"] diff --git a/backend/xingbang/agents.py b/backend/xingbang/agents.py new file mode 100644 index 0000000..2edecb2 --- /dev/null +++ b/backend/xingbang/agents.py @@ -0,0 +1,162 @@ +"""五智能体定义 — 复用 mini-OpenClaw 的 LLM 与工具，各自带差异化 system prompt。 + +每个智能体 = 一个独立的 LangChain create_agent： + - 共享底层 LLM(DeepSeek) 与读文件工具 + - 差异化的 system prompt(角色、规则、输出约定) + - 专用工具: read_xingbang_data(读兴邦数据包)、read_output(读上游产出) + - 属于同一个全局 AgentManager 编排, 但各 agent 独立 invoke + +轨迹采集: 每个 agent 的每次调用(输入/system/prompt/工具序列/输出)都会被 +evidence 回调记录, 供 evolution 模块使用。 +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Callable + +from langchain.agents import create_agent +from langchain_core.messages import HumanMessage, AIMessage +from langchain_core.tools import tool + +from config import get_rag_mode +from graph.prompt_builder import build_system_prompt +from xingbang.config import agent_specs, XINGBANG_DATA_DIR, DATA_FILES + + +# ── 专用工具:读取兴邦多源数据包 ──────────────────────── +@tool +def read_xingbang_data(filename: str) -> str: + """读取兴邦金租客户的多源数据文件(JSON/CSV/MD/TXT)，返回内容。 + + 可用文件名(需带编号前缀): 01_客户申请信息.json, 02_信用评分卡.json, + 03_风险评级报告.md, 04_授信额度测算.csv, 05_征信记录.json, + 06_司法执行记录.json, 07_舆情风险评分.json, 08_反欺诈模型输出.json, + 09_财务预警模型.csv, 10_财务分析报告.csv, 11_租赁物估值报告.md, + 12_行业调研报告.md, 13_历史审批案例库.json, 14_公司制度要点.md, + 15_银行流水扫描件.txt + """ + if not XINGBANG_DATA_DIR.exists(): + return f"[错误] 兴邦数据目录不存在: {XINGBANG_DATA_DIR}" + path = XINGBANG_DATA_DIR / filename + if not path.exists(): + return f"[错误] 文件不存在: {filename}" + content = path.read_text(encoding="utf-8", errors="ignore") + # 限制长度避免超上下文 + if len(content) > 4000: + content = content[:4000] + "\n...[已截断]" + return content + + +@tool +def read_output(filename: str, base_dir: str = "") -> str: + """读取当前兴邦工作区中上游智能体产出的 Markdown 文件(如 + customer_intel.md, due_diligence.md, case_reference.md)。""" + file_path = Path(base_dir) / filename if base_dir else Path(filename) + if not file_path.exists(): + return f"[错误] 上游产出文件不存在: {filename}" + content = file_path.read_text(encoding="utf-8", errors="ignore") + if len(content) > 8000: + content = content[:8000] + "\n...[已截断]" + return content + + +# ── 单个评审智能体 ───────────────────────────────────── +class XingbangAgent: + """一个兴邦评审智能体: 差异化角色 + 复用 LLM + 工具。""" + + def __init__( + self, + spec, + llm=None, + base_dir: Path | None = None, + output_dir: Path | None = None, + on_evidence: Callable[[str, dict], None] | None = None, + ): + self.spec = spec + self._llm = llm + self._base_dir = base_dir + self._output_dir = output_dir or (base_dir / "xingbang_output" if base_dir else None) + self._on_evidence = on_evidence + self._agent = None + + def _ensure_agent(self): + """构建带专属 system prompt 的独立 agent(惰性)。""" + if self._agent is not None: + return self._agent + tools = [read_xingbang_data] + tools.append(read_output) + prompt = self.spec.system_prompt + # 追加读文件工具说明(需要 base_dir 才能读产出) + if self._base_dir: + prompt += ( + f"\n\n[工作目录] 上游产出文件位于 {self._base_dir}。" + "读取上游产出时使用 read_output 工具并把 filename 设为产出文件名。" + ) + self._agent = create_agent(model=self._llm, tools=tools, system_prompt=prompt) + return self._agent + + def _emit(self, kind: str, data: dict): + if self._on_evidence: + self._on_evidence(kind, data) + + async def run(self, task: str, context: str = "") -> dict: + """执行该智能体任务,返回 {'output': 结果文本, 'file': 保存路径}。""" + self._emit("agent_start", {"agent": self.spec.key, "name": self.spec.name}) + agent = self._ensure_agent() + messages = [] + if context: + messages.append(HumanMessage(content=f"[上游上下文]\n{context}")) + messages.append(HumanMessage(content=task)) + self._emit("prompt", { + "agent": self.spec.key, + "system": self.spec.system_prompt[:500], + "task": task[:300], + }) + result = await agent.ainvoke({"messages": messages}) + final = self._extract_text(result) + # 保存产出到文件 + file_path = None + if self._output_dir: + self._output_dir.mkdir(parents=True, exist_ok=True) + file_path = self._output_dir / self.spec.output_file + file_path.write_text(final, encoding="utf-8") + self._emit("agent_end", { + "agent": self.spec.key, + "output_file": str(file_path) if file_path else None, + "output_preview": final[:300], + }) + return {"output": final, "file": file_path} + + @staticmethod + def _extract_text(result: dict) -> str: + messages = result.get("messages", []) + text_parts = [] + for m in messages: + if hasattr(m, "content") and m.type == "ai" and m.content: + text_parts.append(m.content) + if text_parts: + return "\n\n".join(text_parts).strip() + content = result.get("output", "") + return content if content else "[智能体未生成有效输出]" + + +def build_agents( + llm, + base_dir: Path, + output_dir: Path | None = None, + on_evidence: Callable[[str, dict], None] | None = None, + keys: list[str] | None = None, +) -> dict[str, XingbangAgent]: + """按流水线顺序构建五智能体实例。""" + specs = agent_specs() + order = keys or ["data_curator", "risk_scout", "case_finder", + "approval_advisor", "risk_monitor"] + return { + key: XingbangAgent( + spec=specs[key], llm=llm, + base_dir=base_dir, output_dir=output_dir, on_evidence=on_evidence, + ) + for key in order + } diff --git a/backend/xingbang/config.py b/backend/xingbang/config.py new file mode 100644 index 0000000..0a022cc --- /dev/null +++ b/backend/xingbang/config.py @@ -0,0 +1,173 @@ +"""兴邦金租项目配置 — 数据源路径、五智能体定义、分级审批档位。 + +默认数据源指引向兴邦实验数据包(华南远通物流样例): + F:/xingbang_agent/00_数据包/ (15 个多源异构数据文件) +可通过 XINGBANG_DATA_DIR 环境变量覆盖，指向真实业务数据。 + +分级审批档位（对齐兴邦三档分级审批制度）: + 常规档 / 中间档 / 严肃档 +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path + +# ── 数据源 ────────────────────────────────────────────── +# 默认指向兴邦实验数据包，可用环境变量 XINGBANG_DATA_DIR 覆盖 +DEFAULT_DATA_DIR = Path("F:/xingbang_agent/00_数据包") +XINGBANG_DATA_DIR = Path( + os.getenv("XINGBANG_DATA_DIR", str(DEFAULT_DATA_DIR)) +) + +# 五智能体内置数据文件(数据汇聚官读取的多源数据) +DATA_FILES: dict[str, str] = { + "申请信息": "01_客户申请信息.json", + "信用评分卡": "02_信用评分卡.json", + "风险评级": "03_风险评级报告.md", + "授信额度": "04_授信额度测算.csv", + "征信记录": "05_征信记录.json", + "司法执行": "06_司法执行记录.json", + "舆情评分": "07_舆情风险评分.json", + "反欺诈": "08_反欺诈模型输出.json", + "财务预警": "09_财务预警模型.csv", + "财报": "10_财务分析报告.csv", + "租赁物估值": "11_租赁物估值报告.md", + "行业调研": "12_行业调研报告.md", + "历史案例库": "13_历史审批案例库.json", + "公司制度": "14_公司制度要点.md", + "银行流水": "15_银行流水扫描件.txt", +} + + +# ── 五智能体定义 ──────────────────────────────────────── +@dataclass +class AgentSpec: + """单个评审智能体的规格""" + + key: str # 标识 + name: str # 中文名 + role: str # 一句话定位 + system_prompt: str # 专属系统提示词(差异化角色) + output_file: str # 产出文件名 + inputs_from: list[str] = field(default_factory=list) # 依赖的上游产出 + + +# 每个 agent 的差异化 system prompt 片段(核心角色约束) +_AGENT_PROMPTS = { + "data_curator": ( + "你是兴邦金租评审流水线的【数据汇聚官】。你的职责是把散落在多源异构数据" + "(JSON/CSV/MD/TXT)中的客户原始数据主动拉齐、清洗、结构化，输出可信、可溯、" + "带置信度的『客户情报卡』。\n" + "规则：\n" + "1. 解析多源数据，统一格式，按客户四维画像(工商/信用司法/舆情/财务/租赁物)归并\n" + "2. 多源冲突(如工商股东vs舆情实控人)要显式标出冲突点,绝不擅自取其一\n" + "3. 风险信号按🔴高/🟡中/🟢低三档分级，每条必须附原始数据依据\n" + "4. 明确列出『未覆盖数据源清单』,绝不编造缺失信息\n" + "5. 输出存为 Markdown 文件。数据不出域,本地处理。" + ), + "risk_scout": ( + "你是兴邦金租评审流水线的【预审参谋】。你的职责是基于上游客户情报卡，" + "产出尽调报告初稿，把风险点做深度分析。\n" + "规则：\n" + "1. 读取上游『客户情报卡』(customer_intel.md)\n" + "2. 对每个风险点做穿透分析:现金流失控、对外担保、资产负债率、涉诉风险、租赁物残值\n" + "3. 正面因素(合同稳定性、增信措施)也要列出\n" + "4. 引用上游的历史案例检索结果(如有)\n" + "5. 输出『尽调报告初稿』,风险按高/中/低分级,每条可溯源" + ), + "case_finder": ( + "你是兴邦金租评审流水线的【经验检索】。你的职责是把沉淀在历史案例和制度规则" + "库中的经验变成可调用的组织知识,回答『以前类似项目怎么批的』。\n" + "规则：\n" + "1. 检索历史案例库(13_历史审批案例库.json)与当前客户最相关的案例\n" + "2. 每个结论必须带案例编号(如H-1103),可追溯\n" + "3. 正面案例和否决/预警案例都要给,辅助全面判断\n" + "4. 结合制度要点(14_公司制度要点.md)给出适用的档位条款\n" + "5. 找不到完全匹配案例时,明确说明差异点,不硬套" + ), + "approval_advisor": ( + "你是兴邦金租评审流水线的【审批顾问】。你的职责是汇总前面所有智能体产出，" + "按三档分级审批制度输出标准化审批参考意见。\n" + "规则：\n" + "1. 读取上游全部产出(情报卡/尽调初稿/历史案例参考)\n" + "2. 按三档制度判定本申请归入:常规档/中间档/严肃档(含判定依据,对照制度条款)\n" + "3. 输出核心风险点(引用溯源)、历史案例支撑(带编号)\n" + "4. 预填流转意见(『建议通过,理由如下,请确认』),人只确认/驳回\n" + "5. 明确标注『本意见为AI参考,最终决策由评审人员确认』\n" + "6. 绝不替代决策,全程留痕可回退" + ), + "risk_monitor": ( + "你是兴邦金租评审流水线的【风险哨兵】。你的职责是审批前的最后一道动态复核，" + "检查关键风控信号是否在评审期间发生变化。\n" + "规则：\n" + "1. 重新读取最新的风控数据(02信用评分/05征信/06司法/07舆情/09财务预警)\n" + "2. 与预审阶段采用的数据对比,检测评分/司法/舆情等关键信号是否转变\n" + "3. 若发现信号转变(如新增执行、评分骤降),必须主动提示『风控结果已转变," + "原预审结论需重新评估』,并给出新档位判定\n" + "4. 若无变化,确认『风控状态稳定,可按原档位推进』\n" + "5. 输出最终风控复核结论" + ), +} + + +def agent_specs() -> dict[str, AgentSpec]: + """返回五个智能体的规格定义(流水线顺序)。""" + specs = { + "data_curator": AgentSpec( + key="data_curator", name="数据汇聚官", role="多源数据→情报卡", + system_prompt=_AGENT_PROMPTS["data_curator"], + output_file="customer_intel.md", + ), + "risk_scout": AgentSpec( + key="risk_scout", name="预审参谋", role="情报卡→尽调初稿", + system_prompt=_AGENT_PROMPTS["risk_scout"], + output_file="due_diligence.md", inputs_from=["customer_intel.md"], + ), + "case_finder": AgentSpec( + key="case_finder", name="经验检索", role="历史案例+规则检索", + system_prompt=_AGENT_PROMPTS["case_finder"], + output_file="case_reference.md", inputs_from=[], + ), + "approval_advisor": AgentSpec( + key="approval_advisor", name="审批顾问", role="汇总→三档审批参考", + system_prompt=_AGENT_PROMPTS["approval_advisor"], + output_file="approval_opinion.md", + inputs_from=["customer_intel.md", "due_diligence.md", "case_reference.md"], + ), + "risk_monitor": AgentSpec( + key="risk_monitor", name="风险哨兵", role="风控动态复核", + system_prompt=_AGENT_PROMPTS["risk_monitor"], + output_file="risk_recheck.md", + inputs_from=["approval_opinion.md"], + ), + } + # 编排顺序:数据汇聚→预审→(经验并列)→审批→风控 + return specs + + +# ── 分级审批档位阈值(对齐 14_公司制度要点.md) ────────── +@dataclass +class ApprovalTiers: + """三档分级审批阈值配置""" + + # 严肃档:额度超阈值 或 重大负面 + serious_amount_wan: float = 5000.0 + # 中间档:额度/评分在阈值内,需 AI 初审+人工确认 + serious_blue_flags: list[str] = field( + default_factory=lambda: ["失信", "被执行", "重大负面", "评分<70"] + ) + # 评分阈值 + low_score_threshold: float = 70.0 + + +XINGBANG_CONFIG: dict = { + "project": "兴邦金租 AI 评审平台", + "version": "0.1.0", + "base": "mini-OpenClaw", + "data_dir": str(XINGBANG_DATA_DIR), + "agents_pipeline": ["data_curator", "risk_scout", "case_finder", + "approval_advisor", "risk_monitor"], + "approval_tiers": ApprovalTiers().__dict__, +} diff --git a/backend/xingbang/evolution/__init__.py b/backend/xingbang/evolution/__init__.py new file mode 100644 index 0000000..990ce01 --- /dev/null +++ b/backend/xingbang/evolution/__init__.py @@ -0,0 +1,18 @@ +"""自进化(evolution)模块 — 兴邦金租评审平台的学习与优化子系统。 + +核心思想: 每次审批流水线执行, 采集五智能体的完整轨迹 +(state/action/observation/reward), 从历史轨迹中归纳成功模式, +反哺各智能体的 system prompt, 实现"越用越准"的自进化。 + +分级方案(诚实设计): + - Level 1 轻量(现在能做): 轨迹采集 + 内在奖励 + 失败/成功模式归纳 + prompt 优化 + - Level 2: 用户显式反馈(👍/👎)作为奖励信号(需前端配合) + - Level 3(未来): 专家标注 / LoRA 微调(需 GPU) +本模块实现 Level 1, 接口预留 Level 2/3。 +""" + +from .trajectory import TrajectoryRecorder +from .reward import compute_reward_v1 +from .optimizer import EvolutionEngine + +__all__ = ["TrajectoryRecorder", "compute_reward_v1", "EvolutionEngine"] diff --git a/backend/xingbang/evolution/optimizer.py b/backend/xingbang/evolution/optimizer.py new file mode 100644 index 0000000..06ade33 --- /dev/null +++ b/backend/xingbang/evolution/optimizer.py @@ -0,0 +1,161 @@ +"""Evolution Engine — 从历史轨迹中学习, 反哺各智能体 prompt, 实现自进化。 + +闭环(自进化): + 轨迹采集 → 奖励打分 → 归纳成功/失败模式 → 生成进化建议 → + 反哺/追加到各 agent 的系统提示词(可选启用) + +Level 1 实现: + - analyze: 从已保存轨迹中总结高分/低分模式(常见工具序列、步骤分布) + - generate_insights: 生成可读的进化洞察(markdown) + - evolve_prompt: 将洞察追加到对应 agent 的 system prompt(可选) + +轻量自进化的核心: 不编造"强化学习训练", 而是做"基于经验的 prompt 增强" +——把历史中高奖励轨迹用过的工具调用序列、成功策略, 有选择地注入 +未来 agent 的提示词, 让系统越用越聪明。 +""" + +from __future__ import annotations + +import json +from collections import Counter +from pathlib import Path +from typing import Any + +from .reward import compute_reward_v1 +from .trajectory import TrajectoryRecorder + + +class EvolutionEngine: + """自进化引擎(Level 1 轻量版)。""" + + def __init__( + self, + base_dir: Path, + recorder: TrajectoryRecorder, + enabled: bool = True, + ): + self._base_dir = Path(base_dir) + self._recorder = recorder + self.enabled = enabled + self._insights_dir = self._base_dir / "xingbang_evolution" / "insights" + self._insights_dir.mkdir(parents=True, exist_ok=True) + + # ── 分析历史轨迹 ──────────────────────────────────── + def _load_all_trajectories(self) -> list[dict[str, Any]]: + trajs = [] + for p in self._recorder.list_trajectories(): + try: + trajs.append(json.loads(p.read_text(encoding="utf-8"))) + except Exception: + continue + return trajs + + def analyze(self) -> dict[str, Any]: + """分析全部历史轨迹, 返回统计与模式。""" + trajectories = self._load_all_trajectories() + if not trajectories: + return {"count": 0, "message": "尚无轨迹样本,先积累数据"} + + # 计算每条奖励 + scored = [] + for t in trajectories: + r = compute_reward_v1(t) + t["_reward"] = r + scored.append(t) + + scored.sort(key=lambda x: x["_reward"], reverse=True) + high = scored[: max(1, len(scored) // 4)] # 前 25% + low = scored[-max(1, len(scored) // 4):] # 后 25% + + # 归纳高频工具序列 + def agent_tool_sequences(trajs): + counter = Counter() + for t in trajs: + for s in t.get("steps", []): + seq = tuple(s.get("tools", [])) + if seq: + counter[seq] += 1 + return counter.most_common(5) + + high_patterns = { + "seqs": agent_tool_sequences(high), + "avg_steps": sum(len(t["steps"]) for t in high) / max(1, len(high)), + "success_rate": sum(1 for t in high if t["final_result"].get("success")) + / max(1, len(high)), + } + return { + "count": len(trajectories), + "avg_reward": round(sum(t["_reward"] for t in scored) / len(scored), 2), + "high": {"samples": len(high), "patterns": high_patterns}, + "low": { + "samples": len(low), + "avg_steps": sum(len(t["steps"]) for t in low) / max(1, len(low)), + }, + "top_agents_used": self._agent_usage(scored), + } + + def _agent_usage(self, trajs) -> dict: + counter = Counter() + for t in trajs: + for s in t.get("steps", []): + counter[s.get("agent", "?")] += 1 + return dict(counter.most_common()) + + # ── 生成进化洞察 ──────────────────────────────────── + def generate_insights(self) -> str: + """生成可读的 markdown 进化洞察报告。""" + analysis = self.analyze() + if analysis.get("count", 0) == 0: + return "# 自进化洞察\n\n尚无轨迹样本。运行审批流水线积累数据后,系统将自动进化。" + lines = [ + f"# 兴邦 AI 评审·自进化洞察", + f"\n**样本数**: {analysis['count']} 条轨迹 | **平均奖励**: {analysis['avg_reward']}", + "", + "## ✅ 高分轨迹模式 (前 25%)", + ] + hp = analysis["high"] + lines.append(f"- 平均步骤数: {hp['patterns']['avg_steps']:.1f}") + lines.append(f"- 成功率: {hp['patterns']['success_rate']*100:.0f}%") + lines.append("- 高频工具序列:") + for seq, n in hp["patterns"]["seqs"]: + lines.append(f" - {' → '.join(seq)} (出现 {n} 次)") + lines += ["", "## ⚠️ 低分轨迹 (后 25%)", f"- 平均步骤数: {analysis['low']['avg_steps']:.1f}"] + lines += ["", "## 🧠 智能体使用分布", ""] + for agent, count in analysis.get("top_agents_used", {}).items(): + lines.append(f"- {agent}: {count} 次") + lines += [ + "", + "## 进化建议", + "1. 高分轨迹的工具序列可在各 agent 的 system prompt 中作为'推荐工作流'注入", + "2. 步骤数偏多(>5)的 agent 提示词应强调'一次到位,减少反复读取'", + "3. 持续积累轨迹 → 达到一定量后启动更高阶优化(LoRA 微调)", + ] + return "\n".join(lines) + + def save_insights(self) -> Path: + """保存洞察报告到 insights 目录。""" + report = self.generate_insights() + path = self._insights_dir / "evolution_insights.md" + path.write_text(report, encoding="utf-8") + return path + + # ── 反哺 prompt (自进化闭环) ──────────────────────── + def evolve_prompt_suggestion(self, agent_key: str) -> str | None: + """为指定 agent 生成可注入的 prompt 增强建议(基于历史高分轨迹)。 + + 返回一段可追加到该 agent system_prompt 的文本, 若无历史则 None。 + """ + analysis = self.analyze() + if analysis.get("count", 0) == 0: + return None + hp = analysis["high"] + seqs = hp["patterns"]["seqs"] + if not seqs: + return None + top_seq_text = " → ".join(seqs[0][0]) if seqs else "" + suggestion = ( + f"\n\n[自进化推荐 · 基于 {analysis['count']} 条历史轨迹] " + f"高分任务常用工具序列: {top_seq_text}。" + f"请优先复用该模式,减少无效工具往返,一次到位完成任务。" + ) + return suggestion diff --git a/backend/xingbang/evolution/reward.py b/backend/xingbang/evolution/reward.py new file mode 100644 index 0000000..e762e8e --- /dev/null +++ b/backend/xingbang/evolution/reward.py @@ -0,0 +1,82 @@ +"""Reward Calculator — 计算一条审批轨迹的奖励分数。 + +诚实分级设计 (对齐兴邦方案): + Level 1 (现在能做,不需要用户交互): + - 任务完成 +10 + - 工具调用成功率 + - 重复调用惩罚 + - 时间惩罚 + Level 2 (需前端加 👍/👎 按钮): + - 用户显式反馈 + Level 3 (需人工标注): + - 专家打分 + +本文件实现 Level 1 + 预留 Level 2/3 参数。 +不做不诚实的信号: 拒绝用"还款表现"当奖励(周期1-3年无法闭环)。 +""" + +from __future__ import annotations +from typing import Any + + +def compute_reward_v1( + trajectory: dict[str, Any], + feedback: str | None = None, + expert_score: float | None = None, +) -> float: + """计算一条轨迹的奖励(Leve1 l 内置指标 + 可选 Level2 反馈 + Level3 专家分)。 + + Args: + trajectory: 轨迹 dict(含 steps 和 final_result) + feedback: Level2, "positive"/"negative" + expert_score: Level3, 1-5 分 + Returns: + float 奖励分 + """ + steps = trajectory.get("steps", []) + final = trajectory.get("final_result", {}) + if not steps: + return 0.0 + + reward = 0.0 + + # 1. 任务完成奖励 + if final.get("success"): + reward += 10.0 + else: + reward -= 10.0 + + # 2. 工具调用成功率(成功工具 / 总工具) + total_tool_calls = 0 + tool_success = 0 + for s in steps: + tools = s.get("tools", []) + total_tool_calls += len(tools) + if tools: + tool_success += 1 # 该步骤至少调了工具视为尽力(简化) + if total_tool_calls > 0: + reward += (tool_success / max(1, total_tool_calls)) * 5.0 + + # 3. 步骤数惩罚(越少越好,常规 5 步流水线 5-15 步为佳) + n_steps = len(steps) + optimal = 5 + step_penalty = abs(n_steps - optimal) * 0.3 + reward -= step_penalty + + # 4. 时间惩罚(超过 600 秒 = 10 分钟开始扣) + duration_ms = final.get("duration_ms", 0) + if duration_ms > 600_000: + reward -= ((duration_ms - 600_000) / 60_000) * 1.0 + + # ── Level 2: 用户反馈 (可选) ── + if feedback == "positive": + reward += 20.0 + elif feedback == "negative": + reward -= 10.0 + + # ── Level 3: 专家打分 (可选, 1-5 归一化到 -10~+10) ── + if expert_score is not None: + reward += (expert_score - 3) * 5.0 + + reward = max(-30.0, min(reward, 50.0)) # 裁剪 + return round(reward, 2) diff --git a/backend/xingbang/evolution/trajectory.py b/backend/xingbang/evolution/trajectory.py new file mode 100644 index 0000000..4d9d4d0 --- /dev/null +++ b/backend/xingbang/evolution/trajectory.py @@ -0,0 +1,110 @@ +"""Trajectory Recorder — 采集五智能体每次调用的完整轨迹。 + +轨迹格式 (trajectory): +{ + "trajectory_id": "traj_", + "customer": "客户名", + "workflow_id": "...", + "steps": [ + { + "agent": "data_curator", + "input": "任务描述", + "prompt_snippet": "...", + "tools": ["read_xingbang_data", ...], + "output_file": "...", + "output_preview": "...", + "success": true/false, + "time_ms": 1234 + }, ... + ], + "final_result": {...}, + "cumulative_reward": 0.0 +} + +轨迹保存在 /xingbang_evolution/trajectories/*.json。 +""" + +from __future__ import annotations + +import json +import time +import uuid +from pathlib import Path +from typing import Any + + +class TrajectoryRecorder: + """记录并持久化一次评审流水线的完整执行轨迹。""" + + def __init__(self, base_dir: Path, enabled: bool = True): + self._base_dir = Path(base_dir) + self._dir = self._base_dir / "xingbang_evolution" / "trajectories" + self._dir.mkdir(parents=True, exist_ok=True) + self.enabled = enabled + # 当前正在记录的轨迹 + self._current: dict[str, Any] | None = None + # metadata (customer 等) + self._meta: dict = {} + self._start_time: float = 0.0 + + def begin(self, workflow_id: str | None = None, customer: str = "") -> str: + """开始记录一条新轨迹,返回 trajectory_id。""" + self._start_time = time.time() + self._current = { + "trajectory_id": f"traj_{int(time.time())}_{uuid.uuid4().hex[:6]}", + "workflow_id": workflow_id or f"wf_{int(time.time())}", + "customer": customer or "", + "created_at": self._start_time, + "steps": [], + "final_result": {}, + "cumulative_reward": 0.0, + } + return self._current["trajectory_id"] + + def add_step(self, step: dict[str, Any]) -> None: + """追加一步智能体执行记录。""" + if self._current is not None: + self._current["steps"].append(step) + + def finalize(self, success: bool = True, summary: str = "") -> dict[str, Any]: + """结束记录,保存轨迹到磁盘,返回轨迹。""" + if self._current is None: + return {} + self._current["final_result"] = { + "success": success, + "summary": summary, + "duration_ms": int((time.time() - self._start_time) * 1000), + "total_steps": len(self._current["steps"]), + } + # 记录成功/失败 + trajectory = self._current + self._save(trajectory) + self._current = None + return trajectory + + def _save(self, trajectory: dict[str, Any]) -> Path: + path = self._dir / f"{trajectory['trajectory_id']}.json" + path.write_text( + json.dumps(trajectory, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + return path + + # ── 从调用侧采集 ──────────────────────────────────── + def on_evidence(self, kind: str, data: dict) -> None: + """订阅 agents 的 evidence 回调,累积成本步记录。""" + if not self.enabled or self._current is None: + return + if kind == "agent_start": + buffered = [] + self._current["_buffered"] = buffered + # 这一步先不展开, 由 workflow 显式调用 add_step 更为清晰 + + def list_trajectories(self) -> list[Path]: + return sorted(self._dir.glob("*.json")) + + def load(self, trajectory_id: str) -> dict[str, Any]: + path = self._dir / f"{trajectory_id}.json" + if path.exists(): + return json.loads(path.read_text(encoding="utf-8")) + return {} diff --git a/backend/xingbang/workflow.py b/backend/xingbang/workflow.py new file mode 100644 index 0000000..dbad315 --- /dev/null +++ b/backend/xingbang/workflow.py @@ -0,0 +1,229 @@ +"""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: 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")