给你的智能体一份它们无法讨好的记录。

pip install hypgraph            # engine, CLI, JSON API, UI — standard library only
pip install "hypgraph[mcp]"     # + the MCP server (FastMCP)

from hypgraph import Graph, Policy

POLICY = Policy(
    reviewers={"critic", "stats-reviewer"},   # whose verdicts count (an investigation can add its own)
    instrument="runner",                      # the only author whose kind=lab runs are "recorded" evidence
    operator="operator",                      # revises anything, settles disputes
    importer="importer",                      # brings legacy records in, with their original dates
)
g = Graph("research.hypgraph", policy=POLICY)

hypgraph --db research.hypgraph --as critic mcp                                # stdio
hypgraph --db research.hypgraph --as worker-1 mcp --transport http --port 8766 # streamable HTTP

from hypgraph import GraphError

def add_claim(statement: str, scope: str, result: str, measurements: list[dict], why: str) -> str:
    """Add a claim supported by a result; every number in the statement must be bound in measurements."""
    try:
        n = g.create(author=AGENT_NAME, type="claim", inv=INV,
                     fields={"statement": statement, "scope": scope, "measurements": measurements},
                     links=[{"rel": "supports", "from": result, "why": why}])
        return f"Added {n['code']}: {n['state']['sentence']}"
    except GraphError as e:
        return f"Refused — fix this and retry:\n{e}"

from hypgraph import file_sha256

data, code = file_sha256("evals/heldout.jsonl"), file_sha256("eval.py")
a_data = g.create(author="runner", type="artifact", inv=INV, fields={**data, "format": "jsonl"})["id"]
a_code = g.create(author="runner", type="artifact", inv=INV, fields={**code, "format": "py"})["id"]
run = g.create(author="runner", type="run", inv=INV, fields={"kind": "lab", "state": "succeeded", "seed": 7},
               links=[{"rel": "uses", "to": a_data, "data": {"role": "data"}},
                      {"rel": "uses", "to": a_code, "data": {"role": "code"}}])["id"]
res = g.create(author="runner", type="result", inv=INV, links=[{"rel": "produced", "from": run}],
               fields={"summary": "Held-out accuracy, fine-tune vs baseline.",
                       "values": {"finetune": {"k": 1624, "n": 2000}, "baseline": {"k": 1568, "n": 2000}}})["id"]

inv = g.create(author="operator", type="investigation",
               fields={"title": "Issue 412", "objective": "Fix the importer and prove it.", "profile": "software@2"})["id"]
q = g.create(author="operator", type="question", inv=inv, fields={"text": "Does the importer keep quoted commas?"})["id"]

g.stage(q)        # {"stage": "implemented", "index": 2, "of": 6, "complete": False,
                  #  "next": "no live artifact kind=diff (meta: base_sha, head_sha, files) part_of Q1", "gates": [...]}
g.stages(inv)     # one row per live question;  GET /api/graph/stages?inv=…  ·  MCP tool: stage
g.completion(inv) # the profile's criteria for the whole investigation

def spend(inv, tasks, since=None):            # your meter: model calls per task id
    rows = ledger.calls(task_ids=tasks, since=since)
    return {"usd": sum(r.usd for r in rows), "calls": len(rows), "input_tokens": ..., "cached_tokens": ...,
            "output_tokens": ..., "by_task": [{"task_id": t, "usd": ...} for t in tasks]}

g = Graph("research.hypgraph", policy=POLICY, spend=spend)
with g.writing_for(task_id):                    # events carry the task
    g.create(...)
g.cost(inv)        # $ for the investigation, per node, per verdict, and since the record last moved

hypgraph serve --db research.hypgraph            # HypGraph UI + JSON API on 127.0.0.1:8765, read-only
hypgraph serve --db research.hypgraph --writable # + POST /api/graph/write

from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from hypgraph.server import dispatch
from hypgraph.ui import static_dir

app = FastAPI()
app.mount("/hypgraph", StaticFiles(directory=static_dir(), html=True))

@app.get("/api/graph/{path:path}")
def graph_api(path: str, request: Request):
    status, body = dispatch(g, "GET", "/" + path, dict(request.query_params))
    return body          # (set the status code too)

hypgraph export -o record.bundle.json     # every event as the exact string that was hashed
hypgraph verify record.bundle.json        # exit 1 if anything was edited

hypgraph check summary.md              # errors fail
hypgraph check --strict summary.md     # warnings (unsourced numbers …) fail too

hypgraph refusals                      # per author and rule, fixed vs open
hypgraph refusals --author analyst     # one agent's refusals in full