diff --git a/AGENTS.md b/AGENTS.md index e0e4234..50fb9f8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,7 @@ alwaysApply: true --- --- -last_reviewed: 2026-07-02 +last_reviewed: 2026-10-19 --- # AGENTS — 项目操作指南 @@ -27,7 +27,7 @@ last_reviewed: 2026-07-02 | 安装依赖 + pre-commit | `make install` | `python tasks.py install` | | 代码检查（lint+format+typecheck+deptry） | `make check` | `python tasks.py check` | | 运行测试（含覆盖率报告） | `make test` | `python tasks.py test` | -| 运行 CLI 全流程 | `make run` | `python tasks.py run` | +| 运行 CLI 全流程（遗留，即将删除） | `make run` | `python tasks.py run` | | 清理缓存和虚拟环境 | `make clean` | `python tasks.py clean` | **注意：** `tasks.py` 中的 `check` 命令使用 `&&` 连接，Windows PowerShell 不支持 `&&`，但 `tasks.py` 内部已处理为单行字符串。 @@ -47,27 +47,37 @@ last_reviewed: 2026-07-02 **注意：** MyPy 和 deptry **不在** pre-commit 中，需要手动运行 `make check`。 -## 项目架构（Agent 调度模式） +## 项目架构（MCP 服务器模式） -核心入口：`src/agent/orchestrator.py` — Agent 是负责调度的中枢，协调以下模块： -- `extraction/extractor.py` — 文件扫描 → LLM 多模态提取 → JSON 缓存 -- `matching/matcher.py` — 支付记录与发票金额匹配 -- `validation/validator.py` — 声明式校验器（规则配置与引擎分离） -- `infra/browser/travel.py` / `normal.py` — 浏览器自动化填报 +**唯一业务入口是 MCP 服务器**：`python -m src.mcp.server`（stdio），工具名前缀 `mcp__arb__`。 +编排由宿主 Agent（DSH）完成，本项目只提供工具与确定性内核。 + +> 过渡期状态：`src/main.py`（CLI）与 `src/web`（Flask，已移除 Agent 端点，仅剩上传/表格编辑/下载/提交） +> 仍在工作，但**已确定删除**，不要把它们的代码当参考实现，也不要往里加功能。`.invoice_cache` 键名约定两边共用，那部分可以信。 + +- `src/mcp/` — 工具与协议层（`server.py` 注册与启动，`tools/` 按职能分模块，`schemas.py` 抽取字段定义，`envelope.py` 统一返回信封） +- `src/cases.py` — 会话目录 `cases//` 的唯一实现：材料、收件箱、产物与**阶段推导**（阶段由产物推导，不存在状态机文件） +- `src/core/` — 确定性内核：`extraction/`（缓存读写、PDF 渲染与文本层）、`matching/`（金额匹配 + `runner.py` 从缓存匹配）、`validation/`（声明式规则校验） +- `src/infra/` — 基础设施：`browser/`（Playwright 填报）、`documents/`（CSV/Word/Excel/PDF） +- `src/upload/` — 上传收件器（**独立进程** `python -m src.upload.server`）：手机扫码/PC 上传落 `cases//inbox/`，不含业务编排，绝不由 MCP 子进程拉起 + +已删除：`src/agent/`（手写状态机与 SSE 事件）。`required: true` 的字段（含值为空串的备注类字段） +必须存在于 info JSON 中——这是刻意约束；用 `mcp__arb__info_skeleton` 取骨架再填值，别凭记忆造结构。 +宿主侧调 MCP 工具时**路径一律用正斜杠**（`D:/Code/...`），反斜杠会被 DSH 的参数绑定层拒收。 ### 数据流关键产物 | 文件 | 生成阶段 | 作用 | |------|---------|------| -| `.invoice_cache/*.json` | extractor 提取 | 单张发票/支付记录的结构化数据 | -| `match_result.json` | matcher 匹配 | 支付截图与发票的关联关系 | -| `travel_info.json` / `normal_info.json` | LLM 综合提取 | 差旅/普通报销所需的全部结构化数据 | -| `invoice_summary.csv` | extractor 提取 | 普通发票汇总（用于生成易耗品出库单） | +| `.invoice_cache/.json` | 宿主 Agent 抽取后 `mcp__arb__case_save_extraction` 落盘 | 单张发票/支付记录的结构化数据 | +| `match_result.json` | `mcp__arb__match_payments` | 支付截图与发票的关联关系 | +| `travel_info.json` / `normal_info.json` | 宿主 Agent 综合，`mcp__arb__save_info` 落盘 | 差旅/普通报销所需的全部结构化数据 | +| `invoice_summary.csv` / `payment_records.csv` | `mcp__arb__match_payments` | 发票级与支付级汇总（出库单、申购单取数） | ### 缓存机制 -CLI 模式：`scripts/data/.invoice_cache/` -Web 模式：`src/web/uploads//.invoice_cache/` +MCP 会话：`cases//.invoice_cache/`（**新**） +遗留只读：`scripts/data/`（CLI）与 `src/web/uploads//`（Web），两者即将删除 缓存文件与源文件同名（如 `发票1.pdf` → `.invoice_cache/发票1.json`），后续步骤均从缓存读取。删除缓存后下次处理会重新提取。 @@ -78,7 +88,7 @@ Web 模式：`src/web/uploads//.invoice_cache/` * **敏感信息**：`scripts/config.json` 含登录凭据，勿提交到公开仓库 * **发票类型区分**：差旅发票（高铁票/酒店住宿）不生成易耗品出库单，走差旅报销流程；普通发票生成出库单 -## Web 服务 +## Web 服务（遗留，即将删除） ```bash uv run python src/web/app.py diff --git a/src/agent/README.md b/src/agent/README.md deleted file mode 100644 index b539a50..0000000 --- a/src/agent/README.md +++ /dev/null @@ -1,242 +0,0 @@ ---- - -## last_reviewed: 2026-06-13 - -# src/agent — Agent 协调模块 - -作为调度中枢，编排信息提取、规则校验和语义校验的完整流程，驱动用户完成材料补充直到信息完整可提交。 - -## 模块清单 - - -| 文件 | 作用 | -| ----------------- | ------------------------------------------ | -| `orchestrator.py` | 调度中枢：状态机管理、轮次调度、校验-修正循环编排、语义校验、用户补充处理、事件发射 | - - -## 架构概览 - -Agent 是报销系统的调度中枢，负责编排各模块完成信息提取和校验。它的核心职责： - -1. **调度 LLM 提取** — 调用 `doc.llm_extractor` 的纯提取接口 -2. **调度规则校验** — 调用 `doc.validator` 按报销规范检查字段完整性 -3. **编排校验-修正循环** — 校验失败时构建修正提示，再次调度 LLM 修正 -4. **调度语义校验** — 调用 LLM 判断提取信息在语义层面是否自洽、充分 -5. **状态持久化** — 每轮结束后将会话状态写入磁盘，支持中断恢复 - -```mermaid -graph TB - subgraph agent["src/agent (调度中枢)"] - O[orchestrator.py] - end - - subgraph doc["src/doc"] - LE[llm_extractor.py] - VA[validator.py] - end - - O -->|"1. 调度 LLM 提取"| LE - O -->|"2. 调度规则校验"| VA - VA -->|"校验失败"| O - O -->|"3. 构建修正提示"| LE - O -->|"4. 调度语义校验"| LE - LE -->|"缓存读写"| C[.invoice_cache/] - O -->|"SSE 事件"| E[agent_events.log] - O -->|"状态持久化"| S[agent_state.json] -``` - - - -### 校验-修正循环 - -Agent 编排的校验-修正循环流程： - -1. Agent 调用 `llm_extractor.llm_query_text()` 让 LLM 提取信息 -2. Agent 调用 `validator.validate_extracted_info()` 规则校验 -3. 校验失败则 Agent 构建修正提示（包含缺失字段列表），再次调用 LLM -4. 最多重试 3 次（`MAX_VALIDATION_RETRIES`），3 次后返回最佳结果 - -**关键点**：校验逻辑不内嵌在 `llm_extractor` 中，而是由 Agent 层调度。`llm_extractor` 只提供纯提取能力，`validator` 只提供纯校验能力，Agent 负责编排。 - -## 状态机 - -Agent 会话通过 `AgentState` 枚举管理生命周期： - -```mermaid -stateDiagram-v2 - [*] --> IDLE - IDLE --> EXTRACTING - EXTRACTING --> AWAITING_SUPPLEMENT - EXTRACTING --> READY - AWAITING_SUPPLEMENT --> EXTRACTING: 用户补充后重新校验 - READY --> [*] - EXTRACTING --> ERROR: 提取失败 - AWAITING_SUPPLEMENT --> ERROR: 超出最大轮次 - READY --> SUBMITTING: 用户提交 - SUBMITTING --> DONE - DONE --> [*] - AWAITING_SUPPLEMENT --> READY: 用户强制提交 -``` - - - -### 状态说明 - - -| 状态 | 含义 | 触发条件 | -| --------------------- | ------------------------ | ------------------------ | -| `IDLE` | 初始状态，等待启动 | 会话创建时 | -| `EXTRACTING` | 正在执行提取-校验-修正循环 | `run_agent_round()` 开始执行 | -| `AWAITING_SUPPLEMENT` | 语义校验未通过，等待用户补充材料或文字说明 | 语义校验失败 | -| `READY` | 规则校验 + 语义校验均通过，信息完整，可以提交 | 双重校验通过 | -| `SUBMITTING` | 用户确认提交，进入提交流程 | 调用提交接口 | -| `DONE` | 提交完成，终态 | 提交成功后 | -| `ERROR` | 提取失败或超出最大轮次（默认 5 轮） | 异常或轮次耗尽 | - - -## 数据流 - -### 单轮处理流程 - -```mermaid -flowchart TD - A[开始第 N 轮] --> B{终态保护?} - B -->|是| Z[跳过返回] - B -->|否| C{轮次超限?} - C -->|是| E[转入 ERROR] - C -->|否| D[EXTRACTING] - - D --> F{缓存命中?} - F -->|是| G[读取缓存数据] - F -->|否| H[Agent 调度校验-修正循环] - - H --> I[调用 LLM 提取] - I --> J[调用 validator 校验] - J --> K{校验通过?} - K -->|是| L[写入缓存] - K -->|否| M{重试次数<3?} - M -->|是| N[构建修正提示] - N --> I - M -->|否| L - - G --> O[语义校验 validate_semantic_completeness] - L --> O - - O --> P{语义通过?} - P -->|是| Q[READY] - P -->|否| R[发射 agent_request_supplement 事件] - R --> S[AWAITING_SUPPLEMENT] - Q --> T[发射 agent_ready 事件] - - S --> U[持久化状态] - T --> U - U --> V[返回 session] - E --> U -``` - - - -### 用户补充流程 - -```mermaid -flowchart TD - A[用户补充] --> B{补充方式} - B -->|文件上传| C[add_supplement 记录文件名] - B -->|文字输入| D[process_user_text_supplement] - B -->|强制提交| E[force_submit 跳过校验] - - D --> F[LLM 解析文字提取字段] - F --> G{有有效字段?} - G -->|是| H[合并到 extracted_info] - H --> I[更新缓存] - I --> J[触发新一轮 run_agent_round] - G -->|否| K[返回无变化提示] - E --> L[直接转入 READY] - - C --> M[等待下一轮 run_agent_round] - J --> M - L --> M - K --> M -``` - - - -## 核心数据结构 - -### AgentSession - -会话状态的完整载体，包含： - -```python -{ - "session_id": "str", # 会话唯一标识 - "state": "AgentState", # 当前状态 - "rounds": 0, # 已执行的轮次数 - "max_rounds": 5, # 最大轮次限制 - "invoice_type": "travel", # 发票类型: "travel" | "normal" - "extracted_info": {}, # 提取的结构化报销信息 - "validation_reports": [], # 历次语义校验报告列表 - "user_supplements": [], # 用户补充的文件名列表 - "error_message": "" # 错误信息 -} -``` - -### 校验报告 - -语义校验生成的报告条目追加到 `validation_reports`： - -```python -{ - "round": 1, - "type": "semantic", - "valid": False, - "issues": ["金额不一致"], - "missing_info": ["报销说明"], - "suggestion": "请确认金额一致性并补充报销说明", - "confidence": 0.6 -} -``` - -### SSE 事件 - -通过 `agent_events.log` 向外部发射实时事件，每行一条 JSON： - - -| 事件类型 | 触发时机 | -| --------------------------- | ------------- | -| `agent_state_change` | 状态切换时 | -| `agent_ready` | 双重校验通过，信息完整 | -| `agent_request_supplement` | 校验未通过，请求用户补充 | -| `agent_supplement_received` | 收到用户补充（文件或文字） | -| `agent_force_submit` | 用户选择强制提交 | -| `agent_error` | 信息提取失败 | -| `agent_max_rounds` | 达到最大轮次限制 | - - -## 持久化机制 - -- **状态文件**：`session_dir/agent_state.json` — 原子写入（先写 `.tmp` 再 rename） -- **事件日志**：`session_dir/agent_events.log` — 追加写入，支持前端 SSE 轮询 -- **缓存目录**：`session_dir/.invoice_cache/` — 存放 `travel_info.json` / `normal_info.json` - -## 依赖说明 - - -| 上游依赖 | 用途 | -| -------------------------- | ------------------------- | -| `src/doc/llm_extractor.py` | 纯 LLM 提取、语义校验、缓存加载、用户补充解析 | -| `src/doc/validator.py` | 纯规则校验 | -| `src/doc/prompt.py` | 系统提示词加载 | - - -Agent 是调度中枢，`llm_extractor` 提供纯提取能力，`validator` 提供纯校验能力，Agent 负责编排校验-修正循环。各模块职责清晰，不互相嵌套。 - -## 设计原则 - -- **Agent 是调度中枢**：校验-修正循环由 Agent 编排，不内嵌在 `llm_extractor` 中 -- **模块职责单一**：`llm_extractor` 只管提取，`validator` 只管校验，Agent 负责编排 -- **缓存优先**：信息提取优先读取 `.invoice_cache`，避免重复调用 LLM -- **轮次保护**：默认 5 轮上限，防止无限循环；校验-修正循环最多重试 3 次 -- **终态保护**：`DONE` / `SUBMITTING` / `READY` 状态下不再重复处理 -- **容错降级**：语义校验失败不阻断流程，仅记录警告日志；规则校验 3 次重试后返回最佳结果 - diff --git a/src/agent/__init__.py b/src/agent/__init__.py deleted file mode 100644 index 3c94dc4..0000000 --- a/src/agent/__init__.py +++ /dev/null @@ -1,32 +0,0 @@ -"""Agent 模块 - -提供多轮对话协调、规则校验和语义校验能力。 - -入口: - - `orchestrator.run_agent_round()` — 执行一轮 Agent 处理 - - `orchestrator.force_submit()` — 用户强制提交 - - `orchestrator.add_supplement()` — 记录用户补充的文件 - - `orchestrator.load_agent_state()` / `save_agent_state()` — 状态持久化 -""" - -from .orchestrator import ( - AgentSession, - AgentState, - add_supplement, - force_submit, - load_agent_state, - process_user_text_supplement, - run_agent_round, - save_agent_state, -) - -__all__ = [ - "AgentSession", - "AgentState", - "add_supplement", - "force_submit", - "load_agent_state", - "process_user_text_supplement", - "run_agent_round", - "save_agent_state", -] diff --git a/src/agent/coordinator.py b/src/agent/coordinator.py deleted file mode 100644 index 1e37387..0000000 --- a/src/agent/coordinator.py +++ /dev/null @@ -1,382 +0,0 @@ -"""Agent 协调器 - -作为调度中枢，编排信息提取、规则校验的完整流程。 - -校验-修正循环由 Agent 层调度： - 1. Agent 调用 LLM 提取信息 - 2. Agent 调用 validator.py 校验 - 3. 校验失败则构建修正提示，再次调用 LLM - 4. 重复直到校验通过或达到最大重试次数 - 5. LLM 在输出中包含 can_submit 和 suggestion 字段，用于判断信息完整性 -""" - -from __future__ import annotations - -from pathlib import Path -from typing import Any - -from .. import get_logger -from ..core.extraction import ( - build_extraction_user_message, - llm_query_text, - load_cache, - load_match_result, - parse_json_response, -) -from ..core.validation import validate_extracted_info -from ..infra.llm import ( - build_normal_info_system_prompt, - build_travel_info_system_prompt, -) -from ..pipeline_core import save_cache_info -from .events import emit_agent_event -from .session import AgentSession, AgentState, save_agent_state - -log = get_logger("agent.coordinator") - -# 规则校验-修正循环的最大重试次数 -MAX_VALIDATION_RETRIES = 3 - - -# ------------------------------------------------------------------ -# 辅助：构建修正提示 -# ------------------------------------------------------------------ - - -def _build_correction_prompt( - base_message: str, - report: Any, -) -> str: - """根据校验报告构建修正提示，追加到原始用户消息后。""" - error_feedback = ( - f"\n\n=== 上一次输出的校验结果 ===\n" - f"校验未通过，发现以下问题：\n" - f"缺失字段 ({len(report.missing_fields)} 个)：{', '.join(report.missing_fields)}\n" - ) - if report.missing_materials: - error_feedback += f"可能需要补充的材料：{', '.join(report.missing_materials)}\n" - if report.suggestion: - error_feedback += f"建议：{report.suggestion}\n" - error_feedback += ( - "\n请根据以上校验结果修正你的输出，确保所有必填字段都有值。" - "如果某个字段确实没有数据，请给出合理的猜测值。" - "再次返回完整的 JSON 结果。" - ) - return base_message + error_feedback - - -# ------------------------------------------------------------------ -# 核心协调逻辑 -# ------------------------------------------------------------------ - - -def _do_extraction_with_validation( - session_dir: Path, - session: AgentSession, - previous_analysis: dict[str, Any] | None = None, - cache_map: dict[str, Any] | None = None, - user_supplement: str | None = None, -) -> dict[str, Any]: - """Agent 调度的提取-校验-修正循环。 - - 流程： - 1. 加载缓存数据和匹配结果 - 2. 构建用户消息 - 3. 调用 LLM 提取 - 4. 调用 validator 校验 - 5. 校验失败则构建修正提示，回到步骤 3 - 6. 最多重试 MAX_VALIDATION_RETRIES 次 - - LLM 输出的 JSON 中额外包含 can_submit 和 suggestion 字段， - 用于判断信息是否完整可提交。 - - Args: - session_dir: 会话目录。 - session: 当前 Agent 会话。 - previous_analysis: 上一轮 LLM 分析结果（可选，补充文件时传入作为历史上下文）。 - user_supplement: 用户文字补充（可选，传入时作为补充上下文）。 - - Returns: - 校验通过的结构化数据（或达到重试上限后的最佳结果）。 - """ - cache_map = cache_map or load_cache(session_dir) - match_result = load_match_result(session_dir) - - # 选择系统提示词 - if session.invoice_type == "travel": - system_prompt = build_travel_info_system_prompt() - else: - system_prompt = build_normal_info_system_prompt() - - base_message = build_extraction_user_message( - cache_map, - match_result, - previous_analysis=previous_analysis, - user_supplement=user_supplement, - ) - current_message = base_message - - for attempt in range(1, MAX_VALIDATION_RETRIES + 1): - log.info( - "LLM 提取第 %d/%d 次尝试 (%s)", - attempt, - MAX_VALIDATION_RETRIES, - session.invoice_type, - ) - emit_agent_event( - session_dir, - "agent_state_change", - state=AgentState.EXTRACTING, - round=session.rounds, - attempt=attempt, - message=f"正在分析文件... (第{attempt}次)", - ) - - # Step 1: 调用 LLM 提取 - try: - response = llm_query_text( - system_prompt=system_prompt, - text=current_message, - reasoning_effort="low", - source_dir=session_dir, - ) - result = parse_json_response(response) - except Exception as e: - log.error("LLM 提取失败: %s", e) - emit_agent_event( - session_dir, - "agent_error", - message=f"LLM 提取失败: {e}", - ) - raise - - # Step 2: 调用 validator 校验 - report = validate_extracted_info(result, invoice_type=session.invoice_type) - - if report.valid: - log.info("规则校验通过 (第 %d 次尝试)", attempt) - emit_agent_event( - session_dir, - "agent_extract_status", - state=AgentState.EXTRACTING, - round=session.rounds, - attempt=attempt, - message=f"规则校验通过 (第{attempt}次)", - ) - return result - - # Step 3: 校验失败，构建修正提示 - log.warning( - "规则校验未通过 (第 %d/%d 次): 缺失 %d 个字段 - %s", - attempt, - MAX_VALIDATION_RETRIES, - len(report.missing_fields), - report.missing_fields, - ) - emit_agent_event( - session_dir, - "agent_extract_status", - state=AgentState.EXTRACTING, - round=session.rounds, - attempt=attempt, - message=f"规则校验未通过，缺失 {len(report.missing_fields)} 个字段，正在请求 LLM 修正...", - ) - current_message = _build_correction_prompt(current_message, report) - - # 所有重试都失败，返回最后一次结果 - log.error( - "LLM 提取经过 %d 次尝试仍未通过规则校验，返回最后一次结果 (置信度: %.0f%%)", - MAX_VALIDATION_RETRIES, - report.confidence * 100, - ) - return result - - -def run_agent_round( - session_dir: Path, - session: AgentSession, - new_files: list[str] | None = None, - user_supplement: str | None = None, -) -> AgentSession: - """执行一轮 Agent 处理：提取-校验-修正循环。 - - Args: - session_dir: 会话目录。 - session: 当前 Agent 会话。 - new_files: 新增的文件列表（可选，补充文件时传入）。 - user_supplement: 用户文字补充（可选，传入时作为补充上下文，强制重新分析）。 - - Returns: - 更新后的 Agent 会话。 - - 注意： - - 信息提取优先从 .invoice_cache 缓存读取，避免重复调用 LLM。 - - Agent 调度提取-校验-修正循环：LLM 提取 -> validator 校验 -> 失败则反馈修正。 - - 若缓存缺失则执行提取后立即写回缓存（travel_info.json / normal_info.json）。 - - 补充文件时（new_files 非空），加载上一轮分析结果作为历史上下文，强制重新分析。 - """ - # 终态保护：会话已提交或已完成时不再重复处理 - if session.is_terminal(): - log.info("Agent 会话已处于终态 (%s)，跳过重复处理", session.state.value) - return session - - if session.rounds >= session.max_rounds: - session.state = AgentState.ERROR - session.error_message = f"已达到最大轮次 ({session.max_rounds})，请检查信息或强制提交" - log.warning("Agent 达到最大轮次限制") - emit_agent_event( - session_dir, - "agent_max_rounds", - message=session.error_message, - ) - return session - - session.rounds += 1 - log.info("开始第 %d 轮 Agent 处理", session.rounds) - - # ---- Step 1: 信息提取（Agent 调度校验-修正循环） ---- - session.state = AgentState.EXTRACTING - - # 判断是否为补充场景：文件补充或文字补充，加载上一轮分析结果作为上下文 - cache_map = load_cache(session_dir) - is_supplement = bool(new_files) or bool(user_supplement) - previous_analysis = None - if is_supplement: - info_key = "travel_info" if session.invoice_type == "travel" else "normal_info" - previous_analysis = cache_map.get(info_key) - if previous_analysis: - log.info("检测到补充内容，加载上一轮分析结果作为历史上下文") - - try: - info_key = "travel_info" if session.invoice_type == "travel" else "normal_info" - should_reanalyze = not cache_map.get(info_key) or is_supplement - if should_reanalyze: - session.extracted_info = _do_extraction_with_validation( - session_dir, - session, - previous_analysis=previous_analysis, - cache_map=cache_map, - user_supplement=user_supplement, - ) - # 提取后立即写入缓存，后续步骤依赖此数据 - save_cache_info(session_dir, info_key, session.extracted_info) - else: - session.extracted_info = cache_map[info_key] - emit_agent_event( - session_dir, - "agent_state_change", - state=session.state, - round=session.rounds, - message="使用缓存数据，无需重新分析", - ) - - except Exception as e: - session.state = AgentState.ERROR - session.error_message = f"信息提取失败: {e}" - log.error("Agent 信息提取失败: %s", e) - emit_agent_event( - session_dir, - "agent_error", - message=session.error_message, - ) - return session - - # ---- 判断结果（从 LLM 提取结果中读 can_submit） ---- - can_submit = session.extracted_info.get("can_submit", True) - suggestion = session.extracted_info.get("suggestion", "") - - if can_submit: - session.state = AgentState.READY - emit_agent_event( - session_dir, - "agent_ready", - round=session.rounds, - message="信息完整，可以提交", - ) - log.info("Agent 校验通过，信息完整") - else: - session.state = AgentState.AWAITING_SUPPLEMENT - - combined_suggestion = suggestion or "信息不完整，请补充材料" - - emit_agent_event( - session_dir, - "agent_request_supplement", - round=session.rounds, - missing_fields=[], - missing_materials=[], - semantic_issues=[], - suggestion=combined_suggestion, - ) - log.info("Agent 请求补充: %s", combined_suggestion) - - save_agent_state(session_dir, session) - return session - - -def force_submit( - session_dir: Path, - session: AgentSession, -) -> AgentSession: - """用户强制提交，跳过校验。""" - session.state = AgentState.READY - log.info("用户强制提交，跳过校验") - emit_agent_event( - session_dir, - "agent_force_submit", - message="用户选择强制提交", - ) - save_agent_state(session_dir, session) - return session - - -def add_supplement( - session_dir: Path, - session: AgentSession, - filenames: list[str], -) -> AgentSession: - """记录用户补充的文件。""" - session.user_supplements.extend(filenames) - emit_agent_event( - session_dir, - "agent_supplement_received", - files=filenames, - ) - log.info("收到用户补充文件: %s", filenames) - save_agent_state(session_dir, session) - return session - - -def process_user_text_supplement( - session_dir: Path, - session: AgentSession, - user_text: str, -) -> AgentSession: - """处理用户通过文字补充的信息。 - - 流程： - 1. LLM 分析用户文字，提取需要更新的字段 - 2. 合并到已提取的信息中 - 3. 保存到缓存 - 4. 重新执行一轮 Agent 校验 - - Args: - session_dir: 会话目录。 - session: 当前 Agent 会话。 - user_text: 用户输入的文字。 - - Returns: - 更新后的 Agent 会话。 - """ - log.info("收到用户文字补充: %s", user_text) - emit_agent_event( - session_dir, - "agent_supplement_received", - supplement_type="text", - text=user_text[:50], - ) - - # 将用户文字补充作为上下文传给主提取流程，LLM 输出完整 JSON - session = run_agent_round(session_dir, session, user_supplement=user_text) - - return session diff --git a/src/agent/events.py b/src/agent/events.py deleted file mode 100644 index 2efc12f..0000000 --- a/src/agent/events.py +++ /dev/null @@ -1,75 +0,0 @@ -"""Agent 事件系统 - -负责 SSE 事件的发射和管理，用于实时通知前端状态变化。 -""" - -from __future__ import annotations - -import json -from pathlib import Path -from typing import Any - -from .. import get_logger - -log = get_logger("agent.events") - -AGENT_EVENT_LOG = "agent_events.log" - -# 去重守卫：记录每个 session 上一次发射的事件类型，防止连续重复发射 -# key: str(session_dir), value: 上一次的 event_type -_last_event_type: dict[str, str] = {} - - -def emit_agent_event(session_dir: Path, event_type: str, **kwargs: Any) -> None: - """向 agent_events.log 追加一行 JSON 事件。 - - 同一 session 连续发射相同 event_type 时直接抛出 RuntimeError， - 强制调用方修复重复发射的代码，而非静默掩盖。 - - 注意：agent_state_change 会在同一轮提取中多次发射不同消息 - （"正在分析"、"校验通过"、"校验失败，请求修正"），这是合法行为。 - 去重守卫仅检查 event_type 字符串是否完全相同，不检查 kwargs。 - 因此不要在同一个 event_type 下连续发射不同消息，应使用不同的事件类型。 - """ - session_key = str(session_dir) - prev = _last_event_type.get(session_key) - if prev == event_type: - raise RuntimeError( - f"事件重复发射: session={session_dir.name!r}, event_type={event_type!r}。" - f"请检查调用链，确保每个事件类型只发射一次。" - ) - _last_event_type[session_key] = event_type - - event = {"type": event_type, **kwargs} - try: - event_path = session_dir / AGENT_EVENT_LOG - with open(event_path, "a", encoding="utf-8") as f: - f.write(json.dumps(event, ensure_ascii=False) + "\n") - except Exception: - pass - - -def clear_event_history(session_dir: Path) -> None: - """清除指定会话的事件历史。""" - session_key = str(session_dir) - _last_event_type.pop(session_key, None) - event_path = session_dir / AGENT_EVENT_LOG - if event_path.exists(): - event_path.unlink() - - -def read_events(session_dir: Path) -> list[dict[str, Any]]: - """读取指定会话的所有事件记录。""" - event_path = session_dir / AGENT_EVENT_LOG - if not event_path.exists(): - return [] - events = [] - try: - with open(event_path, encoding="utf-8") as f: - for line in f: - line = line.strip() - if line: - events.append(json.loads(line)) - except Exception as e: - log.warning("读取事件日志失败: %s", e) - return events diff --git a/src/agent/orchestrator.py b/src/agent/orchestrator.py deleted file mode 100644 index ee0346e..0000000 --- a/src/agent/orchestrator.py +++ /dev/null @@ -1,46 +0,0 @@ -"""Agent 协调器（兼容层） - -此文件为向后兼容而保留，所有功能已迁移到以下子模块： -- session.py: 会话状态定义和持久化 -- events.py: SSE 事件发射系统 -- coordinator.py: 核心协调逻辑 - -原有导入路径保持可用。 -""" - -# 从新模块重新导出所有符号，保持向后兼容 -from .coordinator import ( - add_supplement, - force_submit, - process_user_text_supplement, - run_agent_round, -) -from .events import emit_agent_event as _emit_agent_event -from .session import ( - AgentSession, - AgentState, - load_agent_state, - save_agent_state, -) - -# 为旧代码提供兼容的私有函数 -_emit_agent_event = _emit_agent_event - -# 常量保持不变 -MAX_VALIDATION_RETRIES = 3 -AGENT_STATE_FILE = "agent_state.json" -AGENT_EVENT_LOG = "agent_events.log" - -__all__ = [ - "AgentSession", - "AgentState", - "save_agent_state", - "load_agent_state", - "run_agent_round", - "force_submit", - "add_supplement", - "process_user_text_supplement", - "MAX_VALIDATION_RETRIES", - "AGENT_STATE_FILE", - "AGENT_EVENT_LOG", -] diff --git a/src/agent/session.py b/src/agent/session.py deleted file mode 100644 index ee79c29..0000000 --- a/src/agent/session.py +++ /dev/null @@ -1,101 +0,0 @@ -"""Agent 会话管理 - -负责会话状态的定义、序列化和持久化。 -""" - -from __future__ import annotations - -import json -from dataclasses import asdict, dataclass, field -from enum import StrEnum -from pathlib import Path -from typing import Any - -from .. import get_logger - -log = get_logger("agent.session") - -# ------------------------------------------------------------------ -# 状态枚举 -# ------------------------------------------------------------------ - - -class AgentState(StrEnum): - IDLE = "idle" - EXTRACTING = "extracting" - AWAITING_SUPPLEMENT = "awaiting_supplement" - READY = "ready" - SUBMITTING = "submitting" - DONE = "done" - ERROR = "error" - - -# ------------------------------------------------------------------ -# Agent 会话数据模型 -# ------------------------------------------------------------------ - - -@dataclass -class AgentSession: - """Agent 会话状态""" - - session_id: str - state: AgentState = AgentState.IDLE - rounds: int = 0 - max_rounds: int = 5 - invoice_type: str = "travel" # "travel" 或 "normal" - extracted_info: dict[str, Any] = field(default_factory=dict) - validation_reports: list[dict[str, Any]] = field(default_factory=list) - user_supplements: list[str] = field(default_factory=list) - error_message: str = "" - - def to_dict(self) -> dict[str, Any]: - return asdict(self) - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> AgentSession: - # 兼容旧版本：state 可能是字符串 - if "state" in data and isinstance(data["state"], str): - data["state"] = AgentState(data["state"]) - return cls(**data) - - def is_terminal(self) -> bool: - """判断会话是否处于终态（已提交或已完成）""" - return self.state in (AgentState.DONE, AgentState.SUBMITTING, AgentState.READY) - - -# ------------------------------------------------------------------ -# 持久化 -# ------------------------------------------------------------------ - -AGENT_STATE_FILE = "agent_state.json" - - -def save_agent_state(session_dir: Path, session: AgentSession) -> None: - """将 Agent 会话状态持久化到 session 目录。""" - state_path = session_dir / AGENT_STATE_FILE - tmp_path = session_dir / (AGENT_STATE_FILE + ".tmp") - with open(tmp_path, "w", encoding="utf-8") as f: - json.dump(session.to_dict(), f, ensure_ascii=False, indent=2) - tmp_path.replace(state_path) - - -def load_agent_state(session_dir: Path) -> AgentSession | None: - """从 session 目录加载 Agent 会话状态。""" - state_path = session_dir / AGENT_STATE_FILE - if not state_path.exists(): - return None - try: - with open(state_path, encoding="utf-8") as f: - return AgentSession.from_dict(json.load(f)) - except Exception as e: - log.warning("加载 Agent 状态失败: %s", e) - return None - - -def create_agent_session(session_id: str, invoice_type: str = "travel") -> AgentSession: - """创建新的 Agent 会话。""" - return AgentSession( - session_id=session_id, - invoice_type=invoice_type, - ) diff --git a/src/infra/documents/requisition.py b/src/infra/documents/requisition.py index 2654d96..9a47651 100644 --- a/src/infra/documents/requisition.py +++ b/src/infra/documents/requisition.py @@ -9,7 +9,8 @@ - 物品名称去除 "*分类*" 前缀（如 "*半导体*半导体分立器件" -> "半导体分立器件"） - 规格：优先用提取的型号；无型号或为泛化名称（集成电路/电阻/连接器等）时填"多品种" - 数量固定 1，单价(含税)=金额=发票价税合计 -- 对应实验实训项目：默认取 agent_state.json 的 reimbursement_description +- 对应实验实训项目：默认取 .invoice_cache/normal_info.json 的 reimbursement_description + （旧 Web 会话目录回退 agent_state.json） 用法: uv run python -m src.infra.documents.requisition --session src/web/uploads/35c63986686b @@ -32,7 +33,8 @@ log = get_logger("fill_requisition") REQUISITION_FILENAME = "低值易耗品申购.xls" CACHE_DIR_NAME = ".invoice_cache" -AGENT_STATE_FILENAME = "agent_state.json" +AGENT_STATE_FILENAME = "agent_state.json" # 旧 Web 会话遗留，仅作回退 +INFO_FILENAMES = ("normal_info.json", "travel_info.json") # MCP 流程产物，优先 TITLE_ROW = 1 HEADER_ROW = 2 @@ -128,16 +130,35 @@ def load_requisition_items( return items -def load_project_desc(session_dir: str | Path, fallback: str = DEFAULT_PROJECT) -> str: - """从 agent_state.json 读取报销描述作为实验实训项目，缺失时用回退值。""" - state_path = Path(session_dir) / AGENT_STATE_FILENAME +def _read_desc(path: Path, nested: bool) -> str: + """从 JSON 里取 basic_info.reimbursement_description；nested=True 时先下钻 extracted_info。""" + if not path.exists(): + return "" try: - with open(state_path, encoding="utf-8") as f: - state = json.load(f) - desc = state.get("extracted_info", {}).get("basic_info", {}).get("reimbursement_description") - return desc or fallback - except (OSError, ValueError): - return fallback + state = json.loads(path.read_text(encoding="utf-8")) + except Exception as e: + log.warning("读取 %s 失败: %s", path.name, e) + return "" + node = state.get("extracted_info", {}) if nested else state + if not isinstance(node, dict): + return "" + basic = node.get("basic_info", {}) + desc = basic.get("reimbursement_description") if isinstance(basic, dict) else None + return str(desc).strip() if desc else "" + + +def load_project_desc(session_dir: str | Path, fallback: str = DEFAULT_PROJECT) -> str: + """取报销事由作为「对应实验实训项目」。 + + 优先 .invoice_cache/normal_info.json 与 travel_info.json（MCP 流程产物）， + 再回退历史 agent_state.json（旧 Web 会话目录），最后用默认值。 + """ + root_dir = Path(session_dir) + for name in INFO_FILENAMES: + desc = _read_desc(root_dir / CACHE_DIR_NAME / name, nested=False) + if desc: + return desc + return _read_desc(root_dir / AGENT_STATE_FILENAME, nested=True) or fallback def fill_requisition_excel( @@ -236,10 +257,10 @@ def generate_requisition( def main() -> None: parser = argparse.ArgumentParser(description="生成低值易耗品申购单 Excel") - parser.add_argument("--session", default=".", help="Web 会话目录（含 .invoice_cache/ 与 agent_state.json）") + parser.add_argument("--session", default=".", help="会话目录（含 .invoice_cache/）") parser.add_argument("--template", default=None, help="申购单模板 xls 路径（默认取项目根目录现有模板）") parser.add_argument("--output", default=None, help="输出 xls 路径（默认同目录带时间戳新文件）") - parser.add_argument("--project", default=None, help="对应实验实训项目（默认读 agent_state.json）") + parser.add_argument("--project", default=None, help="对应实验实训项目（默认读 normal_info.json）") parser.add_argument("--exclude", default="", help="排除物品名称含这些关键词的发票，逗号分隔（如: 千斤顶,台钻）") args = parser.parse_args() diff --git a/src/web/routes.py b/src/web/routes.py index 8d6ee56..34fe57e 100644 --- a/src/web/routes.py +++ b/src/web/routes.py @@ -25,7 +25,6 @@ from src.config import ( load_config as load_project_config, ) from src.infra.documents import load_csv, load_invoice_csv -from src.pipeline_core import save_invoice_groups from . import pipeline_web, sse_handler @@ -43,9 +42,6 @@ def init_routes(upload_base: Path) -> None: # ================================================================ # 辅助函数 -# ================================================================ - - def _validate_session(session_id: str) -> Path | tuple[Response, int]: """验证会话 ID 并返回会话目录""" if _UPLOAD_BASE is None: @@ -68,94 +64,34 @@ def _build_web_config(body: dict[str, Any]) -> dict[str, Any]: return config -def _emit_ready_and_submit( - session_dir: Path, - agent_session: Any, - config: dict[str, Any], -) -> dict[str, Any]: - """Agent 校验通过后，直接触发财务提交（不依赖前端）。 - - 返回 result 字典，由调用方 _run_agent_task 的 finally 块统一写入 result.json。 - """ - # agent_ready 事件已由 run_agent_round 发射，此处仅执行财务提交，避免前端收到重复消息 - try: - submit_result = pipeline_web.run_financial_submit(session_dir, config) - if submit_result.get("ok"): - return { - "ok": True, - "agent_ready": True, - "submit_ok": True, - "round": agent_session.rounds, - "message": "信息完整，已自动提交到财务系统", - } - return { - "ok": True, - "agent_ready": True, - "submit_ok": False, - "submit_error": submit_result.get("error"), - "round": agent_session.rounds, - "message": "校验通过但提交失败", - } - except Exception as e: - return { - "ok": True, - "agent_ready": True, - "submit_ok": False, - "submit_error": str(e), - "round": agent_session.rounds, - "message": f"校验通过但提交异常: {e}", - } - - -def _run_agent_task( +def _run_session_task( session_dir: Path, handler: Any, task_fn: Any, config: dict[str, Any] | None = None, ) -> None: - """Agent 任务的通用包装器 + """后台任务的通用包装器 - 封装重复的闭包结构：清除流日志 -> 执行任务 -> 写入 result.json -> 卸载收集器。 + 执行任务并把结果写进 result.json，供 SSE 端点收尾。 Args: session_dir: 会话目录 handler: SSE 日志收集器句柄 - task_fn: 业务逻辑回调，接收 (session_dir, config) 返回 (agent_session, result_dict) 或仅 result_dict + task_fn: 业务逻辑回调，接收 (session_dir, config) 返回 result_dict config: 配置字典（可选） """ result = {"ok": False, "error": "未知错误"} try: # 清除上一轮残留文件，避免 SSE 连接立即读到旧数据 - for fname in ("llm_stream.log", "agent_events.log", pipeline_web.SESSION_RESULT_FILE): + for fname in ("llm_stream.log", pipeline_web.SESSION_RESULT_FILE): try: (session_dir / fname).unlink(missing_ok=True) except Exception: pass ret = task_fn(session_dir, config) - - # task_fn 返回 (agent_session, result) 或仅 result - if isinstance(ret, tuple) and len(ret) == 2: - from src.agent import AgentState - - agent_session, result = ret - - if agent_session.state == AgentState.READY: - if config is not None: - result = _emit_ready_and_submit(session_dir, agent_session, config) - elif agent_session.state == AgentState.AWAITING_SUPPLEMENT: - result = { - "ok": True, - "agent_ready": False, - "agent_state": agent_session.state.value, - "round": agent_session.rounds, - "waiting_for_supplement": True, - } - elif agent_session.state == AgentState.ERROR: - result = { - "ok": False, - "error": agent_session.error_message, - } + if isinstance(ret, dict): + result = ret except Exception as e: result = {"ok": False, "error": str(e)} @@ -464,7 +400,7 @@ def start_process(session_id: str) -> Any: handler = sse_handler.install_log_collector(session_dir) threading.Thread( - target=_run_agent_task, + target=_run_session_task, kwargs={"session_dir": session_dir, "handler": handler, "task_fn": _task, "config": config}, daemon=True, ).start() @@ -484,7 +420,6 @@ def stream_logs(session_id: str) -> Any: file_events_file = session_dir / "file_events.log" last_events_size = 0 last_stream_size = 0 - state = {"agent_size": 0} start_time = time.time() timeout = 900 # 略大于 LLM 请求超时 (600s)，防止 SSE 先断开 @@ -530,22 +465,6 @@ def stream_logs(session_id: str) -> Any: yield f"data: {line}\n\n" last_stream_size = stream_size - # 轮询 Agent 事件 - agent_events_file = session_dir / "agent_events.log" - last_agent_events_size = state["agent_size"] - if agent_events_file.exists(): - agent_size = agent_events_file.stat().st_size - if agent_size > last_agent_events_size: - with open(agent_events_file, encoding="utf-8", errors="replace") as f: - f.seek(last_agent_events_size) - new_agent_events = f.read() - if new_agent_events: - for line in new_agent_events.strip().split("\n"): - line = line.strip() - if line: - yield f"data: {line}\n\n" - state["agent_size"] = agent_size - # 检查是否完成 result_file = session_dir / pipeline_web.SESSION_RESULT_FILE if result_file.exists(): @@ -585,238 +504,7 @@ def submit_financial(session_id: str) -> Any: handler = sse_handler.install_log_collector(session_dir) threading.Thread( - target=_run_agent_task, - kwargs={"session_dir": session_dir, "handler": handler, "task_fn": _task, "config": config}, - daemon=True, - ).start() - return jsonify({"status": "started"}) - - -# ================================================================ -# Agent 交互 API -# ================================================================ - - -@web_bp.route("/api/agent/state/", methods=["GET"]) -def get_agent_state(session_id: str) -> Any: - """获取 Agent 会话状态""" - session_dir = _validate_session(session_id) - if isinstance(session_dir, tuple): - return session_dir - - from src.agent import load_agent_state - - session = load_agent_state(session_dir) - if session is None: - return jsonify({"error": "未找到 Agent 状态，请先处理"}), 404 - - return jsonify(session.to_dict()) - - -@web_bp.route("/api/agent/process/", methods=["POST"]) -def agent_process(session_id: str) -> Any: - """启动 Agent 多轮处理流程""" - session_dir = _validate_session(session_id) - if isinstance(session_dir, tuple): - return session_dir - - body = request.get_json(silent=True) or {} - config = _build_web_config(body) - - with open(session_dir / "config.json", "w", encoding="utf-8") as f: - json.dump(config, f, ensure_ascii=False, indent=2, default=str) - - handler = sse_handler.install_log_collector(session_dir) - - def _task(sd: Path, cfg: dict[str, Any] | None) -> tuple[Any, dict[str, Any]]: - from src.agent import ( - AgentSession, - load_agent_state, - run_agent_round, - save_agent_state, - ) - from src.core.extraction import extract_invoices - from src.infra.documents import ( - save_application_json, - save_invoice_csv, - ) - from src.infra.documents import ( - save_csv as save_payment_csv, - ) - - agent_session = load_agent_state(sd) - if agent_session is None: - invoices, applications, groups = extract_invoices(str(sd)) - if not invoices: - return (None, {"ok": False, "error": "未提取到任何发票数据"}) - - save_payment_csv(invoices, sd / "payment_records.csv") - save_invoice_csv(invoices, sd / "invoice_summary.csv") - - if applications: - save_application_json(applications, sd / "travel_applications.json") - - save_invoice_groups(sd, groups) - - from src.pipeline_core import is_travel_invoice - - agent_session = AgentSession( - session_id=session_id, - invoice_type="travel" if is_travel_invoice(groups) else "normal", - ) - save_agent_state(sd, agent_session) - - agent_session = run_agent_round(sd, agent_session) - return (agent_session, {}) - - threading.Thread( - target=_run_agent_task, - kwargs={"session_dir": session_dir, "handler": handler, "task_fn": _task, "config": config}, - daemon=True, - ).start() - return jsonify({"status": "started"}) - - -@web_bp.route("/api/agent/supplement/", methods=["POST"]) -def agent_supplement(session_id: str) -> Any: - """用户补充文件后触发新一轮 Agent 处理""" - session_dir = _validate_session(session_id) - if isinstance(session_dir, tuple): - return session_dir - - body = request.get_json(silent=True) or {} - filenames = body.get("files", []) - - if not filenames: - return jsonify({"error": "未指定补充文件"}), 400 - - from src.agent import ( - add_supplement, - load_agent_state, - run_agent_round, - ) - - agent_session = load_agent_state(session_dir) - if agent_session is None: - return jsonify({"error": "未找到 Agent 状态"}), 404 - - agent_session = add_supplement(session_dir, agent_session, filenames) - - # 重新提取发票并保存 - from src.core.extraction import extract_invoices - from src.infra.documents import ( - save_application_json, - save_invoice_csv, - ) - from src.infra.documents import ( - save_csv as save_payment_csv, - ) - - invoices, applications, groups = extract_invoices(str(session_dir)) - save_payment_csv(invoices, session_dir / "payment_records.csv") - save_invoice_csv(invoices, session_dir / "invoice_summary.csv") - - if applications: - save_application_json(applications, session_dir / "travel_applications.json") - - save_invoice_groups(session_dir, groups) - - config_path = session_dir / "config.json" - config = {} - if config_path.exists(): - with open(config_path, encoding="utf-8") as f: - config = json.load(f) - - handler = sse_handler.install_log_collector(session_dir) - - def _task(sd: Path, cfg: dict[str, Any] | None) -> tuple[Any, dict[str, Any]]: - new_session = run_agent_round(sd, agent_session, new_files=filenames) - return (new_session, {}) - - threading.Thread( - target=_run_agent_task, - kwargs={"session_dir": session_dir, "handler": handler, "task_fn": _task, "config": config}, - daemon=True, - ).start() - return jsonify({"status": "started"}) - - -@web_bp.route("/api/agent/user-supplement/", methods=["POST"]) -def agent_user_supplement(session_id: str) -> Any: - """用户通过文字补充信息，LLM 分析后更新 JSON，重新校验""" - session_dir = _validate_session(session_id) - if isinstance(session_dir, tuple): - return session_dir - - body = request.get_json(silent=True) or {} - user_text = body.get("text", "").strip() - - if not user_text: - return jsonify({"error": "请输入补充信息"}), 400 - - from src.agent import ( - load_agent_state, - process_user_text_supplement, - ) - - agent_session = load_agent_state(session_dir) - if agent_session is None: - return jsonify({"error": "未找到 Agent 状态"}), 404 - - config_path = session_dir / "config.json" - config = {} - if config_path.exists(): - with open(config_path, encoding="utf-8") as f: - config = json.load(f) - - handler = sse_handler.install_log_collector(session_dir) - - def _task(sd: Path, cfg: dict[str, Any] | None) -> tuple[Any, dict[str, Any]]: - new_session = process_user_text_supplement(sd, agent_session, user_text) - return (new_session, {}) - - threading.Thread( - target=_run_agent_task, - kwargs={"session_dir": session_dir, "handler": handler, "task_fn": _task, "config": config}, - daemon=True, - ).start() - return jsonify({"status": "started"}) - - -@web_bp.route("/api/agent/force-submit/", methods=["POST"]) -def agent_force_submit(session_id: str) -> Any: - """用户强制提交，跳过校验""" - session_dir = _validate_session(session_id) - if isinstance(session_dir, tuple): - return session_dir - - from src.agent import ( - force_submit, - load_agent_state, - ) - - agent_session = load_agent_state(session_dir) - if agent_session is None: - return jsonify({"error": "未找到 Agent 状态"}), 404 - - agent_session = force_submit(session_dir, agent_session) - - config_path = session_dir / "config.json" - if not config_path.exists(): - return jsonify({"error": "未找到配置"}), 400 - - with open(config_path, encoding="utf-8") as f: - config = json.load(f) - - def _task(sd: Path, cfg: dict[str, Any] | None) -> dict[str, Any]: - submit_result = pipeline_web.run_financial_submit(sd, cfg) - if submit_result.get("ok"): - return {"ok": True} - return submit_result - - handler = sse_handler.install_log_collector(session_dir) - threading.Thread( - target=_run_agent_task, + target=_run_session_task, kwargs={"session_dir": session_dir, "handler": handler, "task_fn": _task, "config": config}, daemon=True, ).start() diff --git a/tests/test_requisition.py b/tests/test_requisition.py index 9040041..c5c3a3b 100644 --- a/tests/test_requisition.py +++ b/tests/test_requisition.py @@ -130,6 +130,35 @@ class TestLoadRequisitionItems: class TestLoadProjectDesc: + def test_prefers_normal_info_over_legacy_state(self, tmp_path: Path) -> None: + """MCP 流程产物优先于旧 Web 会话的 agent_state.json。""" + cache = tmp_path / ".invoice_cache" + cache.mkdir() + (cache / "normal_info.json").write_text( + json.dumps({"basic_info": {"reimbursement_description": "来自 normal_info"}}), encoding="utf-8" + ) + (tmp_path / "agent_state.json").write_text( + json.dumps({"extracted_info": {"basic_info": {"reimbursement_description": "来自旧状态"}}}), + encoding="utf-8", + ) + assert load_project_desc(tmp_path) == "来自 normal_info" + + def test_falls_back_to_travel_info(self, tmp_path: Path) -> None: + cache = tmp_path / ".invoice_cache" + cache.mkdir() + (cache / "travel_info.json").write_text( + json.dumps({"basic_info": {"reimbursement_description": "来自 travel_info"}}), encoding="utf-8" + ) + assert load_project_desc(tmp_path) == "来自 travel_info" + + def test_legacy_agent_state_still_works(self, tmp_path: Path) -> None: + """旧会话目录没有 .invoice_cache 时仍要能读出事由。""" + (tmp_path / "agent_state.json").write_text( + json.dumps({"extracted_info": {"basic_info": {"reimbursement_description": "旧链路事由"}}}), + encoding="utf-8", + ) + assert load_project_desc(tmp_path) == "旧链路事由" + def test_reads_from_agent_state(self, tmp_path: Path) -> None: (tmp_path / "agent_state.json").write_text( json.dumps(