From 054d056182f390f72f919a3ba1d4c967b373f9db Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Wed, 11 Mar 2026 00:59:37 +0800 Subject: [PATCH] =?UTF-8?q?=E9=AA=8C=E8=AF=81=E5=BD=93=E5=89=8D=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B=E5=8F=AF=E7=94=A8=E6=80=A7=E3=80=8D}=E0=AA=BE?= =?UTF-8?q?=E0=AA=A1=E0=AB=80=20Reviewing=20format=3F=20no?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 25 +-- CLAUDE.md | 4 +- .../20260311_add_kb_chunk_ivfflat_index.py | 42 +++++ .../20260311_add_pg_trgm_text_indexes.py | 76 +++++++++ backend/app/services/adaptive_rag_service.py | 148 +++++++++++++++++- backend/app/services/llm.py | 21 ++- docker-compose.dev.yml | 13 +- docker-compose.yml | 13 +- 8 files changed, 312 insertions(+), 30 deletions(-) create mode 100644 backend/alembic/versions/20260311_add_kb_chunk_ivfflat_index.py create mode 100644 backend/alembic/versions/20260311_add_pg_trgm_text_indexes.py diff --git a/.env.example b/.env.example index e91ace4d..bf87aa9a 100644 --- a/.env.example +++ b/.env.example @@ -11,27 +11,30 @@ FRONTEND_PORT=5174 # Backend Runtime CORS_ORIGINS=http://localhost:5174,http://127.0.0.1:5174 -LLM_PROVIDER=openai -LLM_FALLBACK_PROVIDER=ollama -OPENAI_BASE_URL=http://openclaw:18789/v1 +LLM_PROVIDER=glm +LLM_FALLBACK_PROVIDER=glm +OPENAI_BASE_URL=https://api.edgefn.net/v1 OPENAI_API_KEY= -OPENCLAW_ROUTING_ENABLED=true +OPENAI_MODEL=MiniMax-M2.5 +OPENCLAW_ROUTING_ENABLED=false OPENCLAW_ROUTING_RULES_PATH=/app/backend/.openclaw/pmflow-routing-rules.json OPENCLAW_AGENTS_DEFAULT_TIMEOUT_SECONDS=120 -LLM_MIXED_ROUTING_ENABLED=false +LLM_MIXED_ROUTING_ENABLED=true LLM_LOW_TO_HIGH_ESCALATION_ENABLED=true -LLM_ROUTING_LOW_PROVIDER=openai -LLM_ROUTING_LOW_MODEL=zai/glm-4-flash +LLM_ROUTING_LOW_PROVIDER=glm +LLM_ROUTING_LOW_MODEL=GLM-4.7 LLM_ROUTING_HIGH_PROVIDER=openai -LLM_ROUTING_HIGH_MODEL=minimax/MiniMax-M2.5 -LLM_ROUTING_SAFETY_PROVIDER=openai -LLM_ROUTING_SAFETY_MODEL=deepseek/deepseek-chat -LLM_DIRECT_EMERGENCY_ENABLED=false +LLM_ROUTING_HIGH_MODEL=MiniMax-M2.5 +LLM_ROUTING_SAFETY_PROVIDER=glm +LLM_ROUTING_SAFETY_MODEL=GLM-4.7 +LLM_DIRECT_EMERGENCY_ENABLED=true RAG_DEEP_RETRIEVAL_STEPS=5 RAG_DEEP_VECTOR_TOP_K=12 RAG_DEEP_CONTEXT_BLOCKS=16 RAG_DEEP_ANSWER_MAX_TOKENS=1400 GLM_API_KEY= +GLM_BASE_URL=https://api.edgefn.net/v1 +GLM_MODEL=GLM-4.7 DEEPSEEK_API_KEY= OLLAMA_HOST=http://host.docker.internal:11434 OLLAMA_BASE_URL=http://host.docker.internal:11434 diff --git a/CLAUDE.md b/CLAUDE.md index d0415c61..fcf1019e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,7 +40,9 @@ backend/ │ └── utils/ # 工具函数 ├── alembic/ # 数据库迁移 │ └── versions/ -│ └── 20260310_add_feishu_outbox_reply_hash.py # 飞书 outbox 回执哈希幂等字段 +│ ├── 20260310_add_feishu_outbox_reply_hash.py # 飞书 outbox 回执哈希幂等字段 +│ ├── 20260311_add_pg_trgm_text_indexes.py # pg_trgm 扩展与文本检索索引 +│ └── 20260311_add_kb_chunk_ivfflat_index.py # knowledge_base_chunks 向量 IVFFlat 索引 ├── requirements.txt # Python 依赖 ├── Dockerfile └── tests/ diff --git a/backend/alembic/versions/20260311_add_kb_chunk_ivfflat_index.py b/backend/alembic/versions/20260311_add_kb_chunk_ivfflat_index.py new file mode 100644 index 00000000..87238b1f --- /dev/null +++ b/backend/alembic/versions/20260311_add_kb_chunk_ivfflat_index.py @@ -0,0 +1,42 @@ +"""add ivfflat index for knowledge chunk embedding + +Revision ID: add_kb_chunk_ivfflat_index +Revises: add_pg_trgm_text_indexes +Create Date: 2026-03-11 +""" + +from __future__ import annotations + +from alembic import op + + +revision = "add_kb_chunk_ivfflat_index" +down_revision = "add_pg_trgm_text_indexes" +branch_labels = None +depends_on = None + + +def _is_postgresql() -> bool: + bind = op.get_bind() + return str(getattr(getattr(bind, "dialect", None), "name", "")).lower().startswith("postgres") + + +def upgrade() -> None: + if not _is_postgresql(): + return + + op.execute( + """ + CREATE INDEX IF NOT EXISTS ix_knowledge_chunks_embedding_ivfflat + ON knowledge_base_chunks + USING ivfflat (embedding vector_cosine_ops) + WITH (lists = 100) + """ + ) + + +def downgrade() -> None: + if not _is_postgresql(): + return + + op.execute("DROP INDEX IF EXISTS ix_knowledge_chunks_embedding_ivfflat") diff --git a/backend/alembic/versions/20260311_add_pg_trgm_text_indexes.py b/backend/alembic/versions/20260311_add_pg_trgm_text_indexes.py new file mode 100644 index 00000000..53dd0f46 --- /dev/null +++ b/backend/alembic/versions/20260311_add_pg_trgm_text_indexes.py @@ -0,0 +1,76 @@ +"""add pg_trgm text indexes for project search + +Revision ID: add_pg_trgm_text_indexes +Revises: add_feishu_outbox_reply_hash +Create Date: 2026-03-11 +""" + +from __future__ import annotations + +from alembic import op + + +revision = "add_pg_trgm_text_indexes" +down_revision = "add_feishu_outbox_reply_hash" +branch_labels = None +depends_on = None + + +def _is_postgresql() -> bool: + bind = op.get_bind() + return str(getattr(getattr(bind, "dialect", None), "name", "")).lower().startswith("postgres") + + +def upgrade() -> None: + if not _is_postgresql(): + return + + op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm") + op.execute( + """ + CREATE INDEX IF NOT EXISTS ix_projects_name_trgm + ON projects USING GIN (name gin_trgm_ops) + """ + ) + op.execute( + """ + CREATE INDEX IF NOT EXISTS ix_projects_title_trgm + ON projects USING GIN (title gin_trgm_ops) + """ + ) + op.execute( + """ + CREATE INDEX IF NOT EXISTS ix_projects_description_trgm + ON projects USING GIN (description gin_trgm_ops) + """ + ) + op.execute( + """ + CREATE INDEX IF NOT EXISTS ix_signals_title_trgm + ON signals USING GIN (title gin_trgm_ops) + """ + ) + op.execute( + """ + CREATE INDEX IF NOT EXISTS ix_signals_content_trgm + ON signals USING GIN (content gin_trgm_ops) + """ + ) + op.execute( + """ + CREATE INDEX IF NOT EXISTS ix_tasks_title_trgm + ON tasks USING GIN (title gin_trgm_ops) + """ + ) + + +def downgrade() -> None: + if not _is_postgresql(): + return + + op.execute("DROP INDEX IF EXISTS ix_tasks_title_trgm") + op.execute("DROP INDEX IF EXISTS ix_signals_content_trgm") + op.execute("DROP INDEX IF EXISTS ix_signals_title_trgm") + op.execute("DROP INDEX IF EXISTS ix_projects_description_trgm") + op.execute("DROP INDEX IF EXISTS ix_projects_title_trgm") + op.execute("DROP INDEX IF EXISTS ix_projects_name_trgm") diff --git a/backend/app/services/adaptive_rag_service.py b/backend/app/services/adaptive_rag_service.py index e1fb526d..8c5a413f 100644 --- a/backend/app/services/adaptive_rag_service.py +++ b/backend/app/services/adaptive_rag_service.py @@ -24,7 +24,7 @@ from datetime import datetime, timezone from typing import Any, Optional import httpx -from sqlalchemy import and_, delete, func, or_, select +from sqlalchemy import and_, bindparam, delete, func, or_, select, text from sqlalchemy.ext.asyncio import AsyncSession from ..config.ai_prompts import PROMPT_RAG_ANSWER, PROMPT_RAG_DB_PROJECT_QA, PROMPT_RAG_INTENT_CLASSIFICATION @@ -4490,7 +4490,147 @@ class AdaptiveRAGService: source_types = self._map_domains_to_source_types(domains) if not source_types: return [] + limit = max(1, min(int(top_k or DEFAULT_TOP_K), 40)) + if self._is_postgresql_session(): + pg_rows = await self._retrieve_vector_context_pg( + question_embedding=question_embedding, + source_types=source_types, + project_id=project_id, + source_chat_id=source_chat_id, + limit=limit, + ) + if pg_rows is not None: + return pg_rows + return await self._retrieve_vector_context_legacy( + question_embedding=question_embedding, + source_types=source_types, + project_id=project_id, + source_chat_id=source_chat_id, + limit=limit, + ) + + def _is_postgresql_session(self) -> bool: + try: + bind = self.db.get_bind() + except Exception: + return False + dialect_name = str(getattr(getattr(bind, "dialect", None), "name", "")).lower() + return dialect_name.startswith("postgres") + + @staticmethod + def _vector_literal(values: list[float]) -> str: + normalized: list[str] = [] + for raw in values: + try: + normalized.append(f"{float(raw):.8f}") + except Exception: + normalized.append("0.0") + return "[" + ",".join(normalized) + "]" + + async def _retrieve_vector_context_pg( + self, + *, + question_embedding: list[float], + source_types: list[str], + project_id: Optional[int], + source_chat_id: Optional[str], + limit: int, + ) -> Optional[list[RetrievedItem]]: + query_vec = self._vector_literal(question_embedding) + scoped_project_id = int(project_id) if project_id is not None else 0 + scoped_source_chat_id = str(source_chat_id or "").strip() + stmt = text( + """ + SELECT + id, + source_type, + content, + project_id, + "metadata" AS chunk_metadata, + 1 - (embedding <=> CAST(:query_vec AS vector)) AS similarity + FROM knowledge_base_chunks + WHERE tenant_id = :tenant_id + AND source_type IN :source_types + AND embedding IS NOT NULL + AND ( + :project_scope = FALSE + OR (source_type IN ('feishu', 'pmflow') AND project_id = :project_id) + OR (source_type NOT IN ('feishu', 'pmflow') AND (project_id = :project_id OR project_id IS NULL)) + ) + AND ( + :source_chat_id = '' + OR source_type <> 'feishu' + OR COALESCE("metadata"->>'chat_id', '') = '' + OR "metadata"->>'chat_id' = :source_chat_id + ) + ORDER BY embedding <=> CAST(:query_vec AS vector) + LIMIT :limit + """ + ).bindparams(bindparam("source_types", expanding=True)) + try: + rows = ( + await self.db.execute( + stmt, + { + "tenant_id": self.tenant_id, + "source_types": list(source_types), + "project_scope": project_id is not None, + "project_id": scoped_project_id, + "source_chat_id": scoped_source_chat_id, + "query_vec": query_vec, + "limit": int(limit), + }, + ) + ).mappings().all() + except Exception: + return None + + scored: list[RetrievedItem] = [] + for row in rows: + source_type = str(row.get("source_type") or "doc") + meta = self._safe_dict(row.get("chunk_metadata")).copy() + chunk_project_id_raw = row.get("project_id") + chunk_project_id = int(chunk_project_id_raw) if chunk_project_id_raw is not None else None + if chunk_project_id is not None: + meta.setdefault("project_id", chunk_project_id) + + if source_type == "pmflow": + target_url = str(meta.get("target_url") or "").strip() + if not target_url: + target_url = self._build_pmflow_target_url( + record_type=str(meta.get("record_type") or ""), + record_id=meta.get("record_id"), + project_id=meta.get("project_id"), + ) + if target_url: + meta["target_url"] = target_url + + score = self._to_float(row.get("similarity"), 0.0) + if score <= 0: + continue + label = self._source_label(source_type, meta) + scored.append( + RetrievedItem( + source=source_type, + content=str(row.get("content") or "")[:1200], + score=score, + source_label=label, + metadata=meta, + ) + ) + scored.sort(key=lambda x: x.score, reverse=True) + return scored[:limit] + + async def _retrieve_vector_context_legacy( + self, + *, + question_embedding: list[float], + source_types: list[str], + project_id: Optional[int], + source_chat_id: Optional[str], + limit: int, + ) -> list[RetrievedItem]: stmt = select(KnowledgeBaseChunk).where( and_( KnowledgeBaseChunk.tenant_id == self.tenant_id, @@ -4498,7 +4638,8 @@ class AdaptiveRAGService: ) ) - chunks = (await self.db.execute(stmt.limit(500))).scalars().all() + scan_limit = max(120, min(1200, limit * 20)) + chunks = (await self.db.execute(stmt.limit(scan_limit))).scalars().all() scored: list[RetrievedItem] = [] for chunk in chunks: meta = self._safe_dict(chunk.chunk_metadata).copy() @@ -4507,8 +4648,6 @@ class AdaptiveRAGService: meta.setdefault("project_id", chunk_project_id) if project_id is not None: - # 防止跨项目泄漏：Feishu/PMFlow 来源必须严格匹配项目组； - # 文档/URL 来源可读取全局（project_id 为空）和当前项目数据。 if chunk.source_type in {"feishu", "pmflow"}: if chunk_project_id != int(project_id): continue @@ -4543,7 +4682,6 @@ class AdaptiveRAGService: ) ) scored.sort(key=lambda x: x.score, reverse=True) - limit = max(1, min(int(top_k or DEFAULT_TOP_K), 40)) return scored[:limit] # ------------------------------------------------------------------ diff --git a/backend/app/services/llm.py b/backend/app/services/llm.py index d6cef769..5c384601 100644 --- a/backend/app/services/llm.py +++ b/backend/app/services/llm.py @@ -13,6 +13,7 @@ import asyncio from dataclasses import dataclass import hashlib import json +import logging import os import re import time @@ -38,6 +39,8 @@ except Exception: # pragma: no cover REDIS_AVAILABLE = False redis_async = None +logger = logging.getLogger(__name__) + SUPPORTED_PROVIDERS = {"deepseek", "glm", "openai", "ollama"} OPENCLAW_PRIMARY_GATEWAY_URL = "http://openclaw:18789/v1" OPENCLAW_DEFAULT_LOW_MODEL = "zai/glm-4-flash" @@ -626,6 +629,7 @@ class LLMService: return await self._generate_structured_with_db(db, prompt, **call_kwargs) async def _generate_with_db(self, db: AsyncSession, prompt: str, **kwargs: Any) -> str: + started_at = time.perf_counter() tenant_id = int(kwargs.get("tenant_id") or 1) function_name = str(kwargs.get("function_name") or "llm_generate") temperature = float(kwargs.get("temperature") or 0.2) @@ -639,6 +643,20 @@ class LLMService: prompt_profile = {"role_name": "", "system_prompt": ""} merged_prompt = self._merge_prompt_with_profile(prompt=prompt, prompt_profile=prompt_profile) safety_attempted = False + + def _log_stub_failure(error_obj: Exception) -> None: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + provider = str(engine_meta.get("provider") or "") + model = str(engine_meta.get("model") or "") + logger.warning( + "[LLM_FAIL] 生成失败 → 兜底 stub function=%s provider=%s model=%s error=%r duration_ms=%.0f", + function_name, + provider, + model, + error_obj, + elapsed_ms, + ) + try: if use_prompt_profile: prompt_profile = await self._resolve_active_prompt_profile(db, tenant_id=tenant_id) @@ -743,6 +761,7 @@ class LLMService: status = "failed_fast" response_text = self._stub_response(prompt) _runtime_log("LLM", f"生成快速失败，已跳过升级重试：{exc}") + _log_stub_failure(exc) return response_text if self._is_low_tier_function(function_name): if self._prefer_fast_local_fallback(function_name): @@ -850,7 +869,7 @@ class LLMService: status = "failed" response_text = self._stub_response(prompt) - _runtime_log("LLM", f"生成失败，已返回兜底 stub：{exc}") + _log_stub_failure(upgraded_exc) return response_text finally: try: diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index b50dfde0..0da93de1 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -65,19 +65,20 @@ services: OPENCLAW_ROUTING_RULES_PATH: ${OPENCLAW_ROUTING_RULES_PATH:-/app/backend/.openclaw/pmflow-routing-rules.json} LLM_MIXED_ROUTING_ENABLED: ${LLM_MIXED_ROUTING_ENABLED:-false} LLM_LOW_TO_HIGH_ESCALATION_ENABLED: ${LLM_LOW_TO_HIGH_ESCALATION_ENABLED:-true} - LLM_ROUTING_LOW_PROVIDER: ${LLM_ROUTING_LOW_PROVIDER:-openai} - LLM_ROUTING_LOW_MODEL: ${LLM_ROUTING_LOW_MODEL:-zai/glm-4-flash} + LLM_ROUTING_LOW_PROVIDER: ${LLM_ROUTING_LOW_PROVIDER:-glm} + LLM_ROUTING_LOW_MODEL: ${LLM_ROUTING_LOW_MODEL:-GLM-4.7} LLM_ROUTING_HIGH_PROVIDER: ${LLM_ROUTING_HIGH_PROVIDER:-openai} - LLM_ROUTING_HIGH_MODEL: ${LLM_ROUTING_HIGH_MODEL:-minimax/MiniMax-M2.5} - LLM_ROUTING_SAFETY_PROVIDER: ${LLM_ROUTING_SAFETY_PROVIDER:-openai} - LLM_ROUTING_SAFETY_MODEL: ${LLM_ROUTING_SAFETY_MODEL:-deepseek/deepseek-chat} + LLM_ROUTING_HIGH_MODEL: ${LLM_ROUTING_HIGH_MODEL:-MiniMax-M2.5} + LLM_ROUTING_SAFETY_PROVIDER: ${LLM_ROUTING_SAFETY_PROVIDER:-glm} + LLM_ROUTING_SAFETY_MODEL: ${LLM_ROUTING_SAFETY_MODEL:-GLM-4.7} LLM_DIRECT_EMERGENCY_ENABLED: ${LLM_DIRECT_EMERGENCY_ENABLED:-false} LLM_REQUEST_TIMEOUT: ${LLM_REQUEST_TIMEOUT:-90} AUTO_CREATE_TABLES: "true" GLM_API_KEY: ${GLM_API_KEY:-} GLM_BASE_URL: ${GLM_BASE_URL:-https://open.bigmodel.cn/api/paas/v4} - GLM_MODEL: ${GLM_MODEL:-glm-4-flash} + GLM_MODEL: ${GLM_MODEL:-GLM-4.7} OPENAI_API_KEY: ${OPENAI_API_KEY:-} + OPENAI_MODEL: ${OPENAI_MODEL:-MiniMax-M2.5} DEEPSEEK_API_KEY: ${DEEPSEEK_API_KEY:-} OLLAMA_HOST: ${OLLAMA_HOST:-http://host.docker.internal:11434} OLLAMA_BASE_URL: ${OLLAMA_BASE_URL:-http://host.docker.internal:11434} diff --git a/docker-compose.yml b/docker-compose.yml index 2520e8eb..de5889d5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -65,19 +65,20 @@ services: OPENCLAW_ROUTING_RULES_PATH: ${OPENCLAW_ROUTING_RULES_PATH:-/app/backend/.openclaw/pmflow-routing-rules.json} LLM_MIXED_ROUTING_ENABLED: ${LLM_MIXED_ROUTING_ENABLED:-false} LLM_LOW_TO_HIGH_ESCALATION_ENABLED: ${LLM_LOW_TO_HIGH_ESCALATION_ENABLED:-true} - LLM_ROUTING_LOW_PROVIDER: ${LLM_ROUTING_LOW_PROVIDER:-openai} - LLM_ROUTING_LOW_MODEL: ${LLM_ROUTING_LOW_MODEL:-zai/glm-4-flash} + LLM_ROUTING_LOW_PROVIDER: ${LLM_ROUTING_LOW_PROVIDER:-glm} + LLM_ROUTING_LOW_MODEL: ${LLM_ROUTING_LOW_MODEL:-GLM-4.7} LLM_ROUTING_HIGH_PROVIDER: ${LLM_ROUTING_HIGH_PROVIDER:-openai} - LLM_ROUTING_HIGH_MODEL: ${LLM_ROUTING_HIGH_MODEL:-minimax/MiniMax-M2.5} - LLM_ROUTING_SAFETY_PROVIDER: ${LLM_ROUTING_SAFETY_PROVIDER:-openai} - LLM_ROUTING_SAFETY_MODEL: ${LLM_ROUTING_SAFETY_MODEL:-deepseek/deepseek-chat} + LLM_ROUTING_HIGH_MODEL: ${LLM_ROUTING_HIGH_MODEL:-MiniMax-M2.5} + LLM_ROUTING_SAFETY_PROVIDER: ${LLM_ROUTING_SAFETY_PROVIDER:-glm} + LLM_ROUTING_SAFETY_MODEL: ${LLM_ROUTING_SAFETY_MODEL:-GLM-4.7} LLM_DIRECT_EMERGENCY_ENABLED: ${LLM_DIRECT_EMERGENCY_ENABLED:-false} LLM_REQUEST_TIMEOUT: ${LLM_REQUEST_TIMEOUT:-90} AUTO_CREATE_TABLES: "true" GLM_API_KEY: ${GLM_API_KEY:-} GLM_BASE_URL: ${GLM_BASE_URL:-https://open.bigmodel.cn/api/paas/v4} - GLM_MODEL: ${GLM_MODEL:-glm-4-flash} + GLM_MODEL: ${GLM_MODEL:-GLM-4.7} OPENAI_API_KEY: ${OPENAI_API_KEY:-} + OPENAI_MODEL: ${OPENAI_MODEL:-MiniMax-M2.5} DEEPSEEK_API_KEY: ${DEEPSEEK_API_KEY:-} OLLAMA_HOST: ${OLLAMA_HOST:-http://host.docker.internal:11434} OLLAMA_BASE_URL: ${OLLAMA_BASE_URL:-http://host.docker.internal:11434}