Build a memory-backed agent

notes/
├── 2026-06-02-kickoff.md
├── 2026-06-16-planning.md
└── 2026-09-17-standup.md

"""Project assistant: answers questions from meeting notes, with citations."""
 
from __future__ import annotations
 
import argparse
import re
import sys
from datetime import UTC, datetime
from pathlib import Path
 
import anthropic
import remember
from remember import ContextBundleV2, MemoryApiError
 
MODEL = "claude-sonnet-5"  # replace with the model you use
SOURCE_KIND = "meeting-notes"
 
SYSTEM_PROMPT = """\
You are the billing migration team's project assistant. You answer only
from the memory context in the user's message; do not fill gaps from
general knowledge.
 
- "Facts" are what the memory holds true. "Source passages" are what
  people said. Never present a passage as an established fact.
- Cite every statement with the IDs in square brackets, such as [C3].
  Use only IDs that appear in the context.
- If the context says nothing is known, say you do not know. If a list is
  marked incomplete, say the answer may be incomplete.
- If facts contradict each other, give both sides with their citations.
- A fact's validity is when it was true; "said on" is when someone said
  it. Do not mix them.
"""
 
# --- memory: ingest ---------------------------------------------------------
 
def meeting_time(path: Path) -> datetime:
    """Meeting date from a 'YYYY-MM-DD-*.md' name, else the file's mtime."""
    match = re.match(r"(\d{4})-(\d{2})-(\d{2})", path.name)
    if match:
        year, month, day = (int(part) for part in match.groups())
        return datetime(year, month, day, tzinfo=UTC)
    return datetime.fromtimestamp(path.stat().st_mtime, tz=UTC)
 
def load_notes(client: remember.Client, folder: Path) -> None:
    """Send every note; wait only for the ones that are new or changed."""
    pending = []
    for path in sorted(folder.glob("*.md")):
        version = client.ingest(
            path,
            source_kind=SOURCE_KIND,
            source_ref=path.name,
            source_modified_at=meeting_time(path),
        )
        state = "new version" if version.created else "unchanged"
        print(f"{path.name}: {state}")
        if version.created:
            pending.append(version.version_id)
 
    if not pending:
        print("Nothing new to process.")
        return
    print(f"Waiting for {len(pending)} version(s); this takes minutes...")
    for start in range(0, len(pending), 1000):
        client.wait_for_readiness(pending[start : start + 1000], timeout=3600)
    print("Ready.")
 
# --- memory: context --------------------------------------------------------
 
def build_context(bundle: ContextBundleV2) -> tuple[str, dict[str, object]]:
    """Render facts and passages with short citation IDs.
 
    Returns the text for the model and a map from citation ID to the claim
    it stands for.
    """
    citations: dict[str, object] = {}
 
    def cite(claim) -> str:
        for key, known in citations.items():
            if known.claim_id == claim.claim_id:
                return key
        key = f"C{len(citations) + 1}"
        citations[key] = claim
        return key
 
    facts = bundle.facts
    lines = ["## Facts the memory holds"]
    if facts.negative is not None:
        lines.append(f"(none: {facts.negative.kind}; {facts.negative.explanation})")
    claims_by_id = {claim.claim_id: claim for claim in facts.evidence}
    for fact in facts.facts:
        v = fact.validity
        when = f"valid from {v.valid_from:%Y-%m-%d}" if v.valid_from else "validity unknown"
        if v.valid_until:
            when += f" until {v.valid_until:%Y-%m-%d}"
        notes = []
        if fact.contradiction is not None:
            rivals = "; ".join(member.label for member in fact.contradiction.co_members)
            notes.append(f"contradicted by: {rivals}")
        if fact.support == "withdrawn":
            notes.append("support withdrawn, unconfirmed")
        refs = [
            cite(claims_by_id[link.claim_id])
            for link in facts.fact_evidence
            if link.fact_id == fact.fact_id and link.claim_id in claims_by_id
        ]
        extra = f"; {'; '.join(notes)}" if notes else ""
        lines.append(f"- {fact.label} ({when}{extra}) [{', '.join(refs)}]")
    if facts.truncation is not None and facts.truncation.truncated:
        lines.append("(This list of facts is incomplete.)")
 
    said = bundle.claims_and_sources
    lines += ["", "## Source passages"]
    if said.negative is not None:
        lines.append(f"(none: {said.negative.kind}; {said.negative.explanation})")
    for claim in said.evidence[:20]:
        when = f"{claim.asserted_at:%Y-%m-%d}" if claim.asserted_at else "undated"
        lines.append(
            f'- [{cite(claim)}] {claim.document_title or claim.doc_id}, said on {when}:'
            f' "{claim.source_span}"'
        )
 
    return "\n".join(lines), citations
 
# --- model -----------------------------------------------------------------
 
def ask_model(question: str, context: str) -> str:
    """The only provider-specific code: send context and question, get text."""
    llm = anthropic.Anthropic()
    response = llm.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[
            {
                "role": "user",
                "content": f"<memory>\n{context}\n</memory>\n\nQuestion: {question}",
            }
        ],
    )
    return "".join(block.text for block in response.content if block.type == "text")
 
# --- answer -----------------------------------------------------------------
 
def answer(client: remember.Client, question: str) -> str:
    bundle = client.combined_context(question)
    context, citations = build_context(bundle)
    reply = ask_model(question, context)
 
    used = sorted(set(re.findall(r"\[?(C\d+)\]?", reply)), key=lambda key: int(key[1:]))
    invented = [key for key in used if key not in citations]
    if invented:
        reply += f"\n\n(Warning: the model cited unknown sources {', '.join(invented)}.)"
 
    sources = ["", "Sources:"]
    for key in used:
        claim = citations.get(key)
        if claim is not None:
            title = claim.document_title or claim.doc_id
            sources.append(f'  [{key}] {title}: "{claim.source_span}"')
    return reply + ("\n" + "\n".join(sources) if len(sources) > 2 else "")
 
def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    load = commands.add_parser("load", help="send a folder of meeting notes")
    load.add_argument("folder", type=Path)
    ask = commands.add_parser("ask", help="ask a question")
    ask.add_argument("question")
    args = parser.parse_args()
 
    with remember.Client.from_env() as client:
        try:
            if args.command == "load":
                load_notes(client, args.folder)
            else:
                print(answer(client, args.question))
        except MemoryApiError as error:
            print(f"memory error: {error.status_code} {error.detail}", file=sys.stderr)
            return 1
        except TimeoutError as error:
            print(f"still processing: {error}", file=sys.stderr)
            return 1
    return 0
 
if __name__ == "__main__":
    sys.exit(main())

python assistant.py load notes/
python assistant.py ask "When does the billing migration go live, and who decided?"

The billing migration now goes live in October 2026 [C1]. Ravi moved it
from June because the invoice exporter needs a rewrite, and Dana agreed
[C2]. The June date from the kickoff no longer holds [C3].
 
Sources:
  [C1] 2026-09-17-standup: "The migration moves from June to October."
  [C2] 2026-09-17-standup: "Dana agreed and will tell finance."
  [C3] 2026-06-02-kickoff: "We go live with the new billing system in June."