基于 Karpathy LLM Wiki + DeepAgents 的本地知识库问答系统

一、背景与核心理念

1.1 传统 RAG 的根本缺陷

当前主流知识库方案（ChatGPT 文件上传、NotebookLM 等）均基于 RAG 模式：上传文档 → 查询时检索片段 → 生成答案。其致命问题是知识不会积累——每次提问，LLM 都从零重新发现知识，无状态、非积累，交叉引用、矛盾标注、综合分析从未被沉淀。

1.2 Karpathy LLM Wiki 核心思想

Karpathy 的比喻：Obsidian 是 IDE，LLM 是程序员，Wiki 是代码库。知识只被"编译"一次，然后持续保持更新。每增加一份资料、每提出一个问题，Wiki 就变得更丰富——知识复利增长。

1.3 DeepAgents 框架定位

DeepAgents 是 LangChain 生态中的生产级"智能体执行环境（Agent Harness）"，构建于 LangGraph 运行时之上。其核心方法 create_deep_agent 在标准工具调用循环之上，内置四大能力：

二、系统整体架构

2.1 三层存储架构

2.2 辅助文件

三、核心工作流详解

3.1 Ingest（录入）工作流

用户上传文档后，系统自动触发 Ingest 流程，将原始资料"编译"进 Wiki。

3.2 Query（问答）工作流

两阶段自动执行：分析（只读）→ 归档（条件写入）。

3.3 Lint（体检）工作流

定期或手动触发，维护 Wiki 健康度。

• 检测页面间矛盾与过时信息
• 发现孤儿页面（无入链）和缺失交叉引用
• 识别提到但未建页面的重要概念
• 建议新研究方向和资料来源
• 直接修复问题并刷新 index.md + log.md

四、核心代码实现

4.1 项目结构

4.2 DeepAgent 创建与编排

核心入口：使用 create_deep_agent 构建具备文件系统、任务规划、子 Agent 能力的深度智能体。

# agent.py
from pathlib import Path
from deepagents import create_deep_agent
from deepagents.middleware.filesystem import FilesystemPermission
from deepagents.backends import FilesystemBackend

DATA_DIR = Path("./data")
WIKI_DIR = DATA_DIR / "wiki"
RAW_DIR  = DATA_DIR / "raw"

WIKI_SYSTEM_PROMPT = """You are a knowledge base curator maintaining a structured Wiki.

## Your Responsibilities
- INGEST: Read raw source documents, extract key entities/concepts/relations,
  and integrate them into the existing Wiki pages. Update cross-references,
  flag contradictions, and create new pages as needed.
- QUERY: Answer questions by first reading wiki/index.md for routing,
  then diving into relevant canonical wiki pages. Provide grounded answers
  with file-path citations. Decide whether the answer should be filed as
  a durable wiki/query/ page.
- LINT: Health-check the wiki — find contradictions, orphan pages,
  missing cross-references, outdated claims. Fix issues directly.

## Directory Layout
- /raw/       : Immutable source documents (READ ONLY, never modify)
- /wiki/      : LLM-maintained knowledge pages (you write here)
- /wiki/index.md  : Content catalog (update after every change)
- AGENTS.md   : Schema and workflow rules (follow strictly)
- log.md      : Append-only timeline (DO NOT edit; runner manages this)

## Wiki Page Format
Each wiki page MUST follow this structure:
```
# Title
> One-line summary

**Tags:** tag1, tag2
**Sources:** /raw/filename.ext

## Content
...
## See Also
- [[related-page]]
```

## Evidence Policy
- /log.md → operational recency context only, NOT primary evidence
- /wiki/query/*.md → routing hints, NOT primary evidence
- Canonical /wiki/*.md pages → primary evidence for all claims
"""

def build_agent(model: str = "openai:gpt-4.1"):
    """Create the deep agent with filesystem access to the local wiki."""
    return create_deep_agent(
        model=model,
        tools=[],
        system_prompt=WIKI_SYSTEM_PROMPT,
        middleware=[
            # FilesystemMiddleware is auto-injected by create_deep_agent
            # with read/write access to the workspace directory
        ],
        checkpointer=None,  # Use SqliteSaver for persistence in production
    )

def run_ingest(workspace_dir: Path, topic: str, model: str | None) -> str:
    """Run ingest mode: agent reads raw sources and updates wiki."""
    agent = build_agent(model or "openai:gpt-4.1")
    prompt = (
        f"INGEST mode for topic '{topic}'.\n"
        f"Read all files in /raw/ that have not yet been processed.\n"
        f"For each source: extract entities, concepts, and relations;\n"
        f"update or create wiki pages under /wiki/; update /wiki/index.md.\n"
        f"Do NOT modify /raw/ or log.md."
    )
    result = agent.invoke(
        {"messages": [{"role": "user", "content": prompt}]},
        config={"configurable": {"thread_id": f"ingest-{topic}"}}
    )
    return result["messages"][-1].content

