|
|
@@ -0,0 +1,1334 @@
|
|
|
+"""
|
|
|
+api.v1.knowledge — Knowledge Base management endpoints.
|
|
|
+
|
|
|
+GET /api/v1/knowledge List all KBs
|
|
|
+POST /api/v1/knowledge Create KB (name, root_dir)
|
|
|
+DELETE /api/v1/knowledge/{kb_id} Delete KB
|
|
|
+GET /api/v1/knowledge/{kb_id}/files List root_dir file tree
|
|
|
+POST /api/v1/knowledge/{kb_id}/files Add files to KB
|
|
|
+DELETE /api/v1/knowledge/{kb_id}/files/{file_id} Remove file from KB
|
|
|
+POST /api/v1/knowledge/{kb_id}/index Trigger index build job
|
|
|
+POST /api/v1/knowledge/{kb_id}/wiki Trigger wiki compile job
|
|
|
+PUT /api/v1/knowledge/{kb_id}/wiki/pause Pause/resume wiki compile
|
|
|
+GET /api/v1/knowledge/{kb_id}/wiki/tree Get wiki file tree
|
|
|
+GET /api/v1/knowledge/{kb_id}/wiki/{page_path} Get wiki page content
|
|
|
+POST /api/v1/knowledge/{kb_id}/search Search KB
|
|
|
+GET /api/v1/knowledge/{kb_id}/search/stream SSE streaming QA
|
|
|
+GET /api/v1/knowledge/{kb_id}/jobs List jobs for KB
|
|
|
+GET /api/v1/knowledge/{kb_id}/jobs/{job_id} Get job status + log
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+import asyncio
|
|
|
+import json
|
|
|
+import os
|
|
|
+import subprocess
|
|
|
+import sys
|
|
|
+import tempfile
|
|
|
+import time
|
|
|
+import threading
|
|
|
+from pathlib import Path
|
|
|
+from typing import Any, Dict, List, Optional, Set
|
|
|
+
|
|
|
+from fastapi import APIRouter, Depends, HTTPException, Query
|
|
|
+from fastapi.responses import StreamingResponse
|
|
|
+from pydantic import BaseModel, Field
|
|
|
+
|
|
|
+from agentpaas.api.deps import get_tenant, get_database
|
|
|
+from agentpaas.api.errors import api_error
|
|
|
+from agentpaas.api.middleware.auth import TenantContext
|
|
|
+from agentpaas.db.models import Database, gen_id, now_utc
|
|
|
+
|
|
|
+router = APIRouter(prefix="/knowledge", tags=["knowledge"])
|
|
|
+
|
|
|
+# ─── Supported file extensions ───
|
|
|
+SUPPORTED_EXTS = {".pdf", ".txt", ".docx", ".md", ".html", ".htm", ".csv"}
|
|
|
+
|
|
|
+
|
|
|
+# ─── Pydantic models ───
|
|
|
+
|
|
|
+class CreateKBRequest(BaseModel):
|
|
|
+ name: str = Field(..., max_length=256)
|
|
|
+ root_dir: str = Field(..., description="Absolute path to knowledge base directory")
|
|
|
+ description: str = Field(default="", max_length=2048)
|
|
|
+
|
|
|
+
|
|
|
+class AddFilesRequest(BaseModel):
|
|
|
+ file_paths: List[str] = Field(..., description="Absolute paths of files to add")
|
|
|
+
|
|
|
+
|
|
|
+class IndexBuildRequest(BaseModel):
|
|
|
+ index_type: str = Field(default="bm25", description="bm25 | vector | graph | all")
|
|
|
+ mode: str = Field(default="full", description="full | incremental")
|
|
|
+
|
|
|
+
|
|
|
+class WikiBuildRequest(BaseModel):
|
|
|
+ mode: str = Field(default="incremental", description="incremental | rebuild")
|
|
|
+ concurrency: int = Field(default=2, ge=1, le=8)
|
|
|
+
|
|
|
+
|
|
|
+class SearchRequest(BaseModel):
|
|
|
+ query: str = Field(..., max_length=4096)
|
|
|
+ mode: str = Field(default="lambda", description="lambda | bm25 | vector | wiki")
|
|
|
+ top_k: int = Field(default=5, ge=1, le=20)
|
|
|
+ weights: Optional[Dict[str, float]] = None # bm25/vector/graph/wiki overrides
|
|
|
+
|
|
|
+
|
|
|
+class UpdateIgnoreDirsRequest(BaseModel):
|
|
|
+ ignored_dirs: List[str] = Field(default_factory=list, description="List of relative dir paths to ignore")
|
|
|
+
|
|
|
+
|
|
|
+# ─── Helper: walk root_dir ───
|
|
|
+
|
|
|
+def _walk_dir(root_dir: str, added_paths: set, ignored_dirs: Optional[Set[str]] = None) -> List[Dict]:
|
|
|
+ """
|
|
|
+ Recursively list files under root_dir, skipping ignored directories.
|
|
|
+ ignored_dirs: set of relative paths (e.g. {'processed', 'raw', 'wiki'}).
|
|
|
+ Both bare names ('raw') and nested paths ('data/raw') are matched.
|
|
|
+ Returns up to 2000 file entries.
|
|
|
+ """
|
|
|
+ ignored = ignored_dirs or set()
|
|
|
+ items: List[Dict] = []
|
|
|
+ root = Path(root_dir)
|
|
|
+ if not root.is_dir():
|
|
|
+ return items
|
|
|
+
|
|
|
+ for dirpath, dirnames, filenames in os.walk(root_dir, topdown=True):
|
|
|
+ rel_dir = Path(dirpath).relative_to(root)
|
|
|
+ parts = rel_dir.parts # () for root itself
|
|
|
+
|
|
|
+ # If any ancestor or the current dir matches the ignore list, skip entire subtree
|
|
|
+ if parts:
|
|
|
+ ignored_hit = any(
|
|
|
+ parts[i] in ignored or "/".join(parts[:i + 1]) in ignored
|
|
|
+ for i in range(len(parts))
|
|
|
+ )
|
|
|
+ if ignored_hit:
|
|
|
+ dirnames.clear() # prune: don't recurse further
|
|
|
+ continue
|
|
|
+
|
|
|
+ # Prune child dirs that match ignore list before recursing
|
|
|
+ dirnames[:] = sorted(
|
|
|
+ d for d in dirnames
|
|
|
+ if d not in ignored and (
|
|
|
+ "/".join((*parts, d)) if parts else d
|
|
|
+ ) not in ignored
|
|
|
+ )
|
|
|
+
|
|
|
+ for fname in sorted(filenames):
|
|
|
+ p = Path(dirpath) / fname
|
|
|
+ if p.suffix.lower() not in SUPPORTED_EXTS:
|
|
|
+ continue
|
|
|
+ rel_file = str(p.relative_to(root))
|
|
|
+ try:
|
|
|
+ stat = p.stat()
|
|
|
+ except OSError:
|
|
|
+ continue
|
|
|
+ items.append({
|
|
|
+ "path": str(p),
|
|
|
+ "relative_path": rel_file,
|
|
|
+ "name": fname,
|
|
|
+ "type": p.suffix.lower().lstrip("."),
|
|
|
+ "size": stat.st_size,
|
|
|
+ "modified": stat.st_mtime,
|
|
|
+ "added": str(p) in added_paths,
|
|
|
+ })
|
|
|
+ if len(items) >= 2000:
|
|
|
+ return items
|
|
|
+
|
|
|
+ return items
|
|
|
+
|
|
|
+
|
|
|
+def _list_dir_tree(root_dir: str, ignored: Set[str], max_depth: int = 8) -> List[Dict]:
|
|
|
+ """
|
|
|
+ Recursively list all subdirectories as a nested tree.
|
|
|
+ Each node: {name, relative_path, ignored, children}.
|
|
|
+ `ignored` is True only if THIS node's own path is in the ignore set —
|
|
|
+ the frontend handles ancestor propagation for inherited-ignore display.
|
|
|
+ """
|
|
|
+ root = Path(root_dir)
|
|
|
+
|
|
|
+ def _scan(path: Path, rel: str, depth: int) -> List[Dict]:
|
|
|
+ if depth > max_depth:
|
|
|
+ return []
|
|
|
+ result = []
|
|
|
+ try:
|
|
|
+ for child in sorted(path.iterdir()):
|
|
|
+ if not child.is_dir():
|
|
|
+ continue
|
|
|
+ child_rel = f"{rel}/{child.name}" if rel else child.name
|
|
|
+ self_ignored = child.name in ignored or child_rel in ignored
|
|
|
+ children = _scan(child, child_rel, depth + 1)
|
|
|
+ result.append({
|
|
|
+ "name": child.name,
|
|
|
+ "relative_path": child_rel,
|
|
|
+ "ignored": self_ignored,
|
|
|
+ "children": children,
|
|
|
+ })
|
|
|
+ except OSError:
|
|
|
+ pass
|
|
|
+ return result
|
|
|
+
|
|
|
+ return _scan(root, "", 0)
|
|
|
+
|
|
|
+
|
|
|
+# ─── Helper: run index build in background ───
|
|
|
+
|
|
|
+def _write_instance_yml(kb_root: str) -> str:
|
|
|
+ """Write a temporary instance.yml that pins all config paths to kb_root.
|
|
|
+
|
|
|
+ config.py loads agent-config.yml first (which may contain hardcoded remote
|
|
|
+ paths like /home/67/knowledge/...) and then overlays instance.yml via deep
|
|
|
+ merge. By setting INSTANCE_CONFIG to this temp file we override those paths
|
|
|
+ with the actual local kb_root before any script reads cfg.base_dir.
|
|
|
+
|
|
|
+ Returns the temp file path; caller is responsible for os.unlink() cleanup.
|
|
|
+ """
|
|
|
+ kb = Path(kb_root)
|
|
|
+ # Single-quote YAML scalars — safe for paths without single quotes
|
|
|
+ lines = [
|
|
|
+ "knowledge:",
|
|
|
+ f" baseDir: '{kb_root}'",
|
|
|
+ f" rawDir: '{kb_root}'",
|
|
|
+ f" processedDir: '{kb / 'processed'}'",
|
|
|
+ f" indexFile: '{kb / 'rag_index.json'}'",
|
|
|
+ f" keywordFile: '{kb / 'rag_index.keywords.json'}'",
|
|
|
+ f" vectorIndex: '{kb / 'rag_vectors_v2.npy'}'",
|
|
|
+ f" vectorMeta: '{kb / 'rag_vectors_meta_v2.pkl'}'",
|
|
|
+ "wiki:",
|
|
|
+ f" dir: '{kb / 'wiki'}'",
|
|
|
+ ]
|
|
|
+ with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
|
|
|
+ f.write("\n".join(lines) + "\n")
|
|
|
+ return f.name
|
|
|
+
|
|
|
+
|
|
|
+def _make_script_env(kb_root: str, scripts_dir: Path, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
|
|
|
+ """Build subprocess env: inherit os.environ + KNOWLEDGE_DIR + PYTHONPATH.
|
|
|
+
|
|
|
+ Does NOT write the instance.yml here (caller manages temp file lifecycle).
|
|
|
+ Pass INSTANCE_CONFIG via `extra` if needed.
|
|
|
+ """
|
|
|
+ env = dict(os.environ)
|
|
|
+ env["KNOWLEDGE_DIR"] = kb_root
|
|
|
+ # Ensure scripts_dir is in PYTHONPATH so `from config import cfg` works
|
|
|
+ existing_py = env.get("PYTHONPATH", "")
|
|
|
+ env["PYTHONPATH"] = str(scripts_dir) + (os.pathsep + existing_py if existing_py else "")
|
|
|
+ if extra:
|
|
|
+ env.update(extra)
|
|
|
+ return env
|
|
|
+
|
|
|
+
|
|
|
+def _popen_stream(
|
|
|
+ script: Path,
|
|
|
+ env: Dict[str, str],
|
|
|
+ cwd: str,
|
|
|
+ log_buf: List[str],
|
|
|
+ db: Database,
|
|
|
+ job_id: str,
|
|
|
+ flush_every: int = 8,
|
|
|
+ timeout: int = 3600,
|
|
|
+) -> int:
|
|
|
+ """Run script with Popen, stream stdout to log_buf, flush to DB every N lines.
|
|
|
+ Returns process returncode. stderr is appended to log_buf on exit.
|
|
|
+ """
|
|
|
+ proc = subprocess.Popen(
|
|
|
+ [sys.executable, str(script)],
|
|
|
+ cwd=cwd, env=env,
|
|
|
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
|
+ text=True, bufsize=1,
|
|
|
+ )
|
|
|
+ flush_count = 0
|
|
|
+ try:
|
|
|
+ for line in proc.stdout: # type: ignore[union-attr]
|
|
|
+ log_buf.append(line.rstrip("\n"))
|
|
|
+ flush_count += 1
|
|
|
+ if flush_count % flush_every == 0:
|
|
|
+ try:
|
|
|
+ db.execute(
|
|
|
+ "UPDATE kb_jobs SET log=? WHERE id=?",
|
|
|
+ ("\n".join(log_buf)[-8000:], job_id),
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ proc.wait(timeout=timeout)
|
|
|
+ except subprocess.TimeoutExpired:
|
|
|
+ proc.kill()
|
|
|
+ log_buf.append("[超时] 进程已终止")
|
|
|
+ stderr_out = (proc.stderr.read() or "").strip() # type: ignore[union-attr]
|
|
|
+ if stderr_out:
|
|
|
+ log_buf.append("--- stderr ---")
|
|
|
+ log_buf.extend(stderr_out[-2000:].splitlines())
|
|
|
+ return proc.returncode
|
|
|
+
|
|
|
+
|
|
|
+def _run_index_job(db: Database, kb_id: str, job_id: str, kb_root: str, index_type: str, mode: str):
|
|
|
+ """Background thread — extract text then build index, streaming log to DB."""
|
|
|
+
|
|
|
+ def _flush(status: str, log_buf: List[str]):
|
|
|
+ try:
|
|
|
+ db.execute(
|
|
|
+ "UPDATE kb_jobs SET status=?, log=? WHERE id=?",
|
|
|
+ (status, "\n".join(log_buf)[-8000:], job_id),
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+ kb_path = Path(kb_root)
|
|
|
+ scripts_dir = _find_scripts_dir(kb_path)
|
|
|
+ instance_yml = _write_instance_yml(kb_root) if scripts_dir else None
|
|
|
+ log_buf: List[str] = []
|
|
|
+ errors: List[str] = []
|
|
|
+
|
|
|
+ _flush("running", log_buf)
|
|
|
+ try:
|
|
|
+ if index_type in ("bm25", "all"):
|
|
|
+ if not scripts_dir:
|
|
|
+ log_buf.append("[BM25] 脚本目录未找到 — 已跳过")
|
|
|
+ else:
|
|
|
+ env = _make_script_env(kb_root, scripts_dir, {"INSTANCE_CONFIG": instance_yml})
|
|
|
+ processed_dir = kb_path / "processed"
|
|
|
+
|
|
|
+ # ── Step 1: 文本提取 ──
|
|
|
+ extract_script = scripts_dir / "extract_docs.py"
|
|
|
+ if extract_script.exists():
|
|
|
+ existing_txts = list(processed_dir.glob("*.txt")) if processed_dir.is_dir() else []
|
|
|
+ if not existing_txts or mode == "full":
|
|
|
+ log_buf.append("[提取] 正在提取原始文档为文本...")
|
|
|
+ _flush("running", log_buf)
|
|
|
+ rc = _popen_stream(extract_script, env, kb_root, log_buf, db, job_id, flush_every=10)
|
|
|
+ if rc != 0:
|
|
|
+ errors.append("文本提取失败(returncode={rc})")
|
|
|
+ else:
|
|
|
+ log_buf.append(f"[提取] 已有 {len(existing_txts)} 个文本文件,跳过(全量模式可强制重新提取)")
|
|
|
+ else:
|
|
|
+ log_buf.append("[提取] extract_docs.py 未找到,跳过(需手动准备 processed/ 目录)")
|
|
|
+
|
|
|
+ # ── Step 2: 构建 BM25 索引 ──
|
|
|
+ bm25_script = next(
|
|
|
+ (scripts_dir / n for n in ("rebuild_index.py", "build_index.py") if (scripts_dir / n).exists()),
|
|
|
+ None,
|
|
|
+ )
|
|
|
+ if bm25_script:
|
|
|
+ log_buf.append(f"[BM25] 正在构建索引({bm25_script.name})...")
|
|
|
+ _flush("running", log_buf)
|
|
|
+ rc = _popen_stream(bm25_script, env, kb_root, log_buf, db, job_id, flush_every=5)
|
|
|
+ if rc != 0:
|
|
|
+ errors.append(f"BM25 构建失败(returncode={rc})")
|
|
|
+ else:
|
|
|
+ log_buf.append("[BM25] rebuild_index.py / build_index.py 均未找到 — 已跳过")
|
|
|
+
|
|
|
+ if index_type in ("vector", "all"):
|
|
|
+ if not scripts_dir:
|
|
|
+ log_buf.append("[Vector] 脚本目录未找到 — 已跳过")
|
|
|
+ else:
|
|
|
+ env = _make_script_env(kb_root, scripts_dir, {"INSTANCE_CONFIG": instance_yml})
|
|
|
+ vec_script = scripts_dir / "build_vector_index.py"
|
|
|
+ if vec_script.exists():
|
|
|
+ log_buf.append("[Vector] 正在构建向量索引...")
|
|
|
+ _flush("running", log_buf)
|
|
|
+ rc = _popen_stream(vec_script, env, kb_root, log_buf, db, job_id, flush_every=5, timeout=3600)
|
|
|
+ if rc != 0:
|
|
|
+ errors.append(f"向量索引失败(returncode={rc})")
|
|
|
+ else:
|
|
|
+ log_buf.append("[Vector] build_vector_index.py 未找到 — 已跳过")
|
|
|
+
|
|
|
+ if index_type in ("graph", "all"):
|
|
|
+ log_buf.append("[Graph] 图谱由 wiki_compile.py 生成 wiki/relations.json — 请先运行 Wiki 编译")
|
|
|
+
|
|
|
+ if errors:
|
|
|
+ log_buf.append("\n❌ ERRORS:\n" + "\n".join(errors))
|
|
|
+ _flush("failed" if errors else "completed", log_buf)
|
|
|
+ db.execute("UPDATE kb_jobs SET completed_at=? WHERE id=?", (now_utc(), job_id))
|
|
|
+ db.commit()
|
|
|
+
|
|
|
+ except Exception as exc:
|
|
|
+ log_buf.append(f"\n❌ 异常: {exc}")
|
|
|
+ _flush("failed", log_buf)
|
|
|
+ db.execute(
|
|
|
+ "UPDATE kb_jobs SET error=?, completed_at=? WHERE id=?",
|
|
|
+ (str(exc)[:500], now_utc(), job_id),
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ finally:
|
|
|
+ if instance_yml:
|
|
|
+ try:
|
|
|
+ os.unlink(instance_yml)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+def _detect_local_llm_url() -> str:
|
|
|
+ """Probe common local LLM endpoints and return the first that responds.
|
|
|
+
|
|
|
+ Priority: VLLM_URL env var → Ollama :11434 → Ollama :11435 → vLLM :8001
|
|
|
+ Falls back to the env default so we never block on a missing service here.
|
|
|
+ """
|
|
|
+ import urllib.request as _ur
|
|
|
+ # If explicitly set by operator/user, trust it
|
|
|
+ if os.environ.get("VLLM_URL"):
|
|
|
+ return os.environ["VLLM_URL"]
|
|
|
+
|
|
|
+ candidates = [
|
|
|
+ "http://127.0.0.1:11434/v1/chat/completions", # Ollama default
|
|
|
+ "http://127.0.0.1:11435/v1/chat/completions", # Ollama alt port
|
|
|
+ "http://127.0.0.1:8001/v1/chat/completions", # vLLM on non-conflicting port
|
|
|
+ ]
|
|
|
+ for url in candidates:
|
|
|
+ try:
|
|
|
+ req = _ur.Request(url, data=b'{}', headers={"Content-Type": "application/json"}, method="POST")
|
|
|
+ _ur.urlopen(req, timeout=2)
|
|
|
+ except Exception as exc:
|
|
|
+ # Connection refused / timeout → not available; 4xx means server IS there
|
|
|
+ msg = str(exc)
|
|
|
+ if "400" in msg or "422" in msg or "405" in msg or "404" in msg:
|
|
|
+ return url
|
|
|
+ else:
|
|
|
+ return url # 2xx on empty payload — unlikely but accept
|
|
|
+ # Nothing found; return Ollama default and let wiki_compile.py fail with a clear message
|
|
|
+ return "http://127.0.0.1:11434/v1/chat/completions"
|
|
|
+
|
|
|
+
|
|
|
+def _run_wiki_job(db: Database, kb_id: str, job_id: str, kb_root: str, mode: str, concurrency: int):
|
|
|
+ """Background thread: run wiki_compile.py with live log streaming to DB."""
|
|
|
+
|
|
|
+ def _flush(status: str, log_buf: List[str]):
|
|
|
+ try:
|
|
|
+ db.execute(
|
|
|
+ "UPDATE kb_jobs SET status=?, log=? WHERE id=?",
|
|
|
+ (status, "\n".join(log_buf)[-8000:], job_id),
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+ kb_path = Path(kb_root)
|
|
|
+ scripts_dir = _find_scripts_dir(kb_path)
|
|
|
+ script = scripts_dir / "wiki_compile.py" if scripts_dir else None
|
|
|
+ instance_yml = _write_instance_yml(kb_root) if scripts_dir else None
|
|
|
+ log_buf: List[str] = []
|
|
|
+
|
|
|
+ _flush("running", log_buf)
|
|
|
+ try:
|
|
|
+ if not script or not script.exists():
|
|
|
+ db.execute(
|
|
|
+ "UPDATE kb_jobs SET status='failed', error=?, completed_at=? WHERE id=?",
|
|
|
+ ("wiki_compile.py 未在脚本目录中找到", now_utc(), job_id),
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ return
|
|
|
+
|
|
|
+ llm_url = _detect_local_llm_url()
|
|
|
+ log_buf.append(f"[Wiki] LLM 端点: {llm_url}")
|
|
|
+ _flush("running", log_buf)
|
|
|
+
|
|
|
+ env = _make_script_env(kb_root, scripts_dir, {
|
|
|
+ "INSTANCE_CONFIG": instance_yml,
|
|
|
+ "WIKI_DIR": str(kb_path / "wiki"),
|
|
|
+ "VLLM_URL": llm_url,
|
|
|
+ })
|
|
|
+ if mode == "rebuild":
|
|
|
+ env["WIKI_REBUILD"] = "1"
|
|
|
+ env["WIKI_CONCURRENCY"] = str(concurrency)
|
|
|
+
|
|
|
+ rc = _popen_stream(script, env, kb_root, log_buf, db, job_id, flush_every=3, timeout=7200)
|
|
|
+
|
|
|
+ _flush("failed" if rc != 0 else "completed", log_buf)
|
|
|
+ db.execute("UPDATE kb_jobs SET completed_at=? WHERE id=?", (now_utc(), job_id))
|
|
|
+ db.commit()
|
|
|
+
|
|
|
+ except Exception as exc:
|
|
|
+ log_buf.append(f"\n❌ 异常: {exc}")
|
|
|
+ _flush("failed", log_buf)
|
|
|
+ db.execute(
|
|
|
+ "UPDATE kb_jobs SET error=?, completed_at=? WHERE id=?",
|
|
|
+ (str(exc)[:500], now_utc(), job_id),
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ finally:
|
|
|
+ if instance_yml:
|
|
|
+ try:
|
|
|
+ os.unlink(instance_yml)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+def _find_scripts_dir(kb_path: Path) -> Optional[Path]:
|
|
|
+ """Locate the scripts dir containing rebuild_index.py / wiki_compile.py etc.
|
|
|
+ Checks kb_path itself (if scripts live alongside data) and the canonical
|
|
|
+ project path agentexample/qaagent67lambda/scripts/.
|
|
|
+ Uses rebuild_index.py or wiki_compile.py as the presence marker
|
|
|
+ (build_index.py is the old hardcoded variant and no longer required).
|
|
|
+ """
|
|
|
+ _markers = ("rebuild_index.py", "wiki_compile.py", "build_index.py")
|
|
|
+
|
|
|
+ # Check if scripts live inside kb_path or its parent
|
|
|
+ for cand in [kb_path, kb_path.parent]:
|
|
|
+ if any((cand / m).exists() for m in _markers):
|
|
|
+ return cand
|
|
|
+
|
|
|
+ # Walk up from this source file looking for the canonical scripts dir
|
|
|
+ here = Path(__file__).resolve()
|
|
|
+ for _ in range(8):
|
|
|
+ here = here.parent
|
|
|
+ cand = here / "agentexample" / "qaagent67lambda" / "scripts"
|
|
|
+ if cand.is_dir() and any((cand / m).exists() for m in _markers):
|
|
|
+ return cand
|
|
|
+
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def _get_index_status(kb_root: str) -> Dict[str, Any]:
|
|
|
+ """Read index file presence/stats from kb_root."""
|
|
|
+ root = Path(kb_root)
|
|
|
+ bm25_file = root / "rag_index.json"
|
|
|
+ vector_file = root / "rag_vectors_v2.npy"
|
|
|
+ wiki_dir = root / "wiki"
|
|
|
+
|
|
|
+ def _fstat(p: Path) -> Optional[Dict]:
|
|
|
+ if p.exists():
|
|
|
+ s = p.stat()
|
|
|
+ return {"exists": True, "size": s.st_size, "mtime": s.st_mtime}
|
|
|
+ return {"exists": False}
|
|
|
+
|
|
|
+ bm25 = _fstat(bm25_file)
|
|
|
+ if bm25["exists"]:
|
|
|
+ try:
|
|
|
+ import json as _json
|
|
|
+ with open(bm25_file) as f:
|
|
|
+ idx = _json.load(f)
|
|
|
+ bm25["total_chunks"] = idx.get("total_chunks", 0)
|
|
|
+ bm25["total_files"] = idx.get("total_files", 0)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+ vector = _fstat(vector_file)
|
|
|
+ wiki = {"exists": wiki_dir.is_dir()}
|
|
|
+ if wiki["exists"]:
|
|
|
+ subdirs = {}
|
|
|
+ for sub in ["sources", "entities", "topics", "analyses"]:
|
|
|
+ d = wiki_dir / sub
|
|
|
+ subdirs[sub] = len(list(d.glob("*.md"))) if d.is_dir() else 0
|
|
|
+ wiki["subdirs"] = subdirs
|
|
|
+ wiki["total_pages"] = sum(subdirs.values())
|
|
|
+
|
|
|
+ return {"bm25": bm25, "vector": vector, "wiki": wiki}
|
|
|
+
|
|
|
+
|
|
|
+# ─── Chunk index cache ───
|
|
|
+# Avoid re-reading a 17 MB JSON on every request; invalidate when file mtime changes.
|
|
|
+_chunk_cache: Dict[str, Any] = {} # kb_id → {"mtime": float, "chunks": list, "total": int}
|
|
|
+
|
|
|
+
|
|
|
+def _load_chunks(kb_root: str, kb_id: str) -> Optional[List[Dict]]:
|
|
|
+ """Load chunks from rag_index.json with mtime-based in-process cache."""
|
|
|
+ index_file = Path(kb_root) / "rag_index.json"
|
|
|
+ if not index_file.exists():
|
|
|
+ return None
|
|
|
+ mtime = index_file.stat().st_mtime
|
|
|
+ cached = _chunk_cache.get(kb_id)
|
|
|
+ if cached and cached["mtime"] == mtime:
|
|
|
+ return cached["chunks"]
|
|
|
+ try:
|
|
|
+ with open(index_file, encoding="utf-8") as f:
|
|
|
+ data = json.load(f)
|
|
|
+ chunks = data.get("chunks", [])
|
|
|
+ _chunk_cache[kb_id] = {"mtime": mtime, "chunks": chunks, "total": len(chunks)}
|
|
|
+ return chunks
|
|
|
+ except Exception:
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+# ─── CRUD Endpoints ───
|
|
|
+
|
|
|
+@router.get("")
|
|
|
+async def list_kbs(
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ rows = db.fetchall(
|
|
|
+ "SELECT * FROM knowledge_bases WHERE tenant_id=? ORDER BY created_at DESC",
|
|
|
+ (tenant.tenant_id,)
|
|
|
+ )
|
|
|
+ result = []
|
|
|
+ for r in rows:
|
|
|
+ status = _get_index_status(r["root_dir"])
|
|
|
+ result.append({**r, "index_status": status})
|
|
|
+ return {"items": result}
|
|
|
+
|
|
|
+
|
|
|
+@router.post("", status_code=201)
|
|
|
+async def create_kb(
|
|
|
+ req: CreateKBRequest,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ root = Path(req.root_dir)
|
|
|
+ if not root.is_dir():
|
|
|
+ raise HTTPException(status_code=400, detail=f"Directory not found: {req.root_dir}")
|
|
|
+
|
|
|
+ kb_id = gen_id("kb_")
|
|
|
+ db.execute(
|
|
|
+ "INSERT INTO knowledge_bases (id, tenant_id, name, root_dir, description, created_at, updated_at) "
|
|
|
+ "VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
|
+ (kb_id, tenant.tenant_id, req.name, str(root.resolve()), req.description, now_utc(), now_utc())
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ return db.fetchone("SELECT * FROM knowledge_bases WHERE id=?", (kb_id,))
|
|
|
+
|
|
|
+
|
|
|
+@router.get("/{kb_id}")
|
|
|
+async def get_kb(
|
|
|
+ kb_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+ status = _get_index_status(kb["root_dir"])
|
|
|
+ return {**kb, "index_status": status}
|
|
|
+
|
|
|
+
|
|
|
+@router.delete("/{kb_id}", status_code=204)
|
|
|
+async def delete_kb(
|
|
|
+ kb_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+ db.execute("DELETE FROM kb_files WHERE kb_id=?", (kb_id,))
|
|
|
+ db.execute("DELETE FROM kb_jobs WHERE kb_id=?", (kb_id,))
|
|
|
+ db.execute("DELETE FROM knowledge_bases WHERE id=?", (kb_id,))
|
|
|
+ db.commit()
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+# ─── Chunk Browser ───
|
|
|
+
|
|
|
+@router.get("/{kb_id}/chunks")
|
|
|
+async def list_chunks(
|
|
|
+ kb_id: str,
|
|
|
+ q: str = Query(default="", description="Keyword filter (case-insensitive substring)"),
|
|
|
+ source: str = Query(default="", description="Filter by source file name"),
|
|
|
+ page: int = Query(default=1, ge=1),
|
|
|
+ limit: int = Query(default=20, ge=1, le=100),
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ chunks = _load_chunks(kb["root_dir"], kb_id)
|
|
|
+ if chunks is None:
|
|
|
+ raise HTTPException(status_code=404, detail="BM25 index not built yet (rag_index.json not found)")
|
|
|
+
|
|
|
+ # Filter
|
|
|
+ q_lower = q.strip().lower()
|
|
|
+ src_lower = source.strip().lower()
|
|
|
+ filtered = chunks
|
|
|
+ if q_lower:
|
|
|
+ filtered = [c for c in filtered if q_lower in c.get("text", "").lower()]
|
|
|
+ if src_lower:
|
|
|
+ filtered = [c for c in filtered if src_lower in c.get("source", "").lower()]
|
|
|
+
|
|
|
+ total = len(filtered)
|
|
|
+ offset = (page - 1) * limit
|
|
|
+ page_chunks = filtered[offset: offset + limit]
|
|
|
+
|
|
|
+ # Return trimmed chunks (avoid huge payloads)
|
|
|
+ return {
|
|
|
+ "total": total,
|
|
|
+ "total_all": len(chunks),
|
|
|
+ "page": page,
|
|
|
+ "limit": limit,
|
|
|
+ "pages": max(1, (total + limit - 1) // limit),
|
|
|
+ "chunks": [
|
|
|
+ {
|
|
|
+ "id": c.get("id", ""),
|
|
|
+ "source": c.get("source", ""),
|
|
|
+ "chunk_id": c.get("chunk_id", 0),
|
|
|
+ "text": c.get("text", ""),
|
|
|
+ "length": c.get("length", 0),
|
|
|
+ }
|
|
|
+ for c in page_chunks
|
|
|
+ ],
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ─── File Management ───
|
|
|
+
|
|
|
+@router.get("/{kb_id}/files")
|
|
|
+async def list_files(
|
|
|
+ kb_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ # Load saved ignore list
|
|
|
+ raw_ignored = kb.get("ignore_dirs") or "[]"
|
|
|
+ try:
|
|
|
+ ignored: Set[str] = set(json.loads(raw_ignored))
|
|
|
+ except Exception:
|
|
|
+ ignored = set()
|
|
|
+
|
|
|
+ added = db.fetchall("SELECT file_path FROM kb_files WHERE kb_id=?", (kb_id,))
|
|
|
+ added_paths = {r["file_path"] for r in added}
|
|
|
+ files = _walk_dir(kb["root_dir"], added_paths, ignored)
|
|
|
+ dirs = _list_dir_tree(kb["root_dir"], ignored)
|
|
|
+
|
|
|
+ return {
|
|
|
+ "kb_id": kb_id,
|
|
|
+ "root_dir": kb["root_dir"],
|
|
|
+ "files": files,
|
|
|
+ "dirs": dirs,
|
|
|
+ "ignored_dirs": sorted(ignored),
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+@router.get("/{kb_id}/ignore-dirs")
|
|
|
+async def get_ignore_dirs(
|
|
|
+ kb_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+ raw = kb.get("ignore_dirs") or "[]"
|
|
|
+ try:
|
|
|
+ ignored = sorted(set(json.loads(raw)))
|
|
|
+ except Exception:
|
|
|
+ ignored = []
|
|
|
+ return {"ignored_dirs": ignored}
|
|
|
+
|
|
|
+
|
|
|
+@router.put("/{kb_id}/ignore-dirs")
|
|
|
+async def update_ignore_dirs(
|
|
|
+ kb_id: str,
|
|
|
+ req: UpdateIgnoreDirsRequest,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ # Normalise: strip slashes, dedup, sort
|
|
|
+ clean = sorted({p.strip("/").strip() for p in req.ignored_dirs if p.strip()})
|
|
|
+ db.execute(
|
|
|
+ "UPDATE knowledge_bases SET ignore_dirs=?, updated_at=? WHERE id=?",
|
|
|
+ (json.dumps(clean), now_utc(), kb_id)
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+ return {"ignored_dirs": clean}
|
|
|
+
|
|
|
+
|
|
|
+@router.post("/{kb_id}/files", status_code=201)
|
|
|
+async def add_files(
|
|
|
+ kb_id: str,
|
|
|
+ req: AddFilesRequest,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ added = []
|
|
|
+ for fp in req.file_paths:
|
|
|
+ p = Path(fp)
|
|
|
+ if not p.is_file():
|
|
|
+ continue
|
|
|
+ existing = db.fetchone("SELECT id FROM kb_files WHERE kb_id=? AND file_path=?", (kb_id, str(p)))
|
|
|
+ if existing:
|
|
|
+ continue
|
|
|
+ fid = gen_id("kbf_")
|
|
|
+ db.execute(
|
|
|
+ "INSERT INTO kb_files (id, kb_id, file_path, file_name, file_type, size, status, added_at) "
|
|
|
+ "VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)",
|
|
|
+ (fid, kb_id, str(p), p.name, p.suffix.lower().lstrip("."), p.stat().st_size, now_utc())
|
|
|
+ )
|
|
|
+ added.append(fid)
|
|
|
+ db.commit()
|
|
|
+ return {"added": len(added)}
|
|
|
+
|
|
|
+
|
|
|
+@router.delete("/{kb_id}/files/{file_id}", status_code=204)
|
|
|
+async def remove_file(
|
|
|
+ kb_id: str,
|
|
|
+ file_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+ db.execute("DELETE FROM kb_files WHERE id=? AND kb_id=?", (file_id, kb_id))
|
|
|
+ db.commit()
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+# ─── Index Build Jobs ───
|
|
|
+
|
|
|
+@router.post("/{kb_id}/index", status_code=202)
|
|
|
+async def trigger_index(
|
|
|
+ kb_id: str,
|
|
|
+ req: IndexBuildRequest,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ job_id = gen_id("kbj_")
|
|
|
+ db.execute(
|
|
|
+ "INSERT INTO kb_jobs (id, kb_id, tenant_id, job_type, status, created_at) VALUES (?,?,?,?,?,?)",
|
|
|
+ (job_id, kb_id, tenant.tenant_id, f"index:{req.index_type}", "pending", now_utc())
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+
|
|
|
+ t = threading.Thread(
|
|
|
+ target=_run_index_job,
|
|
|
+ args=(db, kb_id, job_id, kb["root_dir"], req.index_type, req.mode),
|
|
|
+ daemon=True,
|
|
|
+ )
|
|
|
+ t.start()
|
|
|
+ return {"job_id": job_id, "status": "pending"}
|
|
|
+
|
|
|
+
|
|
|
+# ─── Wiki Compile Jobs ───
|
|
|
+
|
|
|
+@router.post("/{kb_id}/wiki", status_code=202)
|
|
|
+async def trigger_wiki(
|
|
|
+ kb_id: str,
|
|
|
+ req: WikiBuildRequest,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ job_id = gen_id("kbj_")
|
|
|
+ db.execute(
|
|
|
+ "INSERT INTO kb_jobs (id, kb_id, tenant_id, job_type, status, created_at) VALUES (?,?,?,?,?,?)",
|
|
|
+ (job_id, kb_id, tenant.tenant_id, "wiki", "pending", now_utc())
|
|
|
+ )
|
|
|
+ db.commit()
|
|
|
+
|
|
|
+ t = threading.Thread(
|
|
|
+ target=_run_wiki_job,
|
|
|
+ args=(db, kb_id, job_id, kb["root_dir"], req.mode, req.concurrency),
|
|
|
+ daemon=True,
|
|
|
+ )
|
|
|
+ t.start()
|
|
|
+ return {"job_id": job_id, "status": "pending"}
|
|
|
+
|
|
|
+
|
|
|
+@router.put("/{kb_id}/wiki/pause", status_code=200)
|
|
|
+async def pause_wiki(
|
|
|
+ kb_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ # Write a stop flag that wiki_compile.py checks
|
|
|
+ stop_file = Path(kb["root_dir"]) / ".wiki_stop"
|
|
|
+ if stop_file.exists():
|
|
|
+ stop_file.unlink()
|
|
|
+ return {"action": "resumed"}
|
|
|
+ else:
|
|
|
+ stop_file.touch()
|
|
|
+ return {"action": "paused"}
|
|
|
+
|
|
|
+
|
|
|
+# ─── Wiki Content ───
|
|
|
+
|
|
|
+@router.get("/{kb_id}/wiki/tree")
|
|
|
+async def wiki_tree(
|
|
|
+ kb_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ wiki_path = Path(kb["root_dir"]) / "wiki"
|
|
|
+ result: Dict[str, Any] = {}
|
|
|
+ for sub in ["sources", "entities", "topics", "analyses"]:
|
|
|
+ d = wiki_path / sub
|
|
|
+ pages = []
|
|
|
+ if d.is_dir():
|
|
|
+ for f in sorted(d.glob("*.md")):
|
|
|
+ s = f.stat()
|
|
|
+ pages.append({
|
|
|
+ "name": f.stem,
|
|
|
+ "path": f"{sub}/{f.name}",
|
|
|
+ "size": s.st_size,
|
|
|
+ "modified": s.st_mtime,
|
|
|
+ })
|
|
|
+ result[sub] = pages
|
|
|
+
|
|
|
+ idx = wiki_path / "index.md"
|
|
|
+ result["index_content"] = idx.read_text(encoding="utf-8") if idx.exists() else ""
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+@router.get("/{kb_id}/wiki/{page_path:path}")
|
|
|
+async def wiki_page(
|
|
|
+ kb_id: str,
|
|
|
+ page_path: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ wiki_path = Path(kb["root_dir"]) / "wiki"
|
|
|
+ full = (wiki_path / page_path).resolve()
|
|
|
+ # Path traversal check
|
|
|
+ if not str(full).startswith(str(wiki_path.resolve())):
|
|
|
+ raise HTTPException(status_code=403, detail="Forbidden")
|
|
|
+ if not full.exists():
|
|
|
+ raise HTTPException(status_code=404, detail="Page not found")
|
|
|
+ return {"path": page_path, "content": full.read_text(encoding="utf-8"), "size": full.stat().st_size}
|
|
|
+
|
|
|
+
|
|
|
+@router.delete("/{kb_id}/wiki/analyses/{name}", status_code=204)
|
|
|
+async def delete_analysis(
|
|
|
+ kb_id: str,
|
|
|
+ name: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ p = Path(kb["root_dir"]) / "wiki" / "analyses" / f"{name}.md"
|
|
|
+ if p.exists() and str(p.resolve()).startswith(str((Path(kb["root_dir"]) / "wiki").resolve())):
|
|
|
+ p.unlink()
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+# ─── Search helpers ───
|
|
|
+
|
|
|
+def _search_subprocess(kb_root: str, scripts_dir: Path, mode: str, query: str, top_k: int) -> List[Dict]:
|
|
|
+ """Run search in an isolated subprocess to avoid config singleton / module cache issues.
|
|
|
+
|
|
|
+ search_engine.py uses `from config import cfg` at module level. Once imported
|
|
|
+ in-process the singleton is frozen to the first kb_root it saw. Running as a
|
|
|
+ subprocess gives each search a fresh Python interpreter with the correct
|
|
|
+ INSTANCE_CONFIG / KNOWLEDGE_DIR env vars.
|
|
|
+ """
|
|
|
+ instance_yml = _write_instance_yml(kb_root)
|
|
|
+ try:
|
|
|
+ env = _make_script_env(kb_root, scripts_dir, {"INSTANCE_CONFIG": instance_yml})
|
|
|
+
|
|
|
+ # Build the inline runner script — use json.dumps for safe string embedding
|
|
|
+ import_stmt = (
|
|
|
+ "from search_engine import search"
|
|
|
+ if mode == "bm25"
|
|
|
+ else (
|
|
|
+ "try:\n from search_unified import search\n"
|
|
|
+ "except ImportError:\n from search_engine import search"
|
|
|
+ )
|
|
|
+ )
|
|
|
+ runner = (
|
|
|
+ f"import sys, os, json\n"
|
|
|
+ f"sys.path.insert(0, {json.dumps(str(scripts_dir))})\n"
|
|
|
+ f"os.environ['INSTANCE_CONFIG'] = {json.dumps(instance_yml)}\n"
|
|
|
+ f"os.environ['KNOWLEDGE_DIR'] = {json.dumps(kb_root)}\n"
|
|
|
+ f"{import_stmt}\n"
|
|
|
+ f"results = search({json.dumps(query)}, top_k={top_k})\n"
|
|
|
+ f"print(json.dumps(results, ensure_ascii=False))\n"
|
|
|
+ )
|
|
|
+ r = subprocess.run(
|
|
|
+ [sys.executable, "-c", runner],
|
|
|
+ capture_output=True, text=True, timeout=30, env=env,
|
|
|
+ )
|
|
|
+ if r.returncode != 0:
|
|
|
+ raise RuntimeError(r.stderr[-800:] or "搜索子进程异常退出")
|
|
|
+ # Last line of stdout is the JSON result (earlier lines may be progress logs)
|
|
|
+ lines = [l for l in r.stdout.strip().splitlines() if l.strip()]
|
|
|
+ if not lines:
|
|
|
+ return []
|
|
|
+ return json.loads(lines[-1])
|
|
|
+ finally:
|
|
|
+ try:
|
|
|
+ os.unlink(instance_yml)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+# ─── Search ───
|
|
|
+
|
|
|
+@router.post("/{kb_id}/search")
|
|
|
+async def search_kb(
|
|
|
+ kb_id: str,
|
|
|
+ req: SearchRequest,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ kb_root = kb["root_dir"]
|
|
|
+ scripts_dir = _find_scripts_dir(Path(kb_root))
|
|
|
+
|
|
|
+ if not scripts_dir:
|
|
|
+ raise HTTPException(status_code=503, detail="搜索脚本未找到,请先构建 BM25 索引。")
|
|
|
+
|
|
|
+ # Check index exists
|
|
|
+ index_file = Path(kb_root) / "rag_index.json"
|
|
|
+ if not index_file.exists():
|
|
|
+ raise HTTPException(status_code=503, detail="BM25 索引不存在,请先在「索引构建」标签中触发构建。")
|
|
|
+
|
|
|
+ try:
|
|
|
+ results = await asyncio.get_event_loop().run_in_executor(
|
|
|
+ None,
|
|
|
+ lambda: _search_subprocess(kb_root, scripts_dir, req.mode, req.query, req.top_k)
|
|
|
+ )
|
|
|
+ return {
|
|
|
+ "query": req.query,
|
|
|
+ "mode": req.mode,
|
|
|
+ "results": results[:req.top_k],
|
|
|
+ }
|
|
|
+ except RuntimeError as e:
|
|
|
+ raise HTTPException(status_code=500, detail=str(e))
|
|
|
+ except Exception as e:
|
|
|
+ raise HTTPException(status_code=500, detail=str(e))
|
|
|
+
|
|
|
+
|
|
|
+def _detect_ollama_model() -> tuple[str, str]:
|
|
|
+ """Return (base_url, model_name) for the best available local LLM.
|
|
|
+ Probes Ollama at 11434/11435, picks qwen2.5:7b if available, else first model found.
|
|
|
+ """
|
|
|
+ import urllib.request as _ur
|
|
|
+ preferred = ["qwen2.5:7b", "qwen2.5:14b", "qwen2.5:32b", "qwen2.5-coder:7b"]
|
|
|
+ for port in (11434, 11435):
|
|
|
+ base = f"http://127.0.0.1:{port}"
|
|
|
+ try:
|
|
|
+ with _ur.urlopen(f"{base}/api/tags", timeout=2) as r:
|
|
|
+ data = json.loads(r.read())
|
|
|
+ models = [m["name"] for m in data.get("models", [])]
|
|
|
+ if not models:
|
|
|
+ continue
|
|
|
+ chosen = next((m for m in preferred if m in models), models[0])
|
|
|
+ return base, chosen
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+ return "http://127.0.0.1:11434", "qwen2.5:7b"
|
|
|
+
|
|
|
+
|
|
|
+def _resolve_llm(provider_id: str, model: str) -> tuple[str, dict, str, str]:
|
|
|
+ """Resolve LLM call parameters from a configured provider.
|
|
|
+
|
|
|
+ Returns (chat_url, headers, model_name, display_label).
|
|
|
+ Falls back to local Ollama if provider_id is empty or "ollama".
|
|
|
+ """
|
|
|
+ from agentpaas.api.v1.providers_api import DEFAULT_PROVIDERS, _load_providers
|
|
|
+
|
|
|
+ # ── Ollama fallback ──────────────────────────────────────────
|
|
|
+ if not provider_id or provider_id == "ollama":
|
|
|
+ base, auto_model = _detect_ollama_model()
|
|
|
+ chosen = model or auto_model
|
|
|
+ chat_url = f"{base}/v1/chat/completions"
|
|
|
+ return chat_url, {"Content-Type": "application/json"}, chosen, f"Ollama / {chosen}"
|
|
|
+
|
|
|
+ pdef = DEFAULT_PROVIDERS.get(provider_id)
|
|
|
+ if not pdef:
|
|
|
+ raise ValueError(f"未知提供商: {provider_id}")
|
|
|
+
|
|
|
+ saved = _load_providers()
|
|
|
+ entry = saved.get(provider_id, {})
|
|
|
+ base_url = (entry.get("base_url") or pdef.get("base_url", "")).rstrip("/")
|
|
|
+ api_key = entry.get("api_key") or os.getenv(pdef.get("env_key", ""), "")
|
|
|
+ chosen = model or (pdef["models"][0] if pdef.get("models") else "")
|
|
|
+
|
|
|
+ ptype = pdef.get("type", "openai_compatible")
|
|
|
+
|
|
|
+ # ── Ollama type ──────────────────────────────────────────────
|
|
|
+ if ptype == "ollama":
|
|
|
+ # Detect actual running port from saved/default base_url
|
|
|
+ # but try to also auto-probe in case port changed
|
|
|
+ try:
|
|
|
+ auto_base, auto_model = _detect_ollama_model()
|
|
|
+ except Exception:
|
|
|
+ auto_base = base_url
|
|
|
+ auto_model = chosen
|
|
|
+ final_base = entry.get("base_url") or auto_base
|
|
|
+ chosen = model or auto_model
|
|
|
+ chat_url = f"{final_base.rstrip('/')}/v1/chat/completions"
|
|
|
+ return chat_url, {"Content-Type": "application/json"}, chosen, f"Ollama / {chosen}"
|
|
|
+
|
|
|
+ # ── claude-code type (not streamable via HTTP) ────────────────
|
|
|
+ if ptype == "claude-code":
|
|
|
+ raise ValueError("claude-code 提供商不支持 HTTP 流式调用,请选择其他提供商")
|
|
|
+
|
|
|
+ # ── Anthropic native API ──────────────────────────────────────
|
|
|
+ if ptype == "anthropic":
|
|
|
+ if not api_key:
|
|
|
+ raise ValueError("Anthropic API Key 未配置")
|
|
|
+ chat_url = f"{base_url}/v1/messages"
|
|
|
+ headers = {
|
|
|
+ "Content-Type": "application/json",
|
|
|
+ "x-api-key": api_key,
|
|
|
+ "anthropic-version": "2023-06-01",
|
|
|
+ }
|
|
|
+ return chat_url, headers, chosen, f"Anthropic / {chosen}"
|
|
|
+
|
|
|
+ # ── OpenAI / OpenAI-compatible ────────────────────────────────
|
|
|
+ if not api_key:
|
|
|
+ raise ValueError(f"提供商 {pdef['name']} 未配置 API Key")
|
|
|
+ chat_url = f"{base_url}/chat/completions"
|
|
|
+ headers = {
|
|
|
+ "Content-Type": "application/json",
|
|
|
+ "Authorization": f"Bearer {api_key}",
|
|
|
+ }
|
|
|
+ return chat_url, headers, chosen, f"{pdef['name']} / {chosen}"
|
|
|
+
|
|
|
+
|
|
|
+def _stream_openai_compat(
|
|
|
+ chat_url: str, headers: dict, model: str, messages: list, temperature: float = 0.3
|
|
|
+):
|
|
|
+ """Generator that yields text delta strings from an OpenAI-compatible streaming endpoint."""
|
|
|
+ import urllib.request as _ur
|
|
|
+ body = json.dumps({
|
|
|
+ "model": model,
|
|
|
+ "messages": messages,
|
|
|
+ "stream": True,
|
|
|
+ "temperature": temperature,
|
|
|
+ }, ensure_ascii=False).encode("utf-8")
|
|
|
+ req = _ur.Request(chat_url, data=body, headers=headers, method="POST")
|
|
|
+ with _ur.urlopen(req, timeout=180) as resp:
|
|
|
+ for raw_line in resp:
|
|
|
+ line = raw_line.decode("utf-8").strip()
|
|
|
+ if not line.startswith("data:"):
|
|
|
+ continue
|
|
|
+ payload = line[5:].strip()
|
|
|
+ if payload == "[DONE]":
|
|
|
+ break
|
|
|
+ try:
|
|
|
+ d = json.loads(payload)
|
|
|
+ delta = d["choices"][0]["delta"].get("content", "")
|
|
|
+ if delta:
|
|
|
+ yield delta
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+
|
|
|
+
|
|
|
+def _stream_anthropic(
|
|
|
+ chat_url: str, headers: dict, model: str, messages: list, temperature: float = 0.3
|
|
|
+):
|
|
|
+ """Generator that yields text delta strings from Anthropic Messages streaming API."""
|
|
|
+ import urllib.request as _ur
|
|
|
+ # Convert OpenAI-style messages to Anthropic format
|
|
|
+ system_parts = [m["content"] for m in messages if m["role"] == "system"]
|
|
|
+ user_msgs = [m for m in messages if m["role"] != "system"]
|
|
|
+ body_dict: dict = {
|
|
|
+ "model": model,
|
|
|
+ "messages": user_msgs,
|
|
|
+ "max_tokens": 2048,
|
|
|
+ "stream": True,
|
|
|
+ "temperature": temperature,
|
|
|
+ }
|
|
|
+ if system_parts:
|
|
|
+ body_dict["system"] = "\n\n".join(system_parts)
|
|
|
+ body = json.dumps(body_dict, ensure_ascii=False).encode("utf-8")
|
|
|
+ req = _ur.Request(chat_url, data=body, headers=headers, method="POST")
|
|
|
+ with _ur.urlopen(req, timeout=180) as resp:
|
|
|
+ for raw_line in resp:
|
|
|
+ line = raw_line.decode("utf-8").strip()
|
|
|
+ if not line.startswith("data:"):
|
|
|
+ continue
|
|
|
+ payload = line[5:].strip()
|
|
|
+ try:
|
|
|
+ d = json.loads(payload)
|
|
|
+ if d.get("type") == "content_block_delta":
|
|
|
+ delta = d.get("delta", {}).get("text", "")
|
|
|
+ if delta:
|
|
|
+ yield delta
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+
|
|
|
+
|
|
|
+@router.get("/{kb_id}/search/stream")
|
|
|
+async def search_stream(
|
|
|
+ kb_id: str,
|
|
|
+ q: str = Query(..., description="Question to answer"),
|
|
|
+ mode: str = Query(default="bm25"),
|
|
|
+ top_k: int = Query(default=5, ge=1, le=20),
|
|
|
+ provider_id: str = Query(default="", description="Provider ID from providers config"),
|
|
|
+ model: str = Query(default="", description="Model name override"),
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ """SSE streaming QA: retrieve → build prompt → stream LLM answer token by token."""
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ kb_root = kb["root_dir"]
|
|
|
+ scripts_dir = _find_scripts_dir(Path(kb_root))
|
|
|
+
|
|
|
+ async def event_stream():
|
|
|
+ # ── Step 1: retrieve ──
|
|
|
+ yield f"data: {json.dumps({'type': 'status', 'text': '正在检索相关文档...'})}\n\n"
|
|
|
+ await asyncio.sleep(0)
|
|
|
+
|
|
|
+ chunks: List[Dict] = []
|
|
|
+ if scripts_dir and (Path(kb_root) / "rag_index.json").exists():
|
|
|
+ try:
|
|
|
+ chunks = await asyncio.get_event_loop().run_in_executor(
|
|
|
+ None,
|
|
|
+ lambda: _search_subprocess(kb_root, scripts_dir, mode, q, top_k)
|
|
|
+ )
|
|
|
+ except Exception as e:
|
|
|
+ yield f"data: {json.dumps({'type': 'status', 'text': f'检索出错: {e}'})}\n\n"
|
|
|
+ else:
|
|
|
+ yield f"data: {json.dumps({'type': 'status', 'text': 'BM25 索引不存在,请先构建索引'})}\n\n"
|
|
|
+
|
|
|
+ # Emit retrieved chunks so the UI can show sources panel
|
|
|
+ yield f"data: {json.dumps({'type': 'chunks', 'chunks': chunks[:top_k]})}\n\n"
|
|
|
+ await asyncio.sleep(0)
|
|
|
+
|
|
|
+ if not chunks:
|
|
|
+ yield f"data: {json.dumps({'type': 'chunk', 'text': '未检索到相关文档,无法生成回答。'})}\n\n"
|
|
|
+ yield f"data: {json.dumps({'type': 'done'})}\n\n"
|
|
|
+ return
|
|
|
+
|
|
|
+ # ── Step 2: build prompt ──
|
|
|
+ context_parts = []
|
|
|
+ for i, c in enumerate(chunks[:top_k]):
|
|
|
+ context_parts.append(f"[来源{i+1}: {c.get('source', '?')}]\n{c.get('text', '')}")
|
|
|
+ context = "\n\n---\n\n".join(context_parts)[:6000]
|
|
|
+
|
|
|
+ messages = [
|
|
|
+ {
|
|
|
+ "role": "system",
|
|
|
+ "content": (
|
|
|
+ "你是一个知识库问答助手。请严格基于以下检索到的参考资料回答问题,"
|
|
|
+ "若资料中无相关信息请如实说明,回答时用「[来源N]」标注依据。"
|
|
|
+ ),
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "role": "user",
|
|
|
+ "content": f"## 参考资料\n{context}\n\n## 问题\n{q}\n\n## 回答",
|
|
|
+ },
|
|
|
+ ]
|
|
|
+
|
|
|
+ # ── Step 3: resolve provider + stream ──
|
|
|
+ yield f"data: {json.dumps({'type': 'status', 'text': '正在生成回答...'})}\n\n"
|
|
|
+ await asyncio.sleep(0)
|
|
|
+
|
|
|
+ try:
|
|
|
+ chat_url, llm_headers, model_name, display_label = _resolve_llm(provider_id, model)
|
|
|
+ except Exception as e:
|
|
|
+ err_msg = f"提供商配置错误: {e}"
|
|
|
+ yield f"data: {json.dumps({'type': 'chunk', 'text': err_msg})}\n\n"
|
|
|
+ yield f"data: {json.dumps({'type': 'done'})}\n\n"
|
|
|
+ return
|
|
|
+
|
|
|
+ yield f"data: {json.dumps({'type': 'status', 'text': f'模型: {display_label}'})}\n\n"
|
|
|
+ await asyncio.sleep(0)
|
|
|
+
|
|
|
+ # Determine stream generator based on provider type
|
|
|
+ from agentpaas.api.v1.providers_api import DEFAULT_PROVIDERS
|
|
|
+ pdef = DEFAULT_PROVIDERS.get(provider_id, {})
|
|
|
+ is_anthropic = pdef.get("type") == "anthropic"
|
|
|
+
|
|
|
+ def _do_stream():
|
|
|
+ if is_anthropic:
|
|
|
+ yield from _stream_anthropic(chat_url, llm_headers, model_name, messages)
|
|
|
+ else:
|
|
|
+ yield from _stream_openai_compat(chat_url, llm_headers, model_name, messages)
|
|
|
+
|
|
|
+ try:
|
|
|
+ loop = asyncio.get_event_loop()
|
|
|
+ # Use asyncio.Queue so the thread can push tokens without blocking the event loop
|
|
|
+ aio_q: asyncio.Queue = asyncio.Queue()
|
|
|
+
|
|
|
+ def _fill_q():
|
|
|
+ """Runs in a thread — pushes (kind, val) into the asyncio queue."""
|
|
|
+ try:
|
|
|
+ for delta in _do_stream():
|
|
|
+ loop.call_soon_threadsafe(aio_q.put_nowait, ("chunk", delta))
|
|
|
+ loop.call_soon_threadsafe(aio_q.put_nowait, ("done", None))
|
|
|
+ except Exception as exc:
|
|
|
+ loop.call_soon_threadsafe(aio_q.put_nowait, ("error", str(exc)))
|
|
|
+
|
|
|
+ # Start the blocking stream in a thread
|
|
|
+ stream_fut = loop.run_in_executor(None, _fill_q)
|
|
|
+
|
|
|
+ # Drain the async queue in the event loop
|
|
|
+ while True:
|
|
|
+ kind, val = await aio_q.get()
|
|
|
+ if kind == "chunk":
|
|
|
+ yield f"data: {json.dumps({'type': 'chunk', 'text': val})}\n\n"
|
|
|
+ elif kind == "error":
|
|
|
+ err_msg = f"\n\n(LLM 错误: {val})"
|
|
|
+ yield f"data: {json.dumps({'type': 'chunk', 'text': err_msg})}\n\n"
|
|
|
+ break
|
|
|
+ else: # done
|
|
|
+ break
|
|
|
+
|
|
|
+ await stream_fut # surface thread exceptions
|
|
|
+ except Exception as e:
|
|
|
+ err_msg = f"\n\n(LLM 错误: {e})"
|
|
|
+ yield f"data: {json.dumps({'type': 'chunk', 'text': err_msg})}\n\n"
|
|
|
+
|
|
|
+ yield f"data: {json.dumps({'type': 'done'})}\n\n"
|
|
|
+
|
|
|
+ return StreamingResponse(
|
|
|
+ event_stream(),
|
|
|
+ media_type="text/event-stream",
|
|
|
+ headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+# ─── Jobs ───
|
|
|
+
|
|
|
+@router.get("/{kb_id}/jobs")
|
|
|
+async def list_jobs(
|
|
|
+ kb_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
|
|
|
+ if not kb:
|
|
|
+ raise HTTPException(status_code=404, detail="Knowledge base not found")
|
|
|
+
|
|
|
+ jobs = db.fetchall(
|
|
|
+ "SELECT * FROM kb_jobs WHERE kb_id=? ORDER BY created_at DESC LIMIT 20",
|
|
|
+ (kb_id,)
|
|
|
+ )
|
|
|
+ return {"jobs": jobs}
|
|
|
+
|
|
|
+
|
|
|
+@router.get("/{kb_id}/jobs/{job_id}")
|
|
|
+async def get_job(
|
|
|
+ kb_id: str,
|
|
|
+ job_id: str,
|
|
|
+ tenant: TenantContext = Depends(get_tenant),
|
|
|
+ db: Database = Depends(get_database),
|
|
|
+):
|
|
|
+ job = db.fetchone("SELECT * FROM kb_jobs WHERE id=? AND kb_id=?", (job_id, kb_id))
|
|
|
+ if not job:
|
|
|
+ raise HTTPException(status_code=404, detail="Job not found")
|
|
|
+ return job
|