feat(M2): Phase E — citation-aware ingestion (PDF pages + Obsidian + cite format)

CR-20260607-001 rev.2 §6.2 (FR-004/005) + Q7 (Obsidian). 让检索结果携带
来源文件 + 页码, 报告能引用 `paper.pdf p.12`。184 test 全过 (+9)。

## lambdagent/ingest.py (NEW, ~210 LOC)

provenance-preserving 摄取的三块积木, 纯逻辑、依赖轻:

- extract_pdf_pages(path) → [(page_no_1indexed, text), ...]
  pdfplumber 优先 (layout-aware), PyPDF2 兜底, 都没有则清晰报错。
  空页也返回 (保持页码与物理文档对齐)。
- chunk_text / chunk_pages → CitedChunk
  CitedChunk 携带 {text, source, page, chunk_index, extra}。
  chunk_pages 保证一个 chunk 不跨页 → 页码永远精确。
  to_metadata() 产出 rag 可直接用的 {source, page} dict。
- import_obsidian_vault(dir) → [CitedChunk] (Q7)
  vault 就是 markdown 文件夹, 零集成成本。walk .md, source=vault 相对
  路径, page=None。`[[Note|alias]]` 正文显示 alias, target 进
  extra['links'] 供未来建链接图。跳过 .obsidian / 隐藏目录 / 非 md。

导出 +6 symbol (157 → 163)。pdfplumber 加进 knowledge extra。

## rag.py 引用格式 (FR-005)

RAGTool numbered 格式新增 _cite(): metadata 有 source(+page) 时渲染
`[Source 1 | paper.pdf p.12, score=0.85]`; markdown 无页码渲染
`| note.md`; 其他 metadata 退到紧凑 k=v; 无则空。之前是把所有 metadata
原样倒成 (k=v) 噪音。

## 测试 (tests/test_ingest.py, NEW, 9 test)

- chunk_text: short/empty/long-overlap
- chunk_pages: 页码保留 + chunk 不跨页 + chunk_index 全局递增
- CitedChunk.to_metadata: paged vs non-paged
- Obsidian import: 相对路径 source / 跳过 .obsidian+png / wikilink alias
  展开 + target 进 links / 缺目录报错
- RAGTool: PDF chunk 渲染 `| qgt.pdf p.12` / markdown 渲染 `| note.md`
  无 `p.None`

extract_pdf_pages 本身需要 pdfplumber + 真 PDF, 故测 chunker (用合成
page 元组, 真正承载 provenance 的部分) + Obsidian (真 tmp .md)。

## 集成边界 (诚实标注)