def run_query(workspace_dir: Path, topic: str, question: str,
              model: str | None) -> str:
    """Run query mode: agent answers a question grounded in wiki pages."""
    agent = build_agent(model or "openai:gpt-4.1")
    prompt = (
        f"QUERY mode for topic '{topic}'.\n"
        f"Question: {question}\n\n"
        f"Workflow:\n"
        f"1) Read /wiki/index.md first for routing.\n"
        f"2) Read recent /log.md entries for recency context.\n"
        f"3) Check /wiki/query/*.md pages as discovery hints.\n"
        f"4) Read canonical /wiki/ pages for grounded evidence.\n"
        f"5) Provide answer with wiki file path citations.\n"
        f"6) Decide: FILE (durable reuse) or SKIP (ad-hoc).\n\n"
        f"Output format:\n"
        f"ANSWER:\n<markdown answer>\n\n"
        f"FILING_DECISION: file|skip\n"
        f"FILING_REASON: <one sentence>"
    )
    result = agent.invoke(
        {"messages": [{"role": "user", "content": prompt}]},
        config={"configurable": {"thread_id": f"query-{topic}"}}
    )
    return result["messages"][-1].content

def run_lint(workspace_dir: Path, topic: str, model: str | None) -> str:
    """Run lint mode: agent health-checks the wiki."""
    agent = build_agent(model or "openai:gpt-4.1")
    prompt = (
        f"LINT mode for topic '{topic}'.\n"
        f"Health-check the wiki:\n"
        f"- Find contradictions between pages\n"
        f"- Find orphan pages (no incoming links)\n"
        f"- Find missing cross-references\n"
        f"- Find important concepts mentioned but without pages\n"
        f"- Fix issues directly in /wiki/ pages\n"
        f"- Update /wiki/index.md\n"
        f"Do NOT modify /raw/ or log.md."
    )
    result = agent.invoke(
        {"messages": [{"role": "user", "content": prompt}]},
        config={"configurable": {"thread_id": f"lint-{topic}"}}
    )
    return result["messages"][-1].content

4.3 FastAPI 应用层

提供文档上传、知识问答、Wiki 浏览的 REST API。

# app.py
import shutil
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from uuid import uuid4

from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse
from pydantic import BaseModel

import agent, parser, index, log
from config import settings

app = FastAPI(title="Local KB QA", version="1.0.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"],
                   allow_methods=["*"], allow_headers=["*"])

DATA_DIR = Path(settings.data_dir)
RAW_DIR  = DATA_DIR / "raw"
WIKI_DIR = DATA_DIR / "wiki"
DB_PATH  = Path(settings.db_path)

RAW_DIR.mkdir(parents=True, exist_ok=True)
WIKI_DIR.mkdir(parents=True, exist_ok=True)

@contextmanager
def get_db():
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    conn.execute("""CREATE TABLE IF NOT EXISTS documents (
        id TEXT PRIMARY KEY,
        filename TEXT NOT NULL,
        original_name TEXT NOT NULL,
        file_type TEXT NOT NULL,
        file_size INTEGER NOT NULL,
        uploaded_at TEXT NOT NULL,
        status TEXT DEFAULT 'uploaded',
        raw_path TEXT NOT NULL
    )""")
    conn.commit()
    try:
        yield conn
    finally:
        conn.close()

class QueryRequest(BaseModel):
    question: str
    topic: str = "default"

class QueryResponse(BaseModel):
    answer: str
    filed: bool = False
    filed_path: str | None = None

