Implémentation — Context Nexus

Architecture technique

Cinq composants, un système cohérent.

Stockage documentaire

Les registres sous forme de fichiers versionnés dans Git. La source de vérité. Simple, auditables, diffables, compatibles avec tous les workflows CI/CD existants.

Index vectoriel

Les artefacts Knowledge découpés en chunks, embarqués et indexés pour la récupération sémantique. Le moteur du mode RAG.

Runtime agent

Le moteur d'exécution des agents. Gère la boucle plan–outil–observation, la mémoire de session et les appels aux Skills MCP.

Context Assembler

La couche d'orchestration propre au Context Nexus. Traverse la chaîne d'héritage, valide les Contracts, assemble le task brief et le passe au runtime agent.

Observabilité

La traçabilité des décisions et des sorties des agents. Indispensable pour détecter les dérives de qualité et calibrer les registres.

Stack recommandée — POC

Développement local

Composant	Outil	Registre(s)
Stockage documents	`Git / filesystem`	Knowledge · Intent · Contracts · Operations
Index vectoriel	`ChromaDB (local)`	Knowledge
Runtime agent	`LangChain / LangGraph`	Tous
Context Assembler	`Python function`	Tous
MCP Skills	`mcp Python SDK`	Contracts · Operations
Observability	`Langfuse (local)`	Tous

Stack recommandée — Production

Prêt pour la production

Composant	Outil recommandé	Notes
Stockage documents	`Git + CI/CD pipeline`	Validation et indexation à chaque PR
Index vectoriel	`Weaviate / Pinecone`	Gestion du schéma et des métadonnées
Runtime agent	`LangGraph + Redis`	Mémoire de session distribuée
Context Assembler	`FastAPI service`	Stateless, scalable horizontalement
MCP Skills	`MCP server cluster`	Registry organisationnel partagé
Observability	`Langfuse Cloud`	3 dimensions de traçabilité

Context Assembler

Le composant central.

Le Context Assembler prend une tâche en entrée et produit un task brief structuré en sortie. Ce brief est tout ce dont l'agent a besoin pour travailler.

from dataclasses import dataclass, field
from typing import Any

@dataclass
class TaskBrief:
    task_id: str
    system_prompt: str          # Knowledge conventions + Intent directives + Contracts assertions
    context_injection: str      # Current spec + applicable Decision Directives
    rag_index: str              # Chroma collection name for RAG calls
    mcp_skills: list[str]       # Callable MCP skill names for this task
    metadata: dict[str, Any] = field(default_factory=dict)

class ContextAssembler:
    def __init__(self, nexus_root: str, rag_client, langfuse):
        self.nexus_root = nexus_root
        self.rag = rag_client
        self.lf = langfuse

    def assemble(self, task_id: str, team: str, spec_path: str) -> TaskBrief:
        with self.lf.trace(name="assemble", input={"task_id": task_id}) as trace:
            # 1. Traverse inheritance chain: org → intermediate → team
            contracts = self._load_contracts(team, trace)
            directives = self._load_directives(team, trace)
            conventions = self._load_conventions(team, trace)

            # 2. Build system prompt (always-active, low-volume, high-density)
            system_prompt = "\n\n".join([conventions, directives, contracts])

            # 3. Load spec for context injection
            spec = open(f"{self.nexus_root}/{spec_path}").read()

            # 4. Resolve available MCP skills for this team
            skills = self._resolve_skills(team)

            brief = TaskBrief(
                task_id=task_id,
                system_prompt=system_prompt,
                context_injection=spec,
                rag_index=f"knowledge-{team}",
                mcp_skills=skills,
            )
            trace.update(output={"brief_size": len(system_prompt)})
            return brief

    def _load_contracts(self, team: str, trace) -> str:
        # Traverse Org → Intermediate → Team, block unapproved exceptions
        contracts = []
        for level_contracts in self._traverse_inheritance("contracts", team):
            for artifact in level_contracts:
                if artifact.get("exception-to") and not artifact.get("exception-approved-by"):
                    trace.event(name="blocked_contract", input={"path": artifact["path"]})
                    continue  # block unapproved exception — never silently include
                contracts.append(artifact["body"])
        return "\n\n".join(contracts)

Indexation Knowledge (RAG)

Les artefacts Knowledge sont découpés en chunks par concept atomique, puis embarqués dans un index vectoriel. Voir le registre Knowledge →

from langchain.text_splitter import MarkdownHeaderTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import SentenceTransformerEmbeddings

def index_knowledge(nexus_root: str, team: str) -> Chroma:
    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("##", "section"), ("###", "subsection")]
    )
    docs = []
    for level in ["org", f"intermediate/{team_bu}", f"teams/{team}"]:
        for path in glob(f"{nexus_root}/{level}/knowledge/**/*.md"):
            chunks = splitter.split_text(open(path).read())
            for chunk in chunks:
                chunk.metadata.update({"level": level, "source": path})
            docs.extend(chunks)

    return Chroma.from_documents(
        docs,
        SentenceTransformerEmbeddings(model_name="all-MiniLM-L6-v2"),
        persist_directory=f".chroma/{team}"
    )

