From af7bb6826863f057f799b605ab28aee289084159 Mon Sep 17 00:00:00 2001 From: wuyang <5700876+banisherwy@user.noreply.gitee.com> Date: Tue, 28 Jul 2026 17:31:37 +0800 Subject: [PATCH] feat: give the curator continuous context and durable idea history --- README.md | 16 +- app/curator.py | 466 ++++++++++++++++++++++++++++++++-------- app/db.py | 165 +++++++++++++- app/main.py | 29 ++- frontend/src/App.tsx | 77 ++++++- frontend/src/api.ts | 9 + frontend/src/styles.css | 60 +++++- tests/test_api.py | 55 ++++- tests/test_curator.py | 66 ++++++ tests/test_db.py | 138 ++++++++++++ 10 files changed, 967 insertions(+), 114 deletions(-) create mode 100644 tests/test_curator.py diff --git a/README.md b/README.md index 9466c9e..7bd8cb1 100644 --- a/README.md +++ b/README.md @@ -3,21 +3,23 @@ 一个“先保存，后理解”的私人 AI 笔记本。 记录页只呈现用户写下的原文，输入历史像对话一样始终可回看。后台的单一策展 -Agent 使用 DeepSeek 的思考模式，按需搜索已有想法与轨迹，再给出结构化的“想法位势” -提案。Agent 不能直接写数据库；应用只在 schema 和引用完整性校验通过后，用事务提交 -更新。 +Agent 使用 DeepSeek 的思考模式，结合最近的连续原文与精简想法目录，必要时再查看 +已有想法的完整轨迹，最后给出结构化的“想法位势”提案。Agent 不能直接写数据库； +应用只在 schema 和引用完整性校验通过后，用事务提交更新。 ## 设计边界 - 原始片段先持久化，AI 失败不影响记录。 - 捕获流不显示保存状态、AI 标签、关联或建议。 - 不要求用户先建页面、取标题、选分类或填写日期。 +- 不要求用户建立会话；系统在后台保留滚动的连续语境。 - 成熟度是可回退的连续位势，不是阶段、成绩或任务完成百分比。 - 运动、张力和可能动作由模型结合上下文动态生成，不使用固定关卡。 - 用户手动校准的位势优先展示，AI 估计仍独立保留。 - 每个用户的记录、想法、Session 与 Agent 上下文都以 `user_id` 在 SQL 层隔离。 - 管理员能看运行元数据；其他用户的原文与 AI 产物默认脱敏，只有用户主动开启 “调试共享”后才可见。 +- 每次想法更新和人工校准都会保存完整版本；原始片段始终是不可替代的证据层。 - 管理后台保存模型轮次、工具调用、耗时、token、错误和结构化决策产物，但不保存或 暴露模型隐藏思维链。 @@ -33,11 +35,14 @@ React/Vite ── cookie session ── FastAPI ── SQLite 核心数据流是： 1. `POST /api/fragments` 先提交原文并立即返回； -2. 后台 Agent 读取当前用户自己的想法上下文； -3. Agent 通过 `search_ideas`、`inspect_idea` 工具选择相关材料； +2. 后台 Agent 读取当前用户最近的连续输入和精简想法目录； +3. Agent 通过 `search_ideas`、`inspect_idea` 工具补充相关材料； 4. 应用校验结构化结果后，以事务写入想法、轨迹和片段关联； 5. 每次运行同时形成 `agent_runs`、`agent_events`，供管理后台聚合分析。 +工具探索超过预算或输出结构无效时，系统会进入一次没有工具的收敛回合；失败运行也可以由 +管理员重新排队。模型用量通过逐回合钩子采集，因此异常退出不会被误记为零。 + ## 本地运行 要求 Python 3.12+、Node.js 22+。 @@ -112,6 +117,7 @@ Session 会归属给配置清单中的第一位管理员；不会把旧数据复 - `GET /api/admin/users`：用户空间、数据量与调试共享状态； - `GET /api/admin/runs?limit=80`：Agent 运行列表； - `GET /api/admin/runs/{run_id}`：一次运行的事件、工具、token 与结构化产物； +- `POST /api/admin/runs/{run_id}/retry`：重新排队一次仍处于错误状态的分析； - `GET /api/admin/audit?limit=120`：登录、启动、记录、校准等应用审计事件。 普通用户可调用 `PATCH /api/account/debug-sharing` 控制自己的调试内容是否向管理员 diff --git a/app/curator.py b/app/curator.py index d2c7979..10acfad 100644 --- a/app/curator.py +++ b/app/curator.py @@ -3,19 +3,23 @@ from __future__ import annotations import asyncio import json import logging +import re import time -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any from agents import ( Agent, ModelSettings, OpenAIChatCompletionsModel, + RunHooks, RunContextWrapper, Runner, function_tool, set_tracing_disabled, ) +from agents.exceptions import MaxTurnsExceeded +from agents.items import ModelResponse from openai import AsyncOpenAI from openai.types.shared import Reasoning from pydantic import ValidationError @@ -25,7 +29,9 @@ from .db import Database from .schemas import CuratorDecision logger = logging.getLogger(__name__) -PROMPT_VERSION = "maturity-2026-07-28.v2" +PROMPT_VERSION = "maturity-2026-07-28.v3" +RECENT_CONTEXT_LIMIT = 12 +IDEA_CATALOG_LIMIT = 40 CURATOR_INSTRUCTIONS = """ @@ -52,8 +58,13 @@ motion 是你对当下运动的自然语言压缩，例如“向现实探去” 一个片段最多关联两个想法。若现在还不值得形成或归入想法，让它 standalone，原文仍会保留。 新建想法时 idea_id 必须为 null；更新时必须使用工具返回的准确 id。 -先查看候选想法；遇到可能相关或可能冲突的候选时，使用 inspect_idea。最终必须只返回 JSON， -不能用 Markdown 包裹，也不能附加解释。JSON 必须符合输入中给出的 schema。 +输入会直接提供最近的连续原文和精简想法目录。把“这个、上面、第二点、继续、刚才”等指代 +放回连续语境中理解；不要因为新片段换了关键词就丢掉思想上的延续。目录中的 id 只供系统内部 +引用。遇到可能相关或冲突的候选时，使用 inspect_idea；search_ideas 只用于目录过大或需要 +补充候选时，最多调用两次；同一个想法最多 inspect 一次。搜索无结果时不要反复改写查询， +也绝不能用 none、unknown 等占位符调用 inspect_idea。工具探索后必须及时收敛并返回最终判断。 + +最终必须只返回 JSON，不能用 Markdown 包裹，也不能附加解释。JSON 必须符合输入中给出的 schema。 不要在输出中评价用户、诊断心理、说教或伪造证据。 """.strip() @@ -67,6 +78,187 @@ class CuratorContext: recent_fragments: list[dict[str, Any]] search_count: int = 0 inspection_count: int = 0 + model_rounds: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + reasoning_tokens: int = 0 + cached_tokens: int = 0 + llm_started_at: float | None = None + inspected_idea_ids: set[str] = field(default_factory=set) + + +def _search_units(value: str) -> list[str]: + units: list[str] = [] + for token in re.findall(r"[a-zA-Z0-9_-]+|[\u3400-\u9fff]+", value.lower()): + if token not in units: + units.append(token) + if re.fullmatch(r"[\u3400-\u9fff]+", token) and len(token) > 2: + for index in range(len(token) - 1): + pair = token[index : index + 2] + if pair not in units: + units.append(pair) + return units[:24] + + +def _rank_ideas( + ideas: list[dict[str, Any]], query: str +) -> tuple[list[dict[str, Any]], bool]: + terms = _search_units(query) + ranked: list[tuple[int, str, dict[str, Any]]] = [] + for idea in ideas: + evidence = " ".join( + str(fragment.get("content", "")) + for fragment in idea.get("recent_fragments", []) + ) + haystack = " ".join( + str(idea.get(key, "")) + for key in ( + "title", + "summary", + "position", + "tension", + "trajectory", + ) + ) + haystack = f"{haystack} {evidence}".lower() + score = sum( + (4 if term in str(idea.get("title", "")).lower() else 1) + * haystack.count(term) + for term in terms + ) + if score: + ranked.append((score, str(idea.get("updated_at", "")), idea)) + fallback = not ranked + if fallback: + ranked = [ + (0, str(idea.get("updated_at", "")), idea) for idea in ideas + ] + ranked.sort(key=lambda item: (item[0], item[1]), reverse=True) + return [idea for _, _, idea in ranked[:8]], fallback + + +class CuratorRunHooks(RunHooks[CuratorContext]): + async def on_llm_start( + self, + context: RunContextWrapper[CuratorContext], + agent: Agent[CuratorContext], + system_prompt: str | None, + input_items: list[Any], + ) -> None: + context.context.llm_started_at = time.perf_counter() + + async def on_llm_end( + self, + context: RunContextWrapper[CuratorContext], + agent: Agent[CuratorContext], + response: ModelResponse, + ) -> None: + state = context.context + state.model_rounds += 1 + usage = response.usage + input_tokens = int(usage.input_tokens or 0) + output_tokens = int(usage.output_tokens or 0) + reasoning_tokens = int( + usage.output_tokens_details.reasoning_tokens or 0 + ) + cached_tokens = int( + usage.input_tokens_details.cached_tokens or 0 + ) + state.input_tokens += input_tokens + state.output_tokens += output_tokens + state.reasoning_tokens += reasoning_tokens + state.cached_tokens += cached_tokens + duration_ms = ( + round((time.perf_counter() - state.llm_started_at) * 1000) + if state.llm_started_at is not None + else None + ) + state.db.add_agent_event( + state.run_id, + state.user_id, + "model_round_completed", + { + "round": state.model_rounds, + "agent": agent.name, + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "reasoning_tokens": reasoning_tokens, + "cached_tokens": cached_tokens, + }, + duration_ms=duration_ms, + ) + state.llm_started_at = None + + +def build_curator_prompt( + fragment: dict[str, Any], + ideas: list[dict[str, Any]], + recent_fragments: list[dict[str, Any]], +) -> str: + prior_context = [ + { + "id": item["id"], + "created_at": item["created_at"], + "content": str(item["content"])[:1_200], + } + for item in recent_fragments + if item["id"] != fragment["id"] + and item["created_at"] <= fragment["created_at"] + ][-RECENT_CONTEXT_LIMIT:] + later_context = [ + { + "created_at": item["created_at"], + "content": str(item["content"])[:1_200], + } + for item in recent_fragments + if item["id"] != fragment["id"] + and item["created_at"] > fragment["created_at"] + ][-6:] + catalog = [ + { + "id": idea["id"], + "title": idea["title"], + "summary": idea["summary"], + "motion": idea["motion"], + "position": idea["position"], + "tension": idea["tension"], + "updated_at": idea["updated_at"], + } + for idea in ideas[:IDEA_CATALOG_LIMIT] + ] + new_fragment = { + "id": fragment["id"], + "created_at": fragment["created_at"], + "content": fragment["content"], + } + output_schema = CuratorDecision.model_json_schema() + catalog_note = ( + "目录已经覆盖全部已有想法。优先从目录直接判断候选，相关时用准确 id 调用 " + "inspect_idea；一般不需要 search_ideas。" + if len(ideas) <= IDEA_CATALOG_LIMIT + else "目录只包含最近更新的想法；找不到可能候选时可调用 search_ideas。" + ) + delayed_note = ( + "这是一次延迟重整。later_context 是这条记录之后才发生的输入，只用于避免把现有想法" + "的状态和轨迹倒退；不要把它们误当成当时的上文。应把当前片段作为迟到的证据，结合" + "想法目录中的最新状态作出增量判断。\n\n" + if later_context + else "" + ) + return ( + "请策展这个刚刚保存的新片段。最近原文按时间从早到晚排列，它们是理解“这个、" + "第二点、继续、刚才”等连续表达的第一依据；不要要求用户显式建立会话。\n\n" + f"{json.dumps(prior_context, ensure_ascii=False)}\n\n" + f"{json.dumps(new_fragment, ensure_ascii=False)}\n\n" + f"{json.dumps(later_context, ensure_ascii=False)}\n\n" + f"{delayed_note}" + f"{json.dumps(catalog, ensure_ascii=False)}\n\n" + f"当前共有 {len(ideas)} 个已有想法。{catalog_note}\n" + "如果新片段既验证了一个既有想法、又形成了一个新的独立问题，可以给出两个 assessment；" + "不要因为表面主题变化而遗漏它对上一段思想过程的反馈。\n\n" + "最终只输出符合以下 JSON Schema 的 JSON 对象：\n" + f"{json.dumps(output_schema, ensure_ascii=False)}" + ) @function_tool(strict_mode=False) @@ -76,17 +268,31 @@ async def search_ideas( """Search existing evolving ideas by a short concept, question, or tension.""" ctx.context.search_count += 1 started = time.perf_counter() - terms = [term.lower() for term in query.split() if term.strip()] - ranked: list[tuple[int, dict[str, Any]]] = [] - for idea in ctx.context.ideas: - haystack = " ".join( - str(idea.get(key, "")) - for key in ("title", "summary", "position", "tension", "trajectory") - ).lower() - score = sum(haystack.count(term) for term in terms) - if score or not terms: - ranked.append((score, idea)) - ranked.sort(key=lambda item: (item[0], item[1].get("updated_at", "")), reverse=True) + if ctx.context.search_count > 2: + candidates = [ + {"id": item["id"], "title": item["title"]} + for item in ctx.context.ideas[:8] + ] + ctx.context.db.add_agent_event( + ctx.context.run_id, + ctx.context.user_id, + "tool_search_ideas", + { + "query": query, + "result_count": len(candidates), + "budget_exhausted": True, + "candidates": candidates, + }, + duration_ms=round((time.perf_counter() - started) * 1000), + ) + return json.dumps( + { + "notice": "搜索额度已用完。使用已有目录并立即形成最终判断。", + "candidates": candidates, + }, + ensure_ascii=False, + ) + ranked, fallback = _rank_ideas(ctx.context.ideas, query) compact = [ { "id": idea["id"], @@ -97,7 +303,7 @@ async def search_ideas( "position": idea["position"], "tension": idea["tension"], } - for _, idea in ranked[:8] + for idea in ranked ] ctx.context.db.add_agent_event( ctx.context.run_id, @@ -106,6 +312,7 @@ async def search_ideas( { "query": query, "result_count": len(compact), + "fallback_to_recent": fallback, "candidates": [ {"id": item["id"], "title": item["title"]} for item in compact ], @@ -134,7 +341,39 @@ async def inspect_idea( {"idea_id": idea_id, "found": False}, duration_ms=round((time.perf_counter() - started) * 1000), ) - return json.dumps({"error": "idea not found"}) + return json.dumps( + { + "error": "idea not found", + "valid_candidates": [ + {"id": item["id"], "title": item["title"]} + for item in ctx.context.ideas[:8] + ], + }, + ensure_ascii=False, + ) + if idea_id in ctx.context.inspected_idea_ids: + ctx.context.db.add_agent_event( + ctx.context.run_id, + ctx.context.user_id, + "tool_inspect_idea", + { + "idea_id": idea_id, + "found": True, + "reused": True, + "title": idea["title"], + }, + duration_ms=round((time.perf_counter() - started) * 1000), + ) + return json.dumps( + { + "notice": ( + "这个想法已经检查过。使用上一份结果并立即形成最终判断，" + "不要再次调用工具。" + ) + }, + ensure_ascii=False, + ) + ctx.context.inspected_idea_ids.add(idea_id) payload = { "id": idea["id"], "title": idea["title"], @@ -244,22 +483,19 @@ class Curator: "fragment_characters": len(fragment["content"]), "idea_count": len(ideas), "recent_fragment_count": len(recent_fragments), + "prior_fragment_count": sum( + item["id"] != fragment_id + and item["created_at"] <= fragment["created_at"] + for item in recent_fragments + ), + "later_fragment_count": sum( + item["id"] != fragment_id + and item["created_at"] > fragment["created_at"] + for item in recent_fragments + ), }, ) - output_schema = CuratorDecision.model_json_schema() - lookup_instruction = ( - "必须先调用 search_ideas 搜索共享的问题或张力；如果候选可能相关，" - "再调用 inspect_idea 查看它的真实轨迹后判断。\n\n" - if ideas - else "目前没有已有想法，不要调用搜索工具。\n\n" - ) - prompt = ( - "请策展这个刚刚保存的新片段：\n" - f"{fragment['content']}\n\n" - f"当前共有 {len(ideas)} 个已有想法。{lookup_instruction}" - "最终只输出符合以下 JSON Schema 的 JSON 对象：\n" - f"{json.dumps(output_schema, ensure_ascii=False)}" - ) + prompt = build_curator_prompt(fragment, ideas, recent_fragments) set_tracing_disabled(True) client = AsyncOpenAI( @@ -270,63 +506,115 @@ class Curator: model=self.settings.deepseek_model, openai_client=client, ) + model_settings = ModelSettings( + max_tokens=2_400, + reasoning=Reasoning(effort="high"), + extra_body={"thinking": {"type": "enabled"}}, + extra_args={"response_format": {"type": "json_object"}}, + ) agent: Agent[CuratorContext] = Agent( name="私人思想策展者", instructions=CURATOR_INSTRUCTIONS, model=model, tools=[search_ideas, inspect_idea], - model_settings=ModelSettings( - max_tokens=2_400, - reasoning=Reasoning(effort="high"), - extra_body={"thinking": {"type": "enabled"}}, - extra_args={"response_format": {"type": "json_object"}}, + model_settings=model_settings, + ) + repair_agent: Agent[CuratorContext] = Agent( + name="判断收敛器", + instructions=( + CURATOR_INSTRUCTIONS + + "\n\n这是收敛回合。不能调用任何工具；只使用输入中已经提供的连续原文、" + "想法目录和 schema，立即给出最终合法 JSON。" ), + model=model, + tools=[], + model_settings=model_settings, ) decision: CuratorDecision | None = None last_error: Exception | None = None attempt_count = 0 - model_rounds = 0 - usage = { - "input_tokens": 0, - "output_tokens": 0, - "reasoning_tokens": 0, - "cached_tokens": 0, - } + hooks = CuratorRunHooks() try: for attempt in range(2): attempt_count = attempt + 1 - repair = ( - "" - if attempt == 0 - else "\n\n上一次输出未通过结构校验。重新完整判断，并只返回合法 JSON。" - ) + is_repair = attempt > 0 self.db.add_agent_event( run_id, user_id, "attempt_started", - {"attempt": attempt_count, "is_repair": attempt > 0}, + {"attempt": attempt_count, "is_repair": is_repair}, ) attempt_started = time.perf_counter() - result = await Runner.run( - agent, - input=prompt + repair, - context=context, - max_turns=4, - ) + rounds_before = context.model_rounds + usage_before = { + "input_tokens": context.input_tokens, + "output_tokens": context.output_tokens, + "reasoning_tokens": context.reasoning_tokens, + "cached_tokens": context.cached_tokens, + } + try: + result = await Runner.run( + repair_agent if is_repair else agent, + input=( + prompt + + ( + "\n\n上一次工具探索或输出没有收敛。不要再探索，" + "重新完整判断并只返回合法 JSON。" + if is_repair + else "" + ) + ), + context=context, + max_turns=2 if is_repair else 6, + hooks=hooks, + ) + except MaxTurnsExceeded as exc: + attempt_duration = round( + (time.perf_counter() - attempt_started) * 1000 + ) + current_usage = { + key: getattr(context, key) - value + for key, value in usage_before.items() + } + self.db.add_agent_event( + run_id, + user_id, + "model_attempt_completed", + { + "attempt": attempt_count, + "model_rounds": ( + context.model_rounds - rounds_before + ), + "completed": False, + "error_type": type(exc).__name__, + **current_usage, + }, + duration_ms=attempt_duration, + ) + last_error = exc + if not is_repair: + self.db.add_agent_event( + run_id, + user_id, + "convergence_repair_started", + {"reason": "tool_loop_exceeded"}, + ) + continue attempt_duration = round( (time.perf_counter() - attempt_started) * 1000 ) - current_usage = self._result_usage(result) - model_rounds += current_usage.pop("model_rounds") - for key in usage: - usage[key] += current_usage[key] + current_usage = { + key: getattr(context, key) - value + for key, value in usage_before.items() + } self.db.add_agent_event( run_id, user_id, "model_attempt_completed", { "attempt": attempt_count, - "model_rounds": len(result.raw_responses), + "model_rounds": context.model_rounds - rounds_before, + "completed": True, **current_usage, }, duration_ms=attempt_duration, @@ -336,9 +624,17 @@ class Curator: if not isinstance(raw, str): raise TypeError("curator output was not text") decision = CuratorDecision.model_validate_json(raw) - if ideas and context.search_count == 0: - decision = None - raise ValueError("curator skipped required idea search") + valid_idea_ids = {idea["id"] for idea in ideas} + unknown_ids = [ + item.idea_id + for item in decision.assessments + if item.idea_id + and item.idea_id not in valid_idea_ids + ] + if unknown_ids: + raise ValueError( + "curator referenced an unknown existing idea" + ) break except (ValidationError, ValueError, TypeError) as exc: decision = None @@ -358,9 +654,16 @@ class Curator: fragment_id, attempt_count, ) + if not is_repair: + self.db.add_agent_event( + run_id, + user_id, + "convergence_repair_started", + {"reason": "output_validation_failed"}, + ) if decision is None: raise RuntimeError( - "curator returned invalid JSON twice" + "curator could not produce a valid decision" ) from last_error assessments = ( [] @@ -386,11 +689,14 @@ class Curator: status="success", duration_ms=round((time.perf_counter() - run_started) * 1000), attempt_count=attempt_count, - model_rounds=model_rounds, + model_rounds=context.model_rounds, tool_calls=context.search_count + context.inspection_count, search_calls=context.search_count, inspection_calls=context.inspection_count, - **usage, + input_tokens=context.input_tokens, + output_tokens=context.output_tokens, + reasoning_tokens=context.reasoning_tokens, + cached_tokens=context.cached_tokens, ) logger.info( "Curator completed fragment %s with %s search and %s inspection calls", @@ -413,32 +719,14 @@ class Curator: status="error", duration_ms=round((time.perf_counter() - run_started) * 1000), attempt_count=attempt_count, - model_rounds=model_rounds, + model_rounds=context.model_rounds, tool_calls=context.search_count + context.inspection_count, search_calls=context.search_count, inspection_calls=context.inspection_count, error=exc, - **usage, + input_tokens=context.input_tokens, + output_tokens=context.output_tokens, + reasoning_tokens=context.reasoning_tokens, + cached_tokens=context.cached_tokens, ) raise - - @staticmethod - def _result_usage(result: Any) -> dict[str, int]: - totals = { - "model_rounds": len(result.raw_responses), - "input_tokens": 0, - "output_tokens": 0, - "reasoning_tokens": 0, - "cached_tokens": 0, - } - for response in result.raw_responses: - response_usage = response.usage - totals["input_tokens"] += response_usage.input_tokens - totals["output_tokens"] += response_usage.output_tokens - totals["reasoning_tokens"] += ( - response_usage.output_tokens_details.reasoning_tokens or 0 - ) - totals["cached_tokens"] += ( - response_usage.input_tokens_details.cached_tokens or 0 - ) - return totals diff --git a/app/db.py b/app/db.py index c37f585..d10d7b5 100644 --- a/app/db.py +++ b/app/db.py @@ -103,12 +103,18 @@ class Database: id TEXT PRIMARY KEY, idea_id TEXT NOT NULL REFERENCES ideas(id) ON DELETE CASCADE, source_fragment_id TEXT REFERENCES fragments(id) ON DELETE SET NULL, + title TEXT, + summary TEXT, maturity_ai REAL NOT NULL, + maturity_override REAL, + confidence REAL, motion TEXT NOT NULL, position TEXT NOT NULL, tension TEXT NOT NULL, trajectory TEXT NOT NULL, possible_moves TEXT NOT NULL, + change_kind TEXT NOT NULL DEFAULT 'legacy_partial', + fragment_ids TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL ); @@ -185,6 +191,26 @@ class Database: self._ensure_column(connection, "fragments", "user_id", "TEXT") self._ensure_column(connection, "ideas", "user_id", "TEXT") self._ensure_column(connection, "sessions", "user_id", "TEXT") + self._ensure_column(connection, "idea_snapshots", "title", "TEXT") + self._ensure_column(connection, "idea_snapshots", "summary", "TEXT") + self._ensure_column( + connection, "idea_snapshots", "maturity_override", "REAL" + ) + self._ensure_column( + connection, "idea_snapshots", "confidence", "REAL" + ) + self._ensure_column( + connection, + "idea_snapshots", + "change_kind", + "TEXT NOT NULL DEFAULT 'legacy_partial'", + ) + self._ensure_column( + connection, + "idea_snapshots", + "fragment_ids", + "TEXT NOT NULL DEFAULT '[]'", + ) now = utc_now() for seed in seeds: @@ -457,8 +483,10 @@ class Database: ).fetchall() snapshots = connection.execute( """ - SELECT s.maturity_ai, s.motion, s.position, s.tension, - s.trajectory, s.possible_moves, s.created_at + SELECT s.title, s.summary, s.maturity_ai, + s.maturity_override, s.confidence, s.motion, + s.position, s.tension, s.trajectory, s.possible_moves, + s.change_kind, s.fragment_ids, s.created_at FROM idea_snapshots s JOIN ideas i ON i.id = s.idea_id WHERE s.idea_id = ? AND i.user_id = ? @@ -471,6 +499,7 @@ class Database: { **dict(row), "possible_moves": _loads(row["possible_moves"], []), + "fragment_ids": _loads(row["fragment_ids"], []), } for row in snapshots ] @@ -588,23 +617,55 @@ class Database: now, ), ) + version_state = connection.execute( + """ + SELECT maturity_override FROM ideas + WHERE id = ? AND user_id = ? + """, + (idea_id, user_id), + ).fetchone() + fragment_ids = [ + row["fragment_id"] + for row in connection.execute( + """ + SELECT l.fragment_id + FROM idea_fragments l + JOIN fragments f ON f.id = l.fragment_id + WHERE l.idea_id = ? AND f.user_id = ? + ORDER BY f.created_at ASC, f.rowid ASC + """, + (idea_id, user_id), + ).fetchall() + ] connection.execute( """ INSERT INTO idea_snapshots ( - id, idea_id, source_fragment_id, maturity_ai, motion, - position, tension, trajectory, possible_moves, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + id, idea_id, source_fragment_id, title, summary, + maturity_ai, maturity_override, confidence, motion, + position, tension, trajectory, possible_moves, + change_kind, fragment_ids, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( str(uuid.uuid4()), idea_id, fragment_id, + assessment["title"], + assessment["summary"], assessment["maturity"], + ( + version_state["maturity_override"] + if version_state + else None + ), + assessment["confidence"], assessment["motion"], assessment["position"], assessment["tension"], assessment["trajectory"], _json(assessment["possible_moves"]), + "analysis", + _json(fragment_ids), now, ), ) @@ -622,16 +683,66 @@ class Database: def update_idea_override( self, user_id: str, idea_id: str, maturity_override: float | None ) -> dict[str, Any] | None: + now = utc_now() with self.connect() as connection: result = connection.execute( """ UPDATE ideas SET maturity_override = ?, updated_at = ? WHERE id = ? AND user_id = ? """, - (maturity_override, utc_now(), idea_id, user_id), + (maturity_override, now, idea_id, user_id), ) if result.rowcount == 0: return None + idea = connection.execute( + """ + SELECT title, summary, maturity_ai, maturity_override, + confidence, motion, position, tension, trajectory, + possible_moves + FROM ideas WHERE id = ? AND user_id = ? + """, + (idea_id, user_id), + ).fetchone() + fragment_ids = [ + row["fragment_id"] + for row in connection.execute( + """ + SELECT l.fragment_id + FROM idea_fragments l + JOIN fragments f ON f.id = l.fragment_id + WHERE l.idea_id = ? AND f.user_id = ? + ORDER BY f.created_at ASC, f.rowid ASC + """, + (idea_id, user_id), + ).fetchall() + ] + connection.execute( + """ + INSERT INTO idea_snapshots ( + id, idea_id, source_fragment_id, title, summary, + maturity_ai, maturity_override, confidence, motion, + position, tension, trajectory, possible_moves, + change_kind, fragment_ids, created_at + ) VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + str(uuid.uuid4()), + idea_id, + idea["title"], + idea["summary"], + idea["maturity_ai"], + idea["maturity_override"], + idea["confidence"], + idea["motion"], + idea["position"], + idea["tension"], + idea["trajectory"], + idea["possible_moves"], + "manual_calibration", + _json(fragment_ids), + now, + ), + ) return self.get_idea(user_id, idea_id) @staticmethod @@ -928,6 +1039,48 @@ class Database: ).hexdigest()[:16] return run + def prepare_agent_retry(self, run_id: str) -> dict[str, str] | None: + with self.connect() as connection: + row = connection.execute( + """ + SELECT r.user_id, r.fragment_id, r.status AS run_status, + f.analysis_status + FROM agent_runs r + JOIN fragments f ON f.id = r.fragment_id + WHERE r.id = ? AND f.user_id = r.user_id + """, + (run_id,), + ).fetchone() + if ( + not row + or row["run_status"] != "error" + or row["analysis_status"] != "error" + ): + return None + newer = connection.execute( + """ + SELECT 1 FROM agent_runs + WHERE fragment_id = ? AND user_id = ? + AND id != ? AND status IN ('running', 'success') + LIMIT 1 + """, + (row["fragment_id"], row["user_id"], run_id), + ).fetchone() + if newer: + return None + connection.execute( + """ + UPDATE fragments + SET analysis_status = 'pending', analysis_error = NULL + WHERE id = ? AND user_id = ? + """, + (row["fragment_id"], row["user_id"]), + ) + return { + "user_id": row["user_id"], + "fragment_id": row["fragment_id"], + } + def list_agent_events(self, run_id: str) -> list[dict[str, Any]]: with self.connect() as connection: rows = connection.execute( diff --git a/app/main.py b/app/main.py index 7fc505f..79c5fa7 100644 --- a/app/main.py +++ b/app/main.py @@ -276,7 +276,12 @@ def _redacted_event(event: dict[str, Any]) -> dict[str, Any]: safe = payload elif event_type == "context_loaded": safe = payload - elif event_type in {"attempt_started", "model_attempt_completed"}: + elif event_type in { + "attempt_started", + "model_round_completed", + "model_attempt_completed", + "convergence_repair_started", + }: safe = payload elif event_type == "tool_search_ideas": safe = {"result_count": payload.get("result_count", 0)} @@ -341,6 +346,28 @@ async def admin_run_detail( } +@app.post( + "/api/admin/runs/{run_id}/retry", + status_code=status.HTTP_202_ACCEPTED, +) +async def retry_admin_run( + run_id: str, admin: AuthUser = Depends(administrator) +): + retry = db.prepare_agent_retry(run_id) + if not retry: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="这次分析现在不能重新运行", + ) + curator.enqueue(retry["fragment_id"]) + db.add_audit_event( + "agent_run_requeued", + admin.id, + {"run_id": run_id, "owner_user_id": retry["user_id"]}, + ) + return {"queued": True} + + static_dir = Path(__file__).parent / "static" assets_dir = static_dir / "assets" if assets_dir.is_dir(): diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index b66d1cf..f147cc1 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -363,12 +363,41 @@ function IdeaDetail({ + {idea.snapshots && idea.snapshots.length > 1 && ( +
+ 这个想法走过的路 +
+ {[...idea.snapshots].reverse().map((snapshot, index) => ( +
+ + + {snapshot.change_kind === "manual_calibration" + ? "你重新放置了它" + : snapshot.motion} + +

{snapshot.trajectory}

+
+ ))} +
+
+ )} + {idea.fragments && idea.fragments.length > 0 && (
- 构成这个想法的片段 + 原始脉络
{idea.fragments.map((fragment) => ( -
{fragment.content}
+
+ +
{fragment.content}
+
))}
@@ -454,6 +483,8 @@ const eventNames: Record = { tool_search_ideas: "搜索已有想法", tool_inspect_idea: "查看想法轨迹", model_attempt_completed: "模型返回", + model_round_completed: "完成一轮判断", + convergence_repair_started: "转入收敛判断", validation_failed: "结构校验未通过", decision_committed: "提交分析结果", run_failed: "运行失败", @@ -466,6 +497,7 @@ const auditNames: Record = { logout: "退出登录", fragment_created: "保存新片段", idea_position_calibrated: "人工校准位势", + agent_run_requeued: "重新分析", debug_sharing_changed: "调整调试共享", }; @@ -510,6 +542,18 @@ function EventPayload({ event }: { event: AgentEvent }) {

); } + if (event.event_type === "model_round_completed") { + return ( +

+ 输入 {compactNumber(Number(payload.input_tokens ?? 0))} · 输出{" "} + {compactNumber(Number(payload.output_tokens ?? 0))} · 推理{" "} + {compactNumber(Number(payload.reasoning_tokens ?? 0))} +

+ ); + } + if (event.event_type === "convergence_repair_started") { + return

工具探索没有及时收束，系统改用已有上下文直接完成判断。

; + } if (event.event_type === "validation_failed") { return (

@@ -561,11 +605,24 @@ function RunDetail({ onClose: () => void; }) { const [detail, setDetail] = useState(null); + const [retrying, setRetrying] = useState(false); + const [requeued, setRequeued] = useState(false); useEffect(() => { api.adminRun(runId).then(setDetail); }, [runId]); + async function retry() { + if (retrying || requeued) return; + setRetrying(true); + try { + await api.retryAdminRun(runId); + setRequeued(true); + } finally { + setRetrying(false); + } + } + return (

{detail.fragment.content} ) : (
- 内容已隔离 · {detail.fragment.content_length} 字 · 指纹{" "} - {detail.fragment.content_sha256} + 内容已隔离 · {detail.fragment.content_length} 字
)} {detail.privacy.reason} @@ -636,6 +692,19 @@ function RunDetail({
))} + {detail.run.status === "error" && ( + + )}
不保存模型隐藏思维链；这里展示的是输入、工具行为、用量、校验与最终结构化产物。
diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 956ac71..ebbe352 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -12,12 +12,17 @@ export type User = { }; export type Snapshot = { + title: string | null; + summary: string | null; maturity_ai: number; + maturity_override: number | null; + confidence: number | null; motion: string; position: string; tension: string; trajectory: string; possible_moves: string[]; + change_kind: "analysis" | "manual_calibration" | "legacy_partial"; created_at: string; }; @@ -226,6 +231,10 @@ export const api = { request<{ items: AgentRun[] }>(`/api/admin/runs?limit=${limit}`), adminRun: (id: string) => request(`/api/admin/runs/${id}`), + retryAdminRun: (id: string) => + request<{ queued: boolean }>(`/api/admin/runs/${id}/retry`, { + method: "POST", + }), adminAudit: (limit = 80) => request<{ items: AuditEvent[] }>(`/api/admin/audit?limit=${limit}`), }; diff --git a/frontend/src/styles.css b/frontend/src/styles.css index 7bdaad1..6452736 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -885,12 +885,18 @@ summary:focus-visible { color: var(--warm); } +.idea-history, .evidence { margin-top: 50px; border-top: 1px solid var(--line); color: var(--muted); } +.idea-history + .evidence { + margin-top: 0; +} + +.idea-history summary, .evidence summary { padding: 18px 0; cursor: pointer; @@ -898,12 +904,44 @@ summary:focus-visible { letter-spacing: 0.05em; } +.idea-history article { + display: grid; + grid-template-columns: 92px 1fr; + gap: 4px 14px; + padding: 14px 0; + border-bottom: 1px solid rgba(23, 37, 31, 0.07); +} + +.idea-history time, +.evidence time { + color: #92958f; + font-size: 9px; +} + +.idea-history strong { + color: var(--ink); + font-size: 12px; + font-weight: 500; +} + +.idea-history p { + grid-column: 2; + margin: 2px 0 0; + color: #626b66; + font-size: 11px; + line-height: 1.7; +} + +.evidence article { + padding: 13px 0; + border-bottom: 1px solid rgba(23, 37, 31, 0.07); +} + .evidence blockquote { - margin: 0; - padding: 15px 0; + margin: 5px 0 0; + padding: 0; color: #4e5752; font: 400 14px/1.85 "Songti SC", "STSong", serif; - border-bottom: 1px solid rgba(23, 37, 31, 0.07); } .admin-view { @@ -1411,6 +1449,22 @@ summary:focus-visible { margin: 0 !important; } +.retry-run { + margin-top: 28px; + padding: 9px 13px; + border: 1px solid rgba(153, 104, 72, 0.28); + border-radius: 4px; + background: transparent; + color: var(--warm); + cursor: pointer; + font-size: 10px; +} + +.retry-run:disabled { + cursor: default; + opacity: 0.6; +} + .reasoning-boundary { margin-top: 45px; padding-top: 16px; diff --git a/tests/test_api.py b/tests/test_api.py index 45667ae..ad397d5 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -82,9 +82,10 @@ def test_login_isolation_roles_and_debug_privacy( headers=WRITE_HEADERS, ) assert member_fragment.status_code == 201 - assert [item["content"] for item in client.get("/api/fragments").json()["items"]] == [ - "用户二的记录" - ] + assert [ + item["content"] + for item in client.get("/api/fragments").json()["items"] + ] == ["用户二的记录"] assert client.get("/api/admin/overview").status_code == 403 run_id = database.start_agent_run( @@ -105,9 +106,10 @@ def test_login_isolation_roles_and_debug_privacy( ) client.cookies.set(COOKIE_NAME, admin_cookie) - assert [item["content"] for item in client.get("/api/fragments").json()["items"]] == [ - "管理员的记录" - ] + assert [ + item["content"] + for item in client.get("/api/fragments").json()["items"] + ] == ["管理员的记录"] redacted = client.get(f"/api/admin/runs/{run_id}").json() assert redacted["fragment"]["content_visible"] is False assert redacted["events"][-1]["payload"] == { @@ -127,3 +129,44 @@ def test_login_isolation_roles_and_debug_privacy( visible = client.get(f"/api/admin/runs/{run_id}").json() assert visible["fragment"]["content_visible"] is True assert visible["fragment"]["content"] == "用户二的记录" + + retry_error = RuntimeError("tool loop exceeded") + database.finish_agent_run( + run_id, + status="error", + duration_ms=100, + attempt_count=1, + model_rounds=4, + tool_calls=4, + search_calls=3, + inspection_calls=1, + input_tokens=400, + output_tokens=80, + reasoning_tokens=60, + cached_tokens=0, + error=retry_error, + ) + database.set_fragment_status( + member_fragment.json()["id"], "error", str(retry_error) + ) + queued: list[str] = [] + monkeypatch.setattr(main.curator, "enqueue", queued.append) + + client.cookies.set(COOKIE_NAME, member_cookie) + assert client.post( + f"/api/admin/runs/{run_id}/retry", + headers=WRITE_HEADERS, + ).status_code == 403 + + client.cookies.set(COOKIE_NAME, admin_cookie) + retried = client.post( + f"/api/admin/runs/{run_id}/retry", + headers=WRITE_HEADERS, + ) + assert retried.status_code == 202 + assert retried.json() == {"queued": True} + assert queued == [member_fragment.json()["id"]] + assert client.post( + f"/api/admin/runs/{run_id}/retry", + headers=WRITE_HEADERS, + ).status_code == 409 diff --git a/tests/test_curator.py b/tests/test_curator.py new file mode 100644 index 0000000..5f418d7 --- /dev/null +++ b/tests/test_curator.py @@ -0,0 +1,66 @@ +import asyncio +from pathlib import Path +from types import SimpleNamespace + +from agents.exceptions import MaxTurnsExceeded + +from app.config import Settings, UserSeed +from app.curator import Curator, Runner +from app.db import Database + + +def test_tool_loop_uses_no_tool_convergence_repair( + tmp_path: Path, monkeypatch +): + admin = UserSeed( + id="admin", + label="管理员", + role="admin", + access_key_hash="unused", + ) + database = Database(tmp_path / "repair.sqlite3") + database.initialize([admin]) + fragment = database.create_fragment("admin", "一个需要收敛的念头") + settings = Settings( + data_dir=tmp_path, + database_path=tmp_path / "repair.sqlite3", + users=(admin,), + session_secret="unused", + deepseek_api_key="test-key", + deepseek_base_url="https://api.deepseek.com", + deepseek_model="deepseek-v4-pro", + cookie_secure=False, + auth_disabled=True, + ) + calls = 0 + + async def fake_run(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + raise MaxTurnsExceeded("Max turns (6) exceeded") + assert kwargs["max_turns"] == 2 + assert args[0].tools == [] + return SimpleNamespace( + final_output=( + '{"standalone":true,"reasoning_note":"暂时独立保留",' + '"assessments":[]}' + ) + ) + + monkeypatch.setattr(Runner, "run", fake_run) + curator = Curator(database, settings) + + asyncio.run(curator.analyze(fragment["id"])) + + assert calls == 2 + assert database.get_fragment(fragment["id"], "admin")[ + "analysis_status" + ] == "done" + run = database.list_agent_runs(1)[0] + events = database.list_agent_events(run["id"]) + assert run["status"] == "success" + assert run["attempt_count"] == 2 + assert "convergence_repair_started" in [ + event["event_type"] for event in events + ] diff --git a/tests/test_db.py b/tests/test_db.py index 17bb344..895115f 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -1,6 +1,7 @@ from pathlib import Path from app.config import UserSeed +from app.curator import _rank_ideas, build_curator_prompt from app.db import Database @@ -97,6 +98,105 @@ def test_apply_assessment_builds_trajectory(tmp_path: Path): assert idea["maturity"] == 46 +def test_curator_prompt_contains_recent_continuity_and_idea_catalog( + tmp_path: Path, +): + db = database(tmp_path) + first = db.create_fragment( + ADMIN.id, + "我先让 AI 完成综述，接下来会抽取几个感兴趣的地方。", + ) + db.apply_assessments(ADMIN.id, first["id"], [assessment()]) + current = db.create_fragment( + ADMIN.id, + "我识别出了两个新的认知，第一点让我耳目一新。", + ) + ideas, fragments = db.curator_context(ADMIN.id) + + prompt = build_curator_prompt(current, ideas, fragments) + + assert "我先让 AI 完成综述" in prompt + assert "我识别出了两个新的认知" in prompt + assert "无分类的笔记入口" in prompt + assert "不要要求用户显式建立会话" in prompt + + +def test_delayed_reanalysis_keeps_later_input_out_of_prior_context( + tmp_path: Path, +): + db = database(tmp_path) + previous = db.create_fragment(ADMIN.id, "当时真正的上文") + target = db.create_fragment(ADMIN.id, "后来需要重新整理的这一条") + future = db.create_fragment(ADMIN.id, "这句话发生在目标片段之后") + timeline = ( + (previous, "2026-07-28T10:00:00.000+00:00"), + (target, "2026-07-28T10:10:00.000+00:00"), + (future, "2026-07-28T10:20:00.000+00:00"), + ) + with db.connect() as connection: + for fragment, created_at in timeline: + fragment["created_at"] = created_at + connection.execute( + "UPDATE fragments SET created_at = ? WHERE id = ?", + (created_at, fragment["id"]), + ) + ideas, fragments = db.curator_context(ADMIN.id) + + prompt = build_curator_prompt(target, ideas, fragments) + recent_section = prompt.split("", 1)[1].split( + "", 1 + )[0] + later_section = prompt.split("", 1)[1].split( + "", 1 + )[0] + + assert previous["content"] in recent_section + assert future["content"] not in recent_section + assert future["content"] in later_section + assert "延迟重整" in prompt + + +def test_idea_search_falls_back_to_recent_catalog(): + idea = { + "id": "idea-one", + "title": "先综述再深入", + "summary": "一种理解策略", + "position": "正在实践", + "tension": "可靠性", + "trajectory": "从原则走向实践", + "updated_at": "2026-07-28", + "recent_fragments": [], + } + + candidates, fallback = _rank_ideas( + [idea], "agent architecture long task reliability" + ) + + assert fallback is True + assert candidates == [idea] + + +def test_idea_versions_preserve_complete_state(tmp_path: Path): + db = database(tmp_path) + first = db.create_fragment(ADMIN.id, "第一条原始证据") + db.apply_assessments(ADMIN.id, first["id"], [assessment(maturity=30)]) + idea = db.list_ideas(ADMIN.id)[0] + first_version = db.get_idea(ADMIN.id, idea["id"])["snapshots"][-1] + + assert first_version["title"] == "无分类的笔记入口" + assert first_version["summary"] == "先保留念头，再让结构在后台形成。" + assert first_version["confidence"] == 0.76 + assert first_version["change_kind"] == "analysis" + assert first_version["fragment_ids"] == [first["id"]] + + db.update_idea_override(ADMIN.id, idea["id"], 61) + versions = db.get_idea(ADMIN.id, idea["id"])["snapshots"] + + assert versions[-1]["change_kind"] == "manual_calibration" + assert versions[-1]["maturity_override"] == 61 + assert versions[-1]["fragment_ids"] == [first["id"]] + + def test_manual_position_is_authoritative(tmp_path: Path): db = database(tmp_path) fragment = db.create_fragment(ADMIN.id, "一个想法") @@ -188,3 +288,41 @@ def test_agent_run_records_metrics_and_events(tmp_path: Path): "tool_search_ideas", ] assert overview["last_24h"]["success_rate"] == 100 + + +def test_failed_agent_run_can_be_safely_requeued(tmp_path: Path): + db = database(tmp_path) + fragment = db.create_fragment(ADMIN.id, "需要重新整理") + run_id = db.start_agent_run( + ADMIN.id, + fragment["id"], + "deepseek-v4-pro", + "prompt.v2", + "high", + ) + error = RuntimeError("temporary failure") + db.finish_agent_run( + run_id, + status="error", + duration_ms=100, + attempt_count=1, + model_rounds=1, + tool_calls=0, + search_calls=0, + inspection_calls=0, + input_tokens=10, + output_tokens=2, + reasoning_tokens=1, + cached_tokens=0, + error=error, + ) + db.set_fragment_status(fragment["id"], "error", str(error)) + + retry = db.prepare_agent_retry(run_id) + + assert retry == { + "user_id": ADMIN.id, + "fragment_id": fragment["id"], + } + assert db.get_fragment(fragment["id"], ADMIN.id)["analysis_status"] == "pending" + assert db.prepare_agent_retry(run_id) is None