@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """Upload a document to the raw sources directory."""
    doc_id = uuid4().hex[:12]
    suffix = Path(file.filename).suffix.lower()
    if suffix not in {".pdf", ".md", ".txt", ".docx", ".csv", ".json"}:
        raise HTTPException(400, f"Unsupported file type: {suffix}")

    raw_path = RAW_DIR / f"{doc_id}{suffix}"
    with open(raw_path, "wb") as f:
        shutil.copyfileobj(file.file, f)

    with get_db() as db:
        db.execute(
            "INSERT INTO documents VALUES (?,?,?,?,?,?,?,?)",
            (doc_id, raw_path.name, file.filename, suffix,
             raw_path.stat().st_size,
             datetime.now(timezone.utc).isoformat(),
             "uploaded", str(raw_path))
        )

    parsed_text = parser.parse_file(raw_path)
    parsed_path = RAW_DIR / f"{doc_id}_parsed.md"
    parsed_path.write_text(parsed_text, encoding="utf-8")

    with get_db() as db:
        db.execute("UPDATE documents SET status='parsed' WHERE id=?",
                   (doc_id,))

    try:
        answer = agent.run_ingest(DATA_DIR, "default", settings.model)
        with get_db() as db:
            db.execute("UPDATE documents SET status='ingested' WHERE id=?",
                       (doc_id,))
        log.append_entry(DATA_DIR, "ingest.apply", "applied",
                         metadata={"source": file.filename},
                         summary=f"Ingested {file.filename}")
    except Exception as e:
        with get_db() as db:
            db.execute("UPDATE documents SET status='error' WHERE id=?",
                       (doc_id,))
        log.append_entry(DATA_DIR, "ingest.apply", "error",
                         metadata={"source": file.filename},
                         summary=str(e))

    return {"id": doc_id, "filename": file.filename, "status": "ingested"}

@app.post("/query", response_model=QueryResponse)
async def query_knowledge_base(req: QueryRequest):
    """Query the wiki and return a grounded answer."""
    raw_response = agent.run_query(
        DATA_DIR, req.topic, req.question, settings.model
    )
    decision = query_parse_decision(raw_response)
    answer_text = decision.answer
    filed_path = None

    if decision.should_file:
        slug = query_slug(req.question)
        filed_path = f"/wiki/query/{slug}.md"
        log.append_entry(DATA_DIR, "query.apply", "filed",
                         metadata={"question": req.question,
                                   "path": filed_path},
                         summary=answer_text[:200])

    log.append_entry(DATA_DIR, "query.review",
                     "file" if decision.should_file else "skip",
                     metadata={"question": req.question},
                     summary=decision.reason)

    return QueryResponse(
        answer=answer_text,
        filed=decision.should_file,
        filed_path=filed_path,
    )

@app.post("/lint")
async def lint_wiki(topic: str = "default"):
    """Run a health-check on the wiki."""
    result = agent.run_lint(DATA_DIR, topic, settings.model)
    log.append_entry(DATA_DIR, "lint.apply", "applied",
                     summary=result[:200])
    index.refresh_index(DATA_DIR)
    return {"status": "completed", "summary": result}

@app.get("/wiki/{path:path}")
async def get_wiki_page(path: str):
    """Serve a wiki markdown file."""
    file_path = WIKI_DIR / path
    if not file_path.exists():
        raise HTTPException(404, f"Wiki page not found: {path}")
    return FileResponse(file_path, media_type="text/markdown")

@app.get("/documents")
async def list_documents():
    """List all uploaded documents."""
    with get_db() as db:
        rows = db.execute("SELECT * FROM documents ORDER BY uploaded_at DESC"
                         ).fetchall()
    return [dict(r) for r in rows]

4.4 文档解析器

将不同格式的上传文档统一转换为 Markdown 文本，供 LLM 读取。

# parser.py
from pathlib import Path

def parse_file(path: Path) -> str:
    """Parse a document file into markdown text."""
    suffix = path.suffix.lower()
    parsers = {
        ".md":   _parse_text,
        ".txt":  _parse_text,
        ".csv":  _parse_csv,
        ".json": _parse_json,
        ".pdf":  _parse_pdf,
        ".docx": _parse_docx,
    }
    parser_fn = parsers.get(suffix)
    if not parser_fn:
        raise ValueError(f"Unsupported format: {suffix}")
    return parser_fn(path)

def _parse_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")

def _parse_csv(path: Path) -> str:
    import csv
    rows = []
    with open(path, encoding="utf-8") as f:
        reader = csv.reader(f)
        for row in reader:
            rows.append("| " + " | ".join(row) + " |")
    return "\n".join(rows)

def _parse_json(path: Path) -> str:
    import json
    data = json.loads(path.read_text(encoding="utf-8"))
    return f"```json\n{json.dumps(data, indent=2, ensure_ascii=False)}\n```"

def _parse_pdf(path: Path) -> str:
    import fitz  # PyMuPDF
    doc = fitz.open(str(path))
    pages = []
    for i, page in enumerate(doc):
        text = page.get_text()
        if text.strip():
            pages.append(f"## Page {i+1}\n\n{text}")
    doc.close()
    return "\n\n".join(pages)

def _parse_docx(path: Path) -> str:
    from docx import Document
    doc = Document(str(path))
    paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
    return "\n\n".join(paragraphs)

4.5 数据模型

# models.py
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

