Phase 7: the extractor

README.md

@@ -145,11 +145,12 @@ sample B on 2026-10-14. The reader and the judge were `gpt-4o-2024-08-06`.
 |---|---|---|
 | Every session of the question, about 114,000 tokens | 39 of 54 (72.2%) | 43 of 54 (79.6%) |
 | The 10 sessions that the search ranked first | 45 of 54 (83.3%) | 48 of 54 (88.9%) |
-| The block of `recall`, after `capture` with a `gpt-4o-mini` extractor | 19 of 54 (35.2%) | 27 of 54 (50.0%) |
+| The block of `recall`, after `capture` with a `gpt-4o-mini` extractor and the `baseline` prompt | 19 of 54 (35.2%) | 27 of 54 (50.0%) |
+| The same, on 2026-10-14, with the `detail` prompt and a session source on each fact | 31 of 54 (57.4%) | 32 of 54 (59.3%) |
 
-The memory contract, with this extractor, scores below the full history. The extractor
-did not store the answer in 19 of the 35 misses on sample A, and in 16 of the 27 misses on
-sample B. With 54 questions, each score can be about 10 points off, and the memory arm
+The memory contract scores below the full history. With the `baseline` prompt, the
+extractor did not store the answer in 19 of the 35 misses on sample A, and in 16 of the 27
+misses on sample B. With the `detail` prompt, these counts are 16 of 23 and 14 of 22. With 54 questions, each score can be about 10 points off, and the memory arm
 differs by 14.8 points between the two samples. In the cloud, a search took 361 ms at p50 with the vector ranking, and
 998 ms with the fused ranking.
 

bench/README.md

@@ -86,6 +86,7 @@ The options are these:
 | `--recall-k` | the `k` of the memory recall | `30` |
 | `--recall-budget` | the `budget` of the memory recall, in characters | `8000` |
 | `--extractor` | a variant name in `VARIANTS` of `bench/prompts.py` | `baseline` |
+| `--no-fact-sources` | write the facts with no session source, as phases 5 and 6 did | off |
 | `--replay` | call no model, and stop at the first reply that the cache does not hold | off |
 | `--workers` | the number of processes that run the questions | `1` |
 | `--tag` | a suffix for the name of the result file | none |
@@ -94,8 +95,14 @@ The options are these:
 Each worker gets the questions in turn, and runs them over a new local database of its
 own. Each table then holds one question at a time, as in a run with one worker. BM25 ranks
 over the whole table before the scope filter, so the rows of a second question in a table
-can change the ranking. An extractor variant names a prompt and a model. The result file is
-`bench/results/e2e-sample-<a or b>[-<tag>].json`, and it records each option.
+can change the ranking. An extractor variant names a prompt and a model. Each fact carries
+the source `session:<hash>`, where the hash is the first 12 hex digits of the SHA-256 of
+the session id. The block shows the sources to the reader, and the id of an evidence
+session holds `answer`, so the source does not show the id. If an extractor reply reaches
+its limit of 4,000 output tokens, the session writes no facts. The cache keeps that reply,
+so a replay skips the same session. The result file counts these sessions in
+`sessions_at_limit`. The result file is `bench/results/e2e-sample-<a or b>[-<tag>].json`,
+and it records each option.
 
 ## The checks of the results
 
@@ -128,4 +135,5 @@ Each miss then gets one of four causes:
 For the last cause, GPT-4o tells whether the marked facts in the block hold every value
 that the answer needs ("all present") or only some of them ("some present"). The command
 prints the groups for each question type, and writes them next to the result file, as
-`<result>-diagnosis.json`.
+`<result>-diagnosis.json`. For each fact that holds the answer, the file gives its session
+id, if the fact has a source.

bench/diagnose.py

@@ -24,7 +24,16 @@
 
 from bench import prompts
 from bench.data import DATA_DIR, DATA_FILE, QUESTION_TYPES, load
-from bench.e2e import ACTOR, BASELINE, RECALL_BUDGET, RECALL_K, Config, extractor, local_store
+from bench.e2e import (
+    ACTOR,
+    BASELINE,
+    RECALL_BUDGET,
+    RECALL_K,
+    Config,
+    extractor,
+    fact_source,
+    local_store,
+)
 from bench.embed import CACHE_DIR
 from bench.models import JUDGE, CachedModels, Offline, Remote
 from bench.retrieval import RANKINGS, RESULTS_DIR, _commit
