diff --git a/PRD.md b/PRD.md index e960eeab..51905f43 100644 --- a/PRD.md +++ b/PRD.md @@ -1,6 +1,6 @@ # Foresight 先见之明 — 产品需求文档 (PRD) -> **版本**:v0.3 — 2026-04-15 大版本前最后一次校准 +> **版本**:v0.3.1 — 2026-04-15 生产链路打通 > **基线**:基于 [MiroFish](https://github.com/666ghj/MiroFish) v0.1.2 二次开发,已大量重构。 --- @@ -449,7 +449,35 @@ FLASK_DEBUG=False ## 8. v0.3 当前状态与已交付 -### v0.3 已完成(本次大版本前的全部更新) +### v0.3.1 hotfix(2026-04-15 生产链路打通) + +经过一次完整的端到端测试,修复了 7 个阻塞生产跑通的 bug: + +- [x] **Manus 式沉浸视图重构** — 回放界面从三栏分析面板 → Manus 风格单栏 cinematic 视图 + - 顶部 breadcrumb、浏览器外壳 + 平台原生帖子卡片、Jump to live 按钮、live 指示灯 + - 右上角 ◫/▦ 可切回三栏分析视图 + - 自动轮询:sim running 时每 10s 拉数据 + - 文件:`frontend/src/views/SimulationReplayView.vue` + +- [x] **LLM 客户端指数退避重试** — 识别 `RateLimitError / 429 / 5xx / 1302 / 超时`,5 次重试 1→2→4→8→16s 带随机抖动 + - 文件:`backend/app/utils/llm_client.py` + +- [x] **GLM → Qwen 32B 双 LLM 降级 fallback** — 主 LLM 重试耗尽后自动切 SiliconFlow Qwen 32B 单次兜底,解决 GLM 低 RPM 配额的间歇性限流 + - 文件:`backend/app/services/ontology_generator.py` + +- [x] **Qwen 32B 上下文爆修复** — `MAX_TEXT_LENGTH_FOR_LLM` 50000 → 28000 chars,保证 prompt+response < 32768 tokens + - 文件:`backend/app/services/ontology_generator.py` + +- [x] **Neo4j entity summary flatten** — monkey-patch `Neo4jEntityNodeOperations.save/save_bulk`,LLM 返回嵌套 dict 时自动取 `.value` 扁平化,防 Neo4j TypeError + - 文件:`backend/app/services/graphiti_client.py` + +- [x] **Frontend 无 pending state 路由修复** — `/process/new` 访问时无上传状态,自动 `router.replace({name:'Home'})` 不再死循环 + - 文件:`frontend/src/views/MainView.vue` + +- [x] **semaphore 100 → 30** — OASIS 模拟并发峰值降低,避免 GLM 限流雪崩 + - 文件:`backend/scripts/run_parallel_simulation.py` + +### v0.3 完成(本次大版本前的全部更新) #### 基础设施迁移 - [x] 从 Zep Cloud → 自托管 Graphiti + Neo4j(脱离外部 SaaS 依赖) @@ -578,3 +606,7 @@ FLASK_DEBUG=False | 2026-04-15 | 200 agents 设为甜点 | 8G 容量 + 95% 置信度足够 | | 2026-04-15 | 上线 Manus 式 replay UI | 给客户演示 + 复盘工具 | | 2026-04-15 | v0.4 路线图:国内平台 + SaaS | 用户战略需求 | +| 2026-04-15 | v0.3.1 hotfix:7 个 bug 修复 | 第一次完整 E2E 跑通压力测试 | +| 2026-04-15 | LLM client 加指数退避重试 + 双 LLM fallback | GLM RPM 配额低,retry 不够兜底 | +| 2026-04-15 | Replay UI 重构为 Manus cinematic 风格 | 三栏分析面板对客户演示不够"沉浸" | +| 2026-04-15 | semaphore 100 → 30 | 高并发触发 GLM 限流雪崩 | diff --git a/backend/app/services/graphiti_client.py b/backend/app/services/graphiti_client.py index 4e51f333..9ff6b7f6 100644 --- a/backend/app/services/graphiti_client.py +++ b/backend/app/services/graphiti_client.py @@ -27,6 +27,34 @@ def _safe_str(val): return str(val) +def _flatten_entity_property(value): + """ + Flatten a potentially nested dict/list value into a Neo4j-safe primitive. + + Priority: + 1. Primitive (str/int/float/bool/None) → return as-is + 2. dict with "value" key → recurse into value (handles LLM JSON-schema pollution) + 3. dict → json.dumps + 4. list → flatten each element recursively + 5. Other → str() + """ + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, dict): + if "value" in value: + return _flatten_entity_property(value["value"]) + import json + return json.dumps(value, ensure_ascii=False) + if isinstance(value, list): + return [_flatten_entity_property(v) for v in value] + return str(value) + + +def _sanitize_entity_data(entity_data: dict) -> dict: + """Sanitize all values in entity_data dict to Neo4j-safe primitives.""" + return {k: _flatten_entity_property(v) for k, v in entity_data.items()} + + from graphiti_core.embedder.client import EmbedderClient @@ -201,11 +229,79 @@ class GraphitiClient: embedder=embedder, cross_encoder=reranker, ) + + # Monkey-patch Neo4j entity node ops to sanitize LLM-polluted summary fields + # before they reach Neo4j (which rejects non-primitive property values). + self._patch_entity_node_ops() + # Build indices _run_async(self._graphiti.build_indices_and_constraints()) self._initialized = True logger.info("Graphiti core initialized with indices") + def _patch_entity_node_ops(self): + """ + Patch Neo4jEntityNodeOperations.save and save_bulk to sanitize entity_data + before writing to Neo4j. Handles the case where Qwen/other LLMs return + nested dicts (JSON-schema metadata) in fields like 'summary'. + """ + from graphiti_core.driver.neo4j.operations.entity_node_ops import Neo4jEntityNodeOperations + from graphiti_core.driver.driver import GraphProvider + from graphiti_core.driver.query_executor import QueryExecutor, Transaction + from graphiti_core.nodes import EntityNode + from graphiti_core.models.nodes.node_db_queries import ( + get_entity_node_save_query, + get_entity_node_save_bulk_query, + ) + from typing import Any + + _flatten = _sanitize_entity_data # closure over module-level helper + + original_save = Neo4jEntityNodeOperations.save + original_save_bulk = Neo4jEntityNodeOperations.save_bulk + + async def patched_save(self_ops, executor, node: EntityNode, tx=None): + entity_data: dict[str, Any] = { + 'uuid': node.uuid, + 'name': node.name, + 'name_embedding': node.name_embedding, + 'group_id': node.group_id, + 'summary': node.summary, + 'created_at': node.created_at, + } + entity_data.update(node.attributes or {}) + entity_data = _flatten(entity_data) + labels = ':'.join(list(set(node.labels + ['Entity']))) + query = get_entity_node_save_query(GraphProvider.NEO4J, labels) + if tx is not None: + await tx.run(query, entity_data=entity_data) + else: + await executor.execute_query(query, entity_data=entity_data) + + async def patched_save_bulk(self_ops, executor, nodes: list, tx=None, batch_size: int = 100): + prepared: list[dict[str, Any]] = [] + for node in nodes: + entity_data: dict[str, Any] = { + 'uuid': node.uuid, + 'name': node.name, + 'group_id': node.group_id, + 'summary': node.summary, + 'created_at': node.created_at, + 'name_embedding': node.name_embedding, + 'labels': list(set(node.labels + ['Entity'])), + } + entity_data.update(node.attributes or {}) + prepared.append(_flatten(entity_data)) + query = get_entity_node_save_bulk_query(GraphProvider.NEO4J, prepared) + if tx is not None: + await tx.run(query, nodes=prepared) + else: + await executor.execute_query(query, nodes=prepared) + + Neo4jEntityNodeOperations.save = patched_save + Neo4jEntityNodeOperations.save_bulk = patched_save_bulk + logger.info("Patched Neo4jEntityNodeOperations to sanitize LLM-polluted entity fields") + @classmethod def get_instance(cls) -> 'GraphitiClient': """Singleton accessor.""" diff --git a/backend/app/services/ontology_generator.py b/backend/app/services/ontology_generator.py index 8c630c41..b69dbc34 100644 --- a/backend/app/services/ontology_generator.py +++ b/backend/app/services/ontology_generator.py @@ -9,6 +9,7 @@ import re from typing import Dict, Any, List, Optional from ..utils.llm_client import LLMClient from ..utils.locale import get_language_instruction +from ..config import Config logger = logging.getLogger(__name__) @@ -181,6 +182,17 @@ class OntologyGenerator: def __init__(self, llm_client: Optional[LLMClient] = None): self.llm_client = llm_client or LLMClient() + # Fallback LLM: SiliconFlow Qwen 32B(高 RPM,GLM 限流时兜底) + self._fallback_llm: Optional[LLMClient] = None + try: + if Config.GRAPHITI_LLM_API_KEY: + self._fallback_llm = LLMClient( + api_key=Config.GRAPHITI_LLM_API_KEY, + base_url=Config.GRAPHITI_LLM_BASE_URL, + model=Config.GRAPHITI_LLM_MODEL, + ) + except Exception as e: + logger.warning(f"Fallback LLM 初始化失败: {e}") def generate( self, @@ -213,20 +225,35 @@ class OntologyGenerator: {"role": "user", "content": user_message} ] - # 调用LLM - result = self.llm_client.chat_json( - messages=messages, - temperature=0.3, - max_tokens=4096 - ) + # 调用 LLM(主 LLM 重试 5 次全失败后,自动降级到 fallback) + try: + result = self.llm_client.chat_json( + messages=messages, + temperature=0.3, + max_tokens=4096 + ) + except Exception as primary_err: + if not self._fallback_llm: + raise + logger.warning( + f"主 LLM (GLM) 生成本体失败,降级到 Qwen 32B fallback: " + f"{type(primary_err).__name__}: {str(primary_err)[:200]}" + ) + result = self._fallback_llm.chat_json( + messages=messages, + temperature=0.3, + max_tokens=4096 + ) # 验证和后处理 result = self._validate_and_process(result) return result - # 传给 LLM 的文本最大长度(5万字) - MAX_TEXT_LENGTH_FOR_LLM = 50000 + # 传给 LLM 的文本最大长度(2.8万字) + # Qwen 32768 token 窗口:system prompt ~2K tokens + 响应 ~4K tokens + # 留给文档的安全预算 ≈ 26K tokens,保守按 1 char ≈ 1 token 估算取 28000 chars + MAX_TEXT_LENGTH_FOR_LLM = 28000 def _build_user_message( self, @@ -243,7 +270,7 @@ class OntologyGenerator: # 如果文本超过5万字,截断(仅影响传给LLM的内容,不影响图谱构建) if len(combined_text) > self.MAX_TEXT_LENGTH_FOR_LLM: combined_text = combined_text[:self.MAX_TEXT_LENGTH_FOR_LLM] - combined_text += f"\n\n...(原文共{original_length}字,已截取前{self.MAX_TEXT_LENGTH_FOR_LLM}字用于本体分析)..." + combined_text += f"\n\n[...文档已截断以适应 LLM 上下文窗口。原文共 {original_length} 字,已截取前 {self.MAX_TEXT_LENGTH_FOR_LLM} 字用于本体分析...]" message = f"""## 模拟需求 diff --git a/backend/app/utils/llm_client.py b/backend/app/utils/llm_client.py index b4f79faa..9477cbf6 100644 --- a/backend/app/utils/llm_client.py +++ b/backend/app/utils/llm_client.py @@ -5,11 +5,41 @@ LLM客户端封装 import json import re +import time +import random +import logging from typing import Optional, Dict, Any, List -from openai import OpenAI +from openai import OpenAI, RateLimitError, APIError, APIConnectionError, APITimeoutError from ..config import Config +logger = logging.getLogger('foresight.llm_client') + +# 重试配置(针对 429 / 5xx / 超时 / 连接错误) +_MAX_RETRIES = 5 +_BASE_BACKOFF = 1.0 # 首次重试等 1s +_MAX_BACKOFF = 30.0 # 单次最多等 30s + + +def _is_rate_limit_error(err: Exception) -> bool: + """判断是否是速率限制/可重试错误""" + if isinstance(err, (RateLimitError, APIConnectionError, APITimeoutError)): + return True + if isinstance(err, APIError): + # 429 / 500 / 502 / 503 / 504 都可重试 + status = getattr(err, "status_code", None) or getattr(err, "code", None) + try: + status = int(status) if status else None + except (ValueError, TypeError): + status = None + if status in (429, 500, 502, 503, 504): + return True + # 智谱 1302 = 速率限制 + msg = str(err) + if "1302" in msg or "rate limit" in msg.lower() or "速率限制" in msg or "too many request" in msg.lower(): + return True + return False + class LLMClient: """LLM客户端""" @@ -61,7 +91,28 @@ class LLMClient: if response_format: kwargs["response_format"] = response_format - response = self.client.chat.completions.create(**kwargs) + # 带指数退避的重试:处理 429/5xx/超时等可恢复错误 + response = None + last_err = None + for attempt in range(_MAX_RETRIES + 1): + try: + response = self.client.chat.completions.create(**kwargs) + break + except Exception as err: + last_err = err + if attempt >= _MAX_RETRIES or not _is_rate_limit_error(err): + raise + # 指数退避 + 随机抖动 + backoff = min(_MAX_BACKOFF, _BASE_BACKOFF * (2 ** attempt)) + backoff += random.uniform(0, backoff * 0.3) + logger.warning( + f"LLM 调用遇到可重试错误 (attempt {attempt + 1}/{_MAX_RETRIES}): " + f"{type(err).__name__}: {str(err)[:200]}. 退避 {backoff:.1f}s 后重试..." + ) + time.sleep(backoff) + if response is None: + raise last_err or RuntimeError("LLM 调用失败(未知原因)") + # Token 追踪(v0.3 新增):记录每次调用的 token 消耗,按当前 stage 归类 try: from . import token_tracker diff --git a/backend/scripts/run_parallel_simulation.py b/backend/scripts/run_parallel_simulation.py index 0eeb527f..ef99677f 100644 --- a/backend/scripts/run_parallel_simulation.py +++ b/backend/scripts/run_parallel_simulation.py @@ -1156,7 +1156,7 @@ async def run_twitter_simulation( agent_graph=result.agent_graph, platform=oasis.DefaultPlatformType.TWITTER, database_path=db_path, - semaphore=100, # 限制最大并发 LLM 请求数,防止 API 过载 + semaphore=30, # 限制最大并发 LLM 请求数,防止 API 过载(GLM RPM 友好值) ) await result.env.reset() @@ -1347,7 +1347,7 @@ async def run_reddit_simulation( agent_graph=result.agent_graph, platform=oasis.DefaultPlatformType.REDDIT, database_path=db_path, - semaphore=100, # 限制最大并发 LLM 请求数,防止 API 过载 + semaphore=30, # 限制最大并发 LLM 请求数,防止 API 过载(GLM RPM 友好值) ) await result.env.reset() diff --git a/frontend/src/views/MainView.vue b/frontend/src/views/MainView.vue index 93a13433..7772f728 100644 --- a/frontend/src/views/MainView.vue +++ b/frontend/src/views/MainView.vue @@ -201,8 +201,13 @@ const initProject = async () => { const handleNewProject = async () => { const pending = getPendingUpload() if (!pending.isPending || pending.files.length === 0) { - error.value = 'No pending files found.' - addLog('Error: No pending files found for new project.') + error.value = 'No pending files found. Redirecting to home to upload documents...' + addLog('No pending upload detected for /process/new. Redirecting to Home.') + // Direct-visit / refresh case: pending state is empty because it's in-memory only. + // Send user back to Home so they can upload files and start a real project. + setTimeout(() => { + router.replace({ name: 'Home' }) + }, 800) return } diff --git a/frontend/src/views/SimulationReplayView.vue b/frontend/src/views/SimulationReplayView.vue index 489b694a..33f8ceb5 100644 --- a/frontend/src/views/SimulationReplayView.vue +++ b/frontend/src/views/SimulationReplayView.vue @@ -1,36 +1,151 @@