Mode = Literal["init", "ingest", "query", "lint"]

@dataclass(frozen=True)
class RunnerConfig:
    mode: Mode
    topic: str
    sources: tuple[Path, ...]
    question: str | None
    model: str | None
    review: bool

@dataclass(frozen=True)
class RunResult:
    answer: str | None
    hub_url: str | None

@dataclass(frozen=True)
class IngestResult:
    answer: str | None
    should_push: bool

@dataclass(frozen=True)
class QueryDecision:
    answer: str
    should_file: bool
    reason: str

@dataclass(frozen=True)
class QueryResult:
    answer: str
    should_push: bool
    filed_path: str | None

4.6 Query 决策解析

从 LLM 输出中提取归档决策，参考 DeepAgents llm-wiki 示例的正则解析模式。

# query.py (decision parsing)
import re
from models import QueryDecision

_DECISION_PATTERN = re.compile(
    r"^FILING_DECISION:\s*(file|skip)\s*$",
    re.IGNORECASE | re.MULTILINE,
)
_REASON_PATTERN = re.compile(
    r"^FILING_REASON:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)

def query_parse_decision(raw_response: str) -> QueryDecision:
    """Parse filing decision markers from model output."""
    response = raw_response.strip()

    decision_match = _DECISION_PATTERN.search(response)
    reason_match = _REASON_PATTERN.search(response)

    should_file = (
        decision_match is not None
        and decision_match.group(1).lower() == "file"
    )
    reason = (
        reason_match.group(1).strip()
        if reason_match is not None
        else "Decision marker missing; defaulted to skip."
    )

    answer_text = response
    if decision_match is not None:
        answer_text = response[: decision_match.start()].strip()
    if answer_text.upper().startswith("ANSWER:"):
        answer_text = answer_text[len("ANSWER:"):].strip()
    if not answer_text:
        answer_text = response or "No answer returned."

    return QueryDecision(
        answer=answer_text, should_file=should_file, reason=reason
    )

def query_slug(question: str) -> str:
    """Create a stable slug for query filing pages."""
    slug = question.lower().strip()
    slug = re.sub(r"[^\w\s-]", "", slug)
    slug = re.sub(r"[\s_]+", "-", slug)
    return slug[:80].rstrip("-") or "query"

4.7 Index 构建与刷新

# index.py
from pathlib import Path

WIKI_DIR_NAME = "wiki"

def refresh_index(data_dir: Path) -> None:
    """Scan wiki/ and rebuild wiki/index.md with categorized summaries."""
    wiki_dir = data_dir / WIKI_DIR_NAME
    if not wiki_dir.exists():
        return

    categories = {"entities": [], "concepts": [], "summaries": [],
                  "query": [], "other": []}

    for md_file in sorted(wiki_dir.rglob("*.md")):
        if md_file.name == "index.md":
            continue
        rel = md_file.relative_to(wiki_dir)
        content = md_file.read_text(encoding="utf-8")
        summary = _extract_summary(content)
        cat = _categorize(rel)
        categories[cat].append((str(rel), md_file.stem, summary))

    lines = ["# Wiki Index\n"]
    for cat, entries in categories.items():
        if not entries:
            continue
        lines.append(f"\n## {cat.title()}\n")
        for rel, stem, summary in entries:
            lines.append(f"- [{stem}]({rel}) — {summary}\n")

    index_path = wiki_dir / "index.md"
    index_path.write_text("".join(lines), encoding="utf-8")

def _extract_summary(content: str) -> str:
    """Extract the first blockquote line as summary."""
    for line in content.splitlines():
        line = line.strip()
        if line.startswith(">"):
            return line.lstrip(">").strip()
    for line in content.splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            return line[:120]
    return ""

def _categorize(rel: Path) -> str:
    parts = rel.parts
    if len(parts) > 1:
        cat = parts[0].lower()
        if cat in {"entities", "concepts", "summaries", "query"}:
            return cat
    return "other"

4.8 Log 追加

# log.py
from datetime import datetime, timezone
from pathlib import Path

def append_entry(
    data_dir: Path,
    mode_phase: str,
    outcome: str,
    metadata: dict | None = None,
    summary: str | None = None,
) -> None:
    """Append a structured timeline entry to log.md."""
    log_path = data_dir / "log.md"
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    ts_full = datetime.now(timezone.utc).isoformat()

    heading = f"## [{ts}] {mode_phase} | outcome={outcome}"
    if metadata:
        meta_str = " ".join(f"{k}={v}" for k, v in metadata.items())
        heading += f" {meta_str}"

    lines = [f"\n{heading}\n"]
    lines.append(f"- timestamp: {ts_full}\n")
    if summary:
        lines.append(f"- summary: {summary}\n")

    with open(log_path, "a", encoding="utf-8") as f:
        f.writelines(lines)

4.9 配置管理

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    data_dir: str = "./data"
    db_path: str = "./kb.db"
    model: str = "openai:gpt-4.1"
    openai_api_key: str = ""
    host: str = "0.0.0.0"
    port: int = 8000

    class Config:
        env_file = ".env"
        env_prefix = "KB_"

settings = Settings()

五、DeepAgents 中间件与 Wiki 工作流的映射

六、与传统 RAG 方案的对比

七、AGENTS.md 规则文件示例

AGENTS.md 是 LLM Wiki 的"Schema 层"，定义了 Wiki 的组织规则和 Agent 行为约束。参考 DeepAgents llm-wiki 示例的设计。

# AGENTS.md — Wiki Schema & Workflow Rules

## Wiki Organization
- All wiki pages live under /wiki/
- Raw sources live under /raw/ and are NEVER modified by the agent
- Each wiki page covers ONE atomic topic (entity, concept, or theme)
- Use consistent internal format: Title → Summary → Tags → Content → See Also

## Page Format
```markdown
# [Page Title]
> [One-line summary]

**Tags:** tag1, tag2, tag3
**Sources:** /raw/filename.ext

## [Content sections...]

## See Also
- [[related-page-1]]
- [[related-page-2]]
```

## Ingest Rules
1. Read the raw source document completely
2. Extract key entities, concepts, and relationships
3. For each entity/concept: create or update a dedicated wiki page
4. Add cross-references using [[wiki-link]] syntax
5. Flag any contradictions with existing wiki content
6. Update /wiki/index.md with new/modified pages
7. One source typically updates 10-15 wiki pages

## Query Rules
1. ALWAYS read /wiki/index.md first for routing
2. Check recent /log.md entries for recency context
3. Use /wiki/query/*.md pages as discovery hints only
4. Read canonical /wiki/ pages for grounded evidence
5. Cite wiki file paths for all claims
6. Decide FILING_DECISION: file (durable) or skip (ad-hoc)

## Lint Rules
1. Find and fix contradictions between pages
2. Identify orphan pages (no incoming [[links]])
3. Find missing cross-references
4. Detect important concepts mentioned without dedicated pages
5. Suggest new research directions and source materials
6. Fix issues directly in /wiki/ pages
7. Update /wiki/index.md after all changes

## Prohibited Actions
- NEVER write to /raw/
- NEVER edit log.md (the runner manages this)
- NEVER delete wiki pages without explicit user instruction
- NEVER fabricate information not present in the sources

八、部署与运行

8.1 依赖安装

# pyproject.toml 核心依赖
[project]
name = "local-kb-qa"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = [
    "deepagents>=0.1.0",
    "fastapi>=0.115.0",
    "uvicorn>=0.30.0",
    "pydantic-settings>=2.0",
    "pymupdf>=1.24.0",
    "python-docx>=1.1.0",
    "langchain-openai>=0.3.0",
    "langgraph>=0.4.0",
]

8.2 启动命令

# 安装依赖
uv sync

# 配置环境变量
export KB_OPENAI_API_KEY="sk-..."
export KB_MODEL="openai:gpt-4.1"

# 初始化 Wiki 结构
python -c "from pathlib import Path; \
  d=Path('./data'); \
  (d/'raw').mkdir(parents=True,exist_ok=True); \
  (d/'wiki'/  'entities').mkdir(parents=True,exist_ok=True); \
  (d/'wiki'/'concepts').mkdir(parents=True,exist_ok=True); \
  (d/'wiki'/'summaries').mkdir(parents=True,exist_ok=True); \
  (d/'wiki'/'query').mkdir(parents=True,exist_ok=True); \
  (d/'AGENTS.md').exists() or print('Create AGENTS.md first')"

# 启动 API 服务
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

8.3 API 使用示例

# 上传文档
curl -X POST http://localhost:8000/upload \
  -F "file=@research-paper.pdf"

# 知识问答
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "该文档中提到的核心架构模式是什么？", "topic": "default"}'

# Wiki 体检
curl -X POST http://localhost:8000/lint?topic=default

# 浏览 Wiki 页面
curl http://localhost:8000/wiki/index.md

# 查看已上传文档
curl http://localhost:8000/documents

九、架构演进方向

十、参考资源

• Karpathy — LLM Wiki: A pattern for building personal knowledge bases using LLMs
• DeepAgents llm-wiki Example
• DeepAgents Official Documentation
• DeepAgents Quickstart