@@ -103,7 +112,8 @@ def verdict(holding: set[str], best_rank: int | None, block: set[str], k: int =
 def memory_config(result: dict[str, Any]) -> Config:
     """Return the configuration of the memory arm that a result file records.
 
-    A result file of phase 5 records no `memory` section, and its run used `BASELINE`.
+    A result file of phase 5 records no `memory` section, and its run used `BASELINE`. A
+    result file of phase 6 records no `fact_sources`, and its facts carried no source.
     """
     found = result["config"].get("memory")
     if found is None:
@@ -113,6 +123,7 @@ def memory_config(result: dict[str, Any]) -> Config:
         k=found["k"],
         budget=found["budget"],
         extractor=found["extractor"],
+        fact_sources=found.get("fact_sources", False),
     )
 
 
@@ -154,15 +165,16 @@ def main(argv: Sequence[str] | None = None) -> Path:
     run_id = f"diagnose-{datetime.now(UTC):%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}"
     with local_store(args.url, run_id) as (stores, empty):
         memory = Memory(stores[config.memory_ranking])
-        extract = extractor(replay, config.variant)
         for count, question_id in enumerate(missed, start=1):
             question = questions[question_id]
             namespace = ("bench", "lme_s", run_id, question.id)
+            sessions = {fact_source(session.id): session.id for session in question.sessions}
             for session in question.sessions:
+                source = fact_source(session.id) if config.fact_sources else ""
                 memory.capture(
                     prompts.session_text(session),
                     namespace,
-                    extract,
+                    extractor(replay, config.variant, source, []),
                     actor=ACTOR,
                     observed_at=session.date,
                 )
@@ -189,6 +201,7 @@ def main(argv: Sequence[str] | None = None) -> Path:
                     "type": question.type,
                     "facts": len(facts),
                     "holding": [facts[n - 1].content for n in marked],
+                    "holding_sessions": [session_of_fact(facts[n - 1], sessions) for n in marked],
                     "best_rank": ranks,
                     "cause": cause,
                     "present": present,
@@ -207,6 +220,7 @@ def main(argv: Sequence[str] | None = None) -> Path:
                 "k": config.k,
                 "budget": config.budget,
                 "extractor": config.extractor,
+                "fact_sources": config.fact_sources,
             },
             "marker": JUDGE,
             "hotmemory_commit": commit,
@@ -224,11 +238,24 @@ def main(argv: Sequence[str] | None = None) -> Path:
     return path
 
 
+def session_of_fact(fact: Record, sessions: dict[str, str]) -> str | None:
+    """Return the session id of `fact` from its sources, or None if no source names one.
+
+    `sessions` maps the result of `fact_source` to the session id.
+    """
+    for source in fact.sources:
+        if source in sessions:
+            return sessions[source]
+    return None
+
+
 def _mark(marker: CachedModels, question: str, answer: str, facts: Sequence[Record]) -> list[int]:
     """Return the numbers, from 1, of the facts that the marker says hold the answer."""
     numbered = "\n".join(f"{n}. {fact.content}" for n, fact in enumerate(facts, 1))
     prompt = MARK.format(question=question, answer=answer, facts=numbered)
     reply = marker(JUDGE, prompt, 1000, MARK_SCHEMA)
+    if reply.stopped:
+        raise RuntimeError(f"{JUDGE} stopped at 1000 tokens")
     return sorted({n for n in json.loads(reply.text)["facts"] if 1 <= n <= len(facts)})
 
 