Operations → MCP Skills

Les runbooks Operations sont encapsulés comme des outils appelables. Voir le registre Operations →

from mcp.server import MCPServer
from skills.catalog_indexation_lag import catalog_indexation_lag
from skills.performance_test import performance_test
from skills.mutation_score import mutation_score

server = MCPServer(name="context-nexus-skills")

# Operations runbooks — executable, not just readable
server.register_tool(catalog_indexation_lag)

# Contracts verification tools — return pass/fail + metrics
server.register_tool(performance_test)
server.register_tool(mutation_score)

if __name__ == "__main__":
    server.run(host="0.0.0.0", port=8765)

Observabilité

Trois dimensions avec Langfuse.

Décisions d'assemblage

Quels Contracts ont été chargés, quels artefacts exclus (exceptions non approuvées, artefacts stale), quels résultats RAG récupérés. Ces traces permettent de détecter les dérives de contexte avant qu'elles n'affectent les sorties.

Résultats des quality gates

Chaque appel à un Skill de vérification Contracts est tracé avec son résultat pass/fail et les métriques associées. Un taux d'échec en hausse sur un gate précis signale soit une régression dans le code généré, soit un gate mal calibré.

Qualité RAG

Les chunks récupérés pour chaque requête sont tracés avec leur score de similarité. Un score moyen en baisse indique que la Knowledge n'est plus bien alignée avec les requêtes des agents.

from langfuse import Langfuse

lf = Langfuse()

def traced_assembly(task_id: str, assembler: ContextAssembler, **kwargs) -> TaskBrief:
    with lf.trace(name="context-assembly", user_id=kwargs.get("user")) as trace:
        trace.span(name="inheritance-traversal")
        brief = assembler.assemble(task_id=task_id, **kwargs)
        trace.score(name="brief-size", value=len(brief.system_prompt))
        return brief

def traced_rag(query: str, collection: str, k: int = 5):
    with lf.trace(name="rag-retrieval") as trace:
        results = chroma.similarity_search_with_score(query, k=k, collection=collection)
        avg_score = sum(score for _, score in results) / len(results)
        trace.score(name="avg-similarity", value=avg_score)  # alert if declining
        return [doc for doc, _ in results]

Architecture avancée

Architecture multi-agents.

Dans un contexte multi-agents, le Context Assembler devient un agent orchestrateur. Il reçoit une tâche complexe, la décompose, crée un task brief spécialisé pour chaque sous-tâche, et délègue à des agents workers. Chaque worker opère avec son propre contexte isolé, exactement le mode de fonctionnement de Deep Agents.

Bootstrap

Démarrer en 2 heures.

Le minimum viable pour tester le Context Nexus sur un vrai projet.

Structure de fichiers à créer

nexus/
├── org/
│   ├── knowledge/
│   │   └── conventions.md          # register: knowledge, level: org
│   ├── intent/
│   │   └── directives/
│   │       └── security.md         # register: intent, level: org
│   └── contracts/
│       └── security.md             # register: contracts, level: org
└── teams/
    └── my-team/
        ├── knowledge/
        │   └── architecture.md     # register: knowledge, level: team
        ├── intent/
        │   ├── directives/
        │   │   └── api-patterns.md
        │   └── specs/
        │       └── TICKET-42.md    # register: intent, consumption-mode: context-injection
        ├── contracts/
        │   └── quality-gates.md    # register: contracts, level: team
        └── operations/
            └── runbooks/
                └── deploy.md       # register: operations, consumption-mode: skill

Installation

pip install langchain langchain-community chromadb langfuse mcp openai

Lancer Langfuse en local

docker run -d -p 3000:3000 langfuse/langfuse:latest

Premier agent fonctionnel

from context_assembler import ContextAssembler
from langchain_openai import ChatOpenAI
from langfuse import Langfuse

assembler = ContextAssembler(
    nexus_root="./nexus",
    rag_client=chroma_client,
    langfuse=Langfuse(),
)

# Assemble the task brief for ticket TICKET-42
brief = assembler.assemble(
    task_id="TICKET-42",
    team="my-team",
    spec_path="teams/my-team/intent/specs/TICKET-42.md",
)

# Run the agent with its assembled context
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke([
    {"role": "system", "content": brief.system_prompt},
    {"role": "user", "content": brief.context_injection},
])
print(response.content)

L'implémentation ne fait pas le framework. Ce guide donne la plomberie. La valeur vient de la qualité des artefacts que l'équipe alimente dans les registres.

Registres associés

Knowledge Intent Contracts Operations

Guide technique