本 phase 交付 lambdagent 层的 citation 原语 + rag 格式, 已测。把它们接进
agentpaas 的 KB index 构建管线 (qaagent67lambda 的 subprocess + pickle
索引, audit #17 标记的子系统) 是后续集成步骤 — 那条管线自成体系, 单独
跟进。3 个研究 pack 当前通过 KBSearch 拿 agentpaas KB 结果; 等 KB 管线
接上 ingest 后才会自动带页码。

M2 三个 phase (D/E/F) 全部完成。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lambdagent/pyproject.toml

@@ -42,6 +42,7 @@ rag = ["chromadb>=0.4.0"]
 # Knowledge / document ingestion (PDF, DOCX, OCR, HTML)
 knowledge = [
     "PyPDF2>=3.0.0",
+    "pdfplumber>=0.10.0",
     "python-docx>=1.0.0",
     "markdown>=3.4.0",
     "markdownify>=0.11.0",

lambdagent/src/lambdagent/__init__.py

@@ -132,11 +132,19 @@ from .agentpack import (
     AgentPackManifest, PackPermissions, AgentPackError,
     load_manifest, validate_manifest_dict,
 )
+# M2 Phase E: citation-aware ingestion (CR §6.2 FR-004/005 + Q7 Obsidian)
+from .ingest import (
+    CitedChunk, extract_pdf_pages, chunk_text, chunk_pages, ingest_pdf,
+    import_obsidian_vault,
+)
 
 __all__ = [
     # AgentPack (M2)
     "AgentPackManifest", "PackPermissions", "AgentPackError",
     "load_manifest", "validate_manifest_dict",
+    # Ingestion (M2 Phase E)
+    "CitedChunk", "extract_pdf_pages", "chunk_text", "chunk_pages",
+    "ingest_pdf", "import_obsidian_vault",
     # 元层级
     "Term", "Context", "TraceEntry",
     # Enhanced trace system

lambdagent/src/lambdagent/ingest.py

@@ -0,0 +1,215 @@
+"""
+lambdagent.ingest — citation-aware document ingestion (M2 Phase E).
+
+CR-20260607-001 rev.2 §6.2 (FR-004/FR-005) + Q7 (Obsidian import).
+
+The research agent packs (Phase F) promise "回答和报告中尽量保留来源文件、
+页码或段落" (FR-005). To deliver that, ingestion must preserve provenance:
+every chunk carries which file + which page it came from, so retrieval can
+render `[Source N | paper.pdf p.12]`.
+
+Three building blocks, all pure + dependency-light:
+  - extract_pdf_pages(path)      → [(page_number, text), ...]   (1-indexed)
+  - chunk_pages(pages, source)   → [CitedChunk, ...]            (carries page)
+  - import_obsidian_vault(dir)   → [CitedChunk, ...]            (markdown vault)
+
+PDF extraction prefers pdfplumber (better layout + reliable page split),
+falls back to PyPDF2, and raises a clear error if neither is installed.
+Obsidian/markdown needs no dependency.
+"""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Tuple
+
+
+@dataclass
+class CitedChunk:
+    """A retrievable chunk that remembers where it came from."""
+    text: str
+    source: str                 # filename or relative path
+    page: Optional[int] = None  # 1-indexed PDF page, or None for non-paged
+    chunk_index: int = 0        # position within the source
+    extra: Dict[str, Any] = field(default_factory=dict)
+
+    def to_metadata(self) -> Dict[str, Any]:
+        """Metadata dict suitable for rag.SimpleVectorStore.add(metadata=...).
+        The keys `source` + `page` are what rag's citation formatter looks
+        for to render `[Source N | source p.page]`."""
+        md: Dict[str, Any] = {"source": self.source, "chunk_index": self.chunk_index}
+        if self.page is not None:
+            md["page"] = self.page
+        md.update(self.extra)
+        return md
+
+
+# ── PDF page extraction ───────────────────────────────────────────────────
+
+def extract_pdf_pages(path: str) -> List[Tuple[int, str]]:
+    """Extract text per page. Returns [(page_number_1indexed, text), ...].
+
+    Prefers pdfplumber (cleaner layout-aware extraction), falls back to
+    PyPDF2. Pages with no extractable text are still returned (empty string)
+    so page numbering stays aligned with the physical document.
+    """
+    if not os.path.isfile(path):
+        raise FileNotFoundError(f"PDF not found: {path}")
+
+    # pdfplumber first
+    try:
+        import pdfplumber  # type: ignore
+        out: List[Tuple[int, str]] = []
+        with pdfplumber.open(path) as pdf:
+            for i, page in enumerate(pdf.pages):
+                out.append((i + 1, page.extract_text() or ""))
+        return out
+    except ImportError:
+        pass
+
+    # PyPDF2 fallback
+    try:
+        import PyPDF2  # type: ignore
+        out = []
+        with open(path, "rb") as f:
+            reader = PyPDF2.PdfReader(f)
+            for i, page in enumerate(reader.pages):
+                out.append((i + 1, page.extract_text() or ""))
+        return out
+    except ImportError:
+        raise RuntimeError(
+            "PDF extraction needs pdfplumber or PyPDF2. "
+            "Install with: pip install pdfplumber"
+        )
+
+
+# ── Page-preserving chunker ────────────────────────────────────────────────
+
+def chunk_text(text: str, chunk_size: int = 800, overlap: int = 100) -> List[str]:
+    """Split text into overlapping windows on paragraph/whitespace
+    boundaries. Pure helper used by chunk_pages."""
+    text = text.strip()
+    if not text:
+        return []
+    if len(text) <= chunk_size:
+        return [text]
+
+    chunks: List[str] = []
+    start = 0
+    n = len(text)
+    while start < n:
+        end = min(start + chunk_size, n)
+        # Try to break on a paragraph or sentence boundary near `end`.
+        if end < n:
+            window = text[start:end]
+            for sep in ("\n\n", "\n", ". ", "。", " "):
+                idx = window.rfind(sep)
+                if idx > chunk_size * 0.5:  # don't break too early
+                    end = start + idx + len(sep)
+                    break
+        chunks.append(text[start:end].strip())
+        if end >= n:
+            break
+        start = max(end - overlap, start + 1)
+    return [c for c in chunks if c]
+
+
+def chunk_pages(
+    pages: List[Tuple[int, str]],
+    source: str,
+    chunk_size: int = 800,
+    overlap: int = 100,
+) -> List[CitedChunk]:
+    """Chunk per-page text into CitedChunks that remember their page.
+
+    A chunk never spans two pages (so the page number is always exact).
+    `chunk_index` is global across the document.
+    """
+    out: List[CitedChunk] = []
+    idx = 0
+    for page_no, page_text in pages:
+        for piece in chunk_text(page_text, chunk_size, overlap):
+            out.append(CitedChunk(
+                text=piece, source=source, page=page_no, chunk_index=idx,
+            ))
+            idx += 1
+    return out
+
+
+def ingest_pdf(path: str, chunk_size: int = 800, overlap: int = 100) -> List[CitedChunk]:
+    """Convenience: extract + chunk a PDF in one call. Source = basename."""
+    pages = extract_pdf_pages(path)
+    return chunk_pages(pages, os.path.basename(path), chunk_size, overlap)
+
+
+# ── Obsidian / markdown vault import (Q7) ──────────────────────────────────
+
+import re as _re
+
+_WIKILINK_RE = _re.compile(r"\[\[([^\]|]+)(?:\|([^\]]+))?\]\]")
+
+
+def _strip_wikilinks(text: str) -> Tuple[str, List[str]]:
+    """Replace Obsidian `[[Note]]` / `[[Note|alias]]` with readable text and
+    return the referenced note targets (for building a link graph later).
+
+    The body keeps the human-facing display text — the alias when present,
+    else the target — so the chunk reads naturally. The link graph keeps the
+    real target (`extra['links']`)."""
+    links: List[str] = []
+
+    def _repl(m):
+        target = m.group(1).strip()
+        alias = (m.group(2) or "").strip()
+        links.append(target)
+        return alias or target
+
+    return _WIKILINK_RE.sub(_repl, text), links
+
+
+def import_obsidian_vault(
+    vault_dir: str,
+    chunk_size: int = 800,
+    overlap: int = 100,
+    include_subdirs: bool = True,
+) -> List[CitedChunk]:
+    """Walk an Obsidian vault (a folder of .md files) and return CitedChunks.
+
+    Q7 decision: Obsidian is a first-class import because a vault is just a
+    markdown folder — zero integration cost. `[[wikilinks]]` are flattened to
+    their display text in the chunk body, and the referenced note names are
+    kept in `extra['links']` so a future pass can build the link graph.
+
+    `source` is the path relative to the vault root (e.g. "notes/qgt.md").
+    Markdown has no pages, so `page` is None.
+    """
+    if not os.path.isdir(vault_dir):
+        raise NotADirectoryError(f"Obsidian vault not found: {vault_dir}")
+
+    vault_abs = os.path.abspath(vault_dir)
+    out: List[CitedChunk] = []
+
+    for root, dirs, files in os.walk(vault_abs):
+        # Skip Obsidian's own config + hidden dirs.
+        dirs[:] = [d for d in dirs if not d.startswith(".")]
+        if not include_subdirs and root != vault_abs:
+            continue
+        for fname in sorted(files):
+            if not fname.lower().endswith((".md", ".markdown")):
+                continue
+            fpath = os.path.join(root, fname)
+            rel = os.path.relpath(fpath, vault_abs)
+            try:
+                with open(fpath, encoding="utf-8") as f:
+                    raw = f.read()
+            except (OSError, UnicodeDecodeError):
+                continue
+            clean, links = _strip_wikilinks(raw)
+            idx = 0
+            for piece in chunk_text(clean, chunk_size, overlap):
+                out.append(CitedChunk(
+                    text=piece, source=rel, page=None, chunk_index=idx,
+                    extra={"links": links} if links else {},
+                ))
+                idx += 1
+    return out

lambdagent/src/lambdagent/rag.py

@@ -328,12 +328,31 @@ class RAGTool(Term):
         else:  # numbered
             parts = []
             for r in results:
-                meta = ""
-                if r.document.metadata:
-                    meta = f" ({', '.join(f'{k}={v}' for k,v in r.document.metadata.items())})"
-                parts.append(f"[Source {r.rank}, score={r.score:.3f}{meta}]\n{r.document.content}")
+                parts.append(f"[Source {r.rank}{self._cite(r.document.metadata)}, "
+                             f"score={r.score:.3f}]\n{r.document.content}")
             return "\n\n".join(parts)
 
+    @staticmethod
+    def _cite(metadata: Dict[str, Any]) -> str:
+        """Render a clean citation suffix (M2 Phase E / FR-005).
+
+        When a chunk carries `source` (+ optional `page`) — as produced by
+        lambdagent.ingest — render `| paper.pdf p.12` so the agent's report
+        can cite a real file + page. Falls back to a compact key=value dump
+        for other metadata, and to nothing when there's no metadata.
+        """
+        if not metadata:
+            return ""
+        source = metadata.get("source")
+        if source:
+            page = metadata.get("page")
+            return f" | {source} p.{page}" if page is not None else f" | {source}"
+        # Generic fallback (skip noisy internal keys).
+        shown = {k: v for k, v in metadata.items() if k != "chunk_index"}
+        if not shown:
+            return ""
+        return " (" + ", ".join(f"{k}={v}" for k, v in shown.items()) + ")"
+
 
 # ════════════════════════════════════════════════════════════
 # AgenticRAG: Agent 自行决定何时检索

tests/test_ingest.py

@@ -0,0 +1,147 @@
+"""
+Tests for citation-aware ingestion (M2 Phase E).
+
+lambdagent.ingest — page-preserving chunker + Obsidian import + the
+rag.RAGTool `[Source N | file p.12]` citation format (FR-005).
+
+PDF extraction itself (extract_pdf_pages) needs pdfplumber/PyPDF2 + a real
+PDF, so we test the chunker against synthetic page tuples (the part that
+actually carries provenance) and exercise Obsidian import with real temp
+markdown files.
+"""
+from __future__ import annotations
+
+import os
+
+import pytest
+
+from lambdagent.ingest import (
+    CitedChunk,
+    chunk_text,
+    chunk_pages,
+    import_obsidian_vault,
+)
+from lambdagent.rag import RAGTool, SimpleVectorStore
+
+
+# ── chunk_text ──────────────────────────────────────────────────────────────
+
+def test_chunk_text_short_returns_single():
+    assert chunk_text("short text", chunk_size=800) == ["short text"]
+
+
+def test_chunk_text_empty_returns_empty():
+    assert chunk_text("") == []
+    assert chunk_text("   \n  ") == []
+
+
+def test_chunk_text_splits_long_with_overlap():
+    text = "para one.\n\n" + ("word " * 400)  # well over 800 chars
+    chunks = chunk_text(text, chunk_size=300, overlap=50)
+    assert len(chunks) > 1
+    # No chunk wildly exceeds chunk_size (allow some boundary slack).
+    assert all(len(c) <= 300 + 50 for c in chunks)
+
+
+# ── chunk_pages: provenance preserved ───────────────────────────────────────
+
+def test_chunk_pages_preserves_page_numbers():
+    pages = [
+        (1, "Introduction. " + "alpha " * 200),
+        (2, "Methods. " + "beta " * 200),
+        (3, ""),  # blank page — produces no chunks but doesn't break numbering
+        (4, "Results. " + "gamma " * 200),
+    ]
+    chunks = chunk_pages(pages, source="paper.pdf", chunk_size=300, overlap=40)
+    assert chunks, "expected chunks"
+    # Every chunk knows its source + a real page number.
+    assert all(c.source == "paper.pdf" for c in chunks)
+    assert all(c.page in (1, 2, 4) for c in chunks)
+    # A chunk never spans pages: page-1 chunks only contain alpha, etc.
+    for c in chunks:
+        if c.page == 1:
+            assert "alpha" in c.text and "beta" not in c.text
+        if c.page == 2:
+            assert "beta" in c.text and "gamma" not in c.text
+    # chunk_index is globally increasing.
+    idxs = [c.chunk_index for c in chunks]
+    assert idxs == sorted(idxs)
+    assert len(set(idxs)) == len(idxs)
+
+
+def test_cited_chunk_to_metadata():
+    c = CitedChunk(text="x", source="paper.pdf", page=12, chunk_index=3)
+    md = c.to_metadata()
+    assert md["source"] == "paper.pdf"
+    assert md["page"] == 12
+    assert md["chunk_index"] == 3
+
+    # Non-paged source (markdown) omits page.
+    c2 = CitedChunk(text="y", source="note.md", page=None)
+    assert "page" not in c2.to_metadata()
+
+
+# ── Obsidian vault import ────────────────────────────────────────────────────
+
+def test_import_obsidian_vault(tmp_path):
+    vault = tmp_path / "vault"
+    (vault / "notes").mkdir(parents=True)
+    (vault / "index.md").write_text("# Index\nSee [[notes/qgt|the QGT note]] for details.")
+    (vault / "notes" / "qgt.md").write_text("# QGT\nQuantum geometric tensor basics. " * 5)
+    # Obsidian config dir + a non-md file must be ignored.
+    (vault / ".obsidian").mkdir()
+    (vault / ".obsidian" / "app.json").write_text("{}")
+    (vault / "image.png").write_bytes(b"\x89PNG")
+
+    chunks = import_obsidian_vault(str(vault))
+    sources = {c.source for c in chunks}
+    # Only the two .md files, with vault-relative paths.
+    assert "index.md" in sources
+    assert os.path.join("notes", "qgt.md") in sources
+    assert not any(".obsidian" in s for s in sources)
+    assert not any(s.endswith(".png") for s in sources)
+    # Markdown chunks have no page.
+    assert all(c.page is None for c in chunks)
+    # Wikilink was flattened to its alias text + recorded in extra.links.
+    index_chunks = [c for c in chunks if c.source == "index.md"]
+    assert index_chunks
+    assert "the QGT note" in index_chunks[0].text
+    assert "[[" not in index_chunks[0].text
+    assert "notes/qgt" in index_chunks[0].extra.get("links", [])
+
+
+def test_import_obsidian_missing_dir_raises(tmp_path):
+    with pytest.raises(NotADirectoryError):
+        import_obsidian_vault(str(tmp_path / "nope"))
+
+
+# ── RAGTool citation format (FR-005) ─────────────────────────────────────────
+
+def test_ragtool_renders_pdf_citation():
+    """A chunk ingested from a PDF (metadata source+page) must render as
+    `[Source N | paper.pdf p.12]` — the headline FR-005 deliverable."""
+    store = SimpleVectorStore()
+    store.add(
+        "The quantum geometric tensor encodes the metric and Berry curvature.",
+        metadata=CitedChunk(text="", source="qgt.pdf", page=12).to_metadata(),
+    )
+    store.add(
+        "Unrelated content about cats and dogs.",
+        metadata=CitedChunk(text="", source="pets.pdf", page=3).to_metadata(),
+    )
+    rag = RAGTool(store, top_k=1, format="numbered")
+    out = rag.apply("quantum geometric tensor metric Berry curvature")
+    assert "| qgt.pdf p.12" in out, out
+
+
+def test_ragtool_renders_source_without_page():
+    """Markdown chunk (no page) renders `| note.md`, no `p.None`."""
+    store = SimpleVectorStore()
+    store.add(
+        "Obsidian note about lambda calculus and Y combinators.",
+        metadata=CitedChunk(text="", source="lambda.md", page=None).to_metadata(),
+    )
+    rag = RAGTool(store, top_k=1, format="numbered")
+    out = rag.apply("lambda calculus Y combinator")
+    assert "| lambda.md" in out
+    assert "p.None" not in out