@@ -237,6 +264,8 @@ def _complete(marker: CachedModels, question: str, answer: str, facts: Sequence[
     listed = "\n".join(f"- {fact}" for fact in facts)
     prompt = COMPLETE.format(question=question, answer=answer, facts=listed)
     reply = marker(JUDGE, prompt, 100, COMPLETE_SCHEMA)
+    if reply.stopped:
+        raise RuntimeError(f"{JUDGE} stopped at 100 tokens")
     return bool(json.loads(reply.text)["complete"])
 
 

bench/e2e.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import multiprocessing
 import random
@@ -71,13 +72,15 @@ class Config:
     `retrieval_ranking` ranks the sessions of the retrieval arm. The memory arm captures
     and recalls with `memory_ranking`, recalls `k` records into a block of at most
     `budget` characters, and extracts with the variant `extractor` of `prompts.VARIANTS`.
+    With `fact_sources`, each fact carries the source of its session from `fact_source`.
     """
 
     retrieval_ranking: str = "fused"
     memory_ranking: str = "fused"
     k: int = RECALL_K
     budget: int = RECALL_BUDGET
     extractor: str = "baseline"
+    fact_sources: bool = True
 
     def __post_init__(self) -> None:
         for ranking in (self.retrieval_ranking, self.memory_ranking):
@@ -94,7 +97,7 @@ def variant(self) -> prompts.Variant:
         return prompts.VARIANTS[self.extractor]
 
 
-BASELINE = Config()
+BASELINE = Config(fact_sources=False)
 """The configuration of the phase 5 run."""
 
 
@@ -147,19 +150,45 @@ def sample_ids(name: str, questions: Sequence[Question], root: Path = HERE.paren
     return ids
 
 
+def fact_source(session_id: str) -> str:
+    """Return the source of the facts of a session: `session:` and 12 hex digits of its SHA-256.
+
+    The reader sees the sources in the block, and an evidence session id holds `answer`, so
+    the source does not show the id.
+    """
+    return "session:" + hashlib.sha256(session_id.encode()).hexdigest()[:12]
+
+
 def extractor(
-    models: Models, variant: prompts.Variant = prompts.VARIANTS["baseline"]
+    models: Models,
+    variant: prompts.Variant = prompts.VARIANTS["baseline"],
+    source: str = "",
+    at_limit: list[str] | None = None,
 ) -> Callable[[str, datetime | None, list[Record]], list[Fact]]:
-    """Return the extractor callable of `capture`: one call to the variant's model per session."""
+    """Return the extractor callable of `capture`: one call to the variant's model per session.
+
+    With `source`, each fact carries it in `sources`. A reply that reached the token limit
+    gives no facts, and `source` is appended to `at_limit`.
+    """
+    sources = (source,) if source else ()
 
     def extract(text: str, observed_at: datetime | None, current: list[Record]) -> list[Fact]:
         date = observed_at.strftime("%Y/%m/%d (%a) %H:%M") if observed_at else "unknown"
         known = [f"- {record.content}" for record in current]
         prompt = prompts.extract(text, date, known, variant.prompt)
         reply = models(variant.model, prompt, EXTRACTOR_MAX_TOKENS, prompts.EXTRACT_SCHEMA)
+        if reply.stopped:
+            if at_limit is not None:
+                at_limit.append(source)
+            return []
         facts = json.loads(reply.text)["facts"]
         return [
-            Fact(kind="fact", subject=item["subject"].strip(), content=item["content"].strip())
+            Fact(
+                kind="fact",
+                subject=item["subject"].strip(),
+                content=item["content"].strip(),
+                sources=sources,
+            )
             for item in facts
             if item["content"].strip()
         ]
@@ -175,19 +204,25 @@ class Arm:
     answer: Reply
     verdict: Reply
     extracted: int = 0
+    at_limit: int = 0
 
     @property
     def correct(self) -> bool:
         return "yes" in self.verdict.text.lower()
 
 
 def answer(
-    models: Models, question: Question, prompt: str, context_chars: int, extracted: int = 0
+    models: Models,
+    question: Question,
+    prompt: str,
+    context_chars: int,
+    extracted: int = 0,
+    at_limit: int = 0,
 ) -> Arm:
     """Read `prompt` with READER, judge the answer with JUDGE, and return the arm."""
     reply = models(READER, prompt, READER_MAX_TOKENS)
     verdict = models(JUDGE, prompts.judge(question, reply.text), JUDGE_MAX_TOKENS)
-    return Arm(context_chars, reply, verdict, extracted)
+    return Arm(context_chars, reply, verdict, extracted, at_limit)
 
 
 def run_question(
@@ -220,14 +255,15 @@ def run_question(
     if "memory" in arms:
         namespace = ("bench", "lme_s", run_id, "memory", question.id)
         memory = Memory(stores[config.memory_ranking])
-        extract = extractor(models, config.variant)
         written = 0
+        at_limit: list[str] = []
         for session in question.sessions:
+            source = fact_source(session.id) if config.fact_sources else ""
             written += len(
                 memory.capture(
                     prompts.session_text(session),
                     namespace,
-                    extract,
+                    extractor(models, config.variant, source, at_limit),
                     actor=ACTOR,
                     observed_at=session.date,
                 )
@@ -237,7 +273,7 @@ def run_question(
         )
         empty(namespace)
         prompt = prompts.read_facts(question, block)
-        result["memory"] = answer(models, question, prompt, len(block), written)
+        result["memory"] = answer(models, question, prompt, len(block), written, len(at_limit))
     return result
 
 
@@ -396,6 +432,7 @@ def row_of(question: Question, result: dict[str, Arm]) -> dict[str, Any]:
                 "correct": found.correct,
                 "context_chars": found.context_chars,
                 "facts_written": found.extracted,
+                "sessions_at_limit": found.at_limit,
                 "reader_input_tokens": found.answer.input_tokens,
                 "answer": found.answer.text,
                 "verdict": found.verdict.text,
@@ -454,6 +491,11 @@ def main(argv: Sequence[str] | None = None) -> Path | None:
         "--recall-budget", type=int, default=RECALL_BUDGET, help="characters of the memory block"
     )
     parser.add_argument("--extractor", choices=tuple(prompts.VARIANTS), default="baseline")
+    parser.add_argument(
+        "--no-fact-sources",
+        action="store_true",
+        help="write the facts with no session source, as phases 5 and 6 did",
+    )
     parser.add_argument(
         "--replay", action="store_true", help="call no model: use cached replies, stop at a miss"
     )
@@ -478,6 +520,7 @@ def main(argv: Sequence[str] | None = None) -> Path | None:
             k=args.recall_k,
             budget=args.recall_budget,
             extractor=args.extractor,
+            fact_sources=not args.no_fact_sources,
         )
     except ValueError as error:
         parser.error(str(error))
@@ -560,6 +603,7 @@ def main(argv: Sequence[str] | None = None) -> Path | None:
                 "k": config.k,
                 "budget": config.budget,
                 "extractor": config.extractor,
+                "fact_sources": config.fact_sources,
             },
             "arms": {
                 "full": "every session of the question",
@@ -580,6 +624,11 @@ def main(argv: Sequence[str] | None = None) -> Path | None:
             "workers": len(shards),
             "spent_usd": round(spent, 2),
             "model_calls": calls,
+            "sessions_at_limit": sum(
+                row["arms"]["memory"]["sessions_at_limit"]
+                for row in rows
+                if "memory" in row["arms"]
+            ),
             "embedder_misses": misses,
             "hotmemory_commit": commit,
             "run_id": run_id,

bench/models.py

@@ -25,11 +25,16 @@
 
 @dataclass(frozen=True)
 class Reply:
-    """The text of one model reply and its token counts."""
+    """The text of one model reply and its token counts.
+
+    `stopped` is True if a reply with a schema reached its token limit, so its text is not
+    whole JSON.
+    """
 
     text: str
     input_tokens: int
     output_tokens: int
+    stopped: bool = False
 
     def cost(self, model: str) -> float:
         """Return the price of this reply in US dollars."""
@@ -106,7 +111,8 @@ class Remote:
     """`Complete` over the OpenAI chat API, at temperature 0, with 120 seconds for each try.
 
     The client reads the key from `OPENAI_MODEL_KEY`. `OPENAI_API_KEY` stays the key of the
-    embedder. With a schema, the reply is JSON that the schema allows.
+    embedder. With a schema, the reply is JSON that the schema allows, or a reply with
+    `stopped` set if it reached `max_tokens`.
     """
 
     def __init__(self) -> None:
@@ -134,11 +140,10 @@ def __call__(
             **extra,
         )
         choice = completion.choices[0]
-        if choice.finish_reason == "length" and schema is not None:
-            raise RuntimeError(f"{model} stopped at {max_tokens} tokens")
         usage = completion.usage
         return Reply(
             text=(choice.message.content or "").strip(),
             input_tokens=usage.prompt_tokens if usage else 0,
             output_tokens=usage.completion_tokens if usage else 0,
+            stopped=choice.finish_reason == "length" and schema is not None,
         )