Donnez à vos agents un registre qu'ils ne peuvent pas flatter.

Installation

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

Python 3.10+. Un fichier SQLite par registre (Graph("research.hypgraph")), ou Graph() en mémoire pour les tests et les notebooks.

Le modèle

Des nœuds typés — question, hypothèse, expérience, exécution, résultat, affirmation, source, relecture, décision — reliés par des liens typés dont les extrémités sont imposées (supports, challenges, requires, tests, produced, replicates, based_on …). Toutes les tables sont en ajout seul ; chaque écriture est un événement dans une chaîne SHA-256. Aucun statut n'est stocké. Chaque lecture dérive :

• relecture — vérifié / non vérifié / contesté / périmé, à partir des verdicts des relecteurs sur la version actuelle ;
• risque — prémisses tombées en amont (AT RISK), décisions dont la base a bougé (REVISIT), tout le soutien tombé (UNSUPPORTED) ;
• lignes indépendantes — des soutiens qui ne partagent ni exécution, ni artefact de données, ni source ;
• évidence — la ligne la plus forte : rapportée par un agent → enregistrée par l'instrument → reproduite → implémentation indépendante → nouvelles données → source externe.

Les règles complètes sont dans le format du registre ; hypgraph schema --json affiche le vocabulaire.

Politique et rôles

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)

Un modèle qui fonctionne pour une équipe d'agents :

Brancher des agents

MCP

Un processus serveur par identité d'agent — l'auteur est fixé au démarrage :

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

Ou construisez-le dans le processus (politique propre, votre propre authentification) : hypgraph.mcp.build(graph, author="worker-1") renvoie le serveur FastMCP.

Des exemples exécutables, la même histoire intégrée de six façons (LangChain, LangGraph, PydanticAI, CrewAI, OpenAI Agents SDK, Claude Agent SDK) : examples/frameworks. Chacun tourne hors ligne avec un modèle scripté, ou avec --real sur un vrai modèle.

Agent Skill

hypgraph skill --install écrit .claude/skills/hypgraph/SKILL.md : la méthode, les refus et leurs corrections, via la CLI.

Vos propres outils

Tout framework qui accepte des fonctions Python comme outils (Claude Agent SDK, LangChain, LangGraph, OpenAI Agents …) — renvoyez le texte du refus pour que le modèle puisse agir dessus :

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}"

Mettez g.report(inv)["markdown"] ou le texte MCP overview dans le contexte de l'agent au début d'une tâche, pour qu'il lise ce qui est établi avant de le redériver.

Enregistrer des exécutions

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"]

Les artefacts sont adressés par contenu (mêmes octets, un nœud), donc deux résultats calculés à partir des mêmes données fusionnent automatiquement en une ligne d'évidence. Une réexécution lie replicates à data.differs (new_seed, new_impl, new_data …) et élève le profil d'évidence.

Profils et workflows

Fixez un profil sur l'enquête et le registre applique les règles de cette discipline — et, avec les profils @2, son workflow : des étapes avec seuils, ce qui invalide l'évidence, les formes typées que doivent avoir un diff, une instance, une source ou une position. L'étape est dérivée à chaque lecture, comme le statut.

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

• Les seuils sont des prédicats. hypothesis_stated, preregistered, diff_registered, tests_pass_current, backtested, positioned, still_valid, solved, result_reviewed … — utilisables aussi comme done_when d'un ordre.
• Ce qui invalide. En plus des prémisses, un workflow ajoute uses (un fichier de données corrigé, un diff ou une instance remplacés) ou cites (une source retirée) à ce qui rend MENACÉS les dépendants — par profil, donc les registres sous @1 gardent exactement le statut qu'ils avaient.
• Fraîcheur. Un résultat dont l'exécution a utilisé un artefact désormais remplacé, ou dont le valid_until est passé, est PÉRIMÉ ; une affirmation est périmée quand tous les résultats qui la soutiennent le sont. Une vérification sur une évidence périmée ne satisfait pas reviewed.
• Formes typées. artifact.kind + meta (un diff exige base_sha, head_sha, files), source.kind, decision.kind=position avec size, stop, review_date, paper — refusés s'ils sont incomplets. Réenregistrer les mêmes octets type un artefact non typé et ajoute les liens indiqués.
• Des fins honnêtes. Une réponse inconclusive vérifiée, une décision no_position, un échec vérifié (un résultat négatif enregistré par l'instrument, vérifié) complètent chacun une question que les seuils seuls ne complèteraient jamais.

La méthode dont sort chaque workflow, et la décision derrière chaque seuil, ont chacune leur page dans docs/workflows. Le planificateur de HypSwarm émet l'ordre d'étape de chaque question exactement à partir de cela.

Recette : des rapports d'évaluation qui tiennent

1. Préenregistrez. Une expérience avec des liens tests vers chaque hypothèse (« le fine-tuning est meilleur », « pas de différence »), chacune avec data.expected, avant l'exécution.
2. Enregistrez l'exécution comme ci-dessus ; mettez les décomptes dans les valeurs du résultat sous la forme {k, n}, pas seulement des pourcentages.
3. Affirmez avec des mesures. « 1 624 sur 2 000 contre 1 568 » est lié ; « 5 points de mieux » sans champ de résultat pour le 5 est refusé.
4. « Pas de différence » exige de la puissance. Plantez un effet connu (injectez une différence de 2 points dans une copie des prédictions), montrez que la comparaison le détecte avec cette taille d'échantillon et marquez ce résultat power: "demonstrated". Sans cela, enregistrez l'affirmation comme inconclusive — le registre y tient.
5. Relecture. Un agent critique relance l'évaluateur sur les prédictions sauvegardées ; son pass ou fail est épinglé à cette version de l'affirmation.
6. Rapport. g.report(inv)["markdown"] est la table des faits pour le début du rapport ; il ne peut pas réénoncer une affirmation retirée.

Coût

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

HTTP et l'interface

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

Dans votre propre application (FastAPI montré ; tout framework fonctionne) :

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)

L'interface n'utilise ni framework ni étape de compilation : HypGraph.mount(el, {source: HypGraph.HttpSource('/api/graph')}), ou HypGraph.BundleSource(bundle) sans aucun serveur. Elle suit GET /api/graph/stream en direct, et la télémétrie d'une exécution HypSwarm (<record>.telemetry.jsonl) ajoute chaque appel au modèle et aux outils. hypgraph export --telemetry la transporte dans un bundle pour la rejouer, hors de la chaîne de hachage ; {play: true} ouvre en relecture. Thémez-la avec les variables CSS --hg-*. Voyez-la dans les exemples en direct.

Bundles en CI

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

Publiez le hash head du bundle quelque part que vous ne contrôlez pas (une étiquette, un article, un ticket) quand un résultat est publié : un registre ultérieur qui passe par ce hash n'a rien pu réécrire d'antérieur. Les relecteurs peuvent vérifier un bundle sur /verify sans rien installer.

Vérifier le rapport

L'essentiel du hype vit dans le rapport final, pas dans le registre. hypgraph check lit du markdown contre le registre et sort avec 1 en cas d'erreur, de sorte que le résumé que remet un agent ne peut pas contredire ce qui a réellement été démontré :

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

Avertissements : des nombres qu'aucun résultat ne contient ni n'implique, des litiges omis, des citations remplacées, un langage causal sans citations. Les blocs de code, le code en ligne, les URL, les passages cités, les négations (« ne peut être démontré ») et les instructions (« n'utilisez jamais le jeu d'évaluation ») ne sont pas traités comme des affirmations — calibré sur deux douzaines de vrais rapports vérifiés contre le registre vivant dont ils rendent compte. En Python : g.check_text(text, inv) ; via HTTP : POST /api/graph/check ; pour les agents : l'outil MCP check_report.

Le journal des refus

Chaque affirmation refusée est enregistrée — auteur, énoncé, codes — et marquée corrigée quand le même auteur revient avec la version honnête. C'est le chiffre que personne d'autre ne peut vous montrer : à quelle fréquence vos agents ont tenté de surinterpréter, quelles règles les ont arrêtés et lesquels ont abandonné en silence.

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

En ajout seul, avec sa propre chaîne de hachage (g.verify_refusals()), séparée de celle du registre. En Python : g.refusal_report(inv) ; HTTP GET /api/graph/refusals ; MCP refusal_log ; et un décompte des surinterprétations refusées dans le Résumé de HypGraph.

Performances

Python pur, mesuré avec python dev/bench.py --nodes N sur un portable Windows (Python 3.13, hypgraph 0.6). Son registre est une enquête de questions, de résultats liés à des valeurs, d'affirmations dérivées de ceux-ci et de relectures sur certaines ; la dérivation croît au même rythme que le registre. Une écriture coûte ce qu'elle touche : les instantanés partagent les données non touchées, et le statut en cache survit pour chaque nœud que l'écriture ne peut pas atteindre (vérifié contre une reconstruction complète avec une batterie de tests aléatoires). L'essentiel d'une écriture est SQLite validant sur disque. Lancez le benchmark sur votre machine, à vos tailles.