KAG multi-agents sur des tables dynamiques de Snowflake avec LangGraph

Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : [2026] La création de systèmes KAG multi-agents avec des tables dynamiques Snowflake, LangGraph et Neo4j. L’accent est mis sur des étapes opérationnelles, des vérifications explicites et du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner l’intention derrière lui.

Le « Fashionista » de l’ingénierie des données

Pour le « Fashionista » de l’ingénierie des données, il convient de définir les entrées, le responsable de chaque étape ainsi que les critères d’achèvement avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans avoir à deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez les terminations partielles silencieuses. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.

Pourquoi c’est important pour les systèmes KAG multi-agents

Pour comprendre l’importance de cela dans les systèmes KAG multi-agents, il faut définir les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts permet d’éviter des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité basée sur des clés métier transforme les fusionnements ultérieurs en opérations d’insertion ou de mise à jour prévisibles, plutôt que en situations de doublons massifs.

Architecture

Pour l’architecture, définissez les entrées, le responsable de chaque étape et les critères de fin avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du système. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.

Prérequis

Pour les prérequis, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie du produit, et non d’une mise en forme ultérieure. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.

# Python packages
pip install langgraph langchain langchain-openai neo4j snowflake-connector-python \
    snowflake-snowpark-python sentence-transformers pydantic

# Neo4j (Docker)
docker run -d --name neo4j-kag \
    -p 7474:7474 -p 7687:7687 \
    -e NEO4J_AUTH=neo4j/password123 \
    -e NEO4J_PLUGINS='["apoc","graph-data-science"]' \
    neo4j:5.15-enterprise

--Snowflake Dynamic Tables Setup
1.1 Create Adaptive Warehouse
-- Adaptive warehouse with high query performance for mixed workloads
-- Reference: https://docs.snowflake.com/en/user-guide/warehouses-adaptive
CREATE ADAPTIVE WAREHOUSE ADAPTIVE_KAG_WH
    WITH MAX_QUERY_PERFORMANCE_LEVEL = XLARGE
    QUERY_THROUGHPUT_MULTIPLIER = 4;

Créer des tables sources

Pour créer des tables sources, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.

CREATE OR REPLACE DATABASE KAG_DB;
CREATE OR REPLACE SCHEMA KAG_DB.PIPELINE;

USE SCHEMA KAG_DB.PIPELINE;
-- Source: Raw documents
CREATE OR REPLACE TABLE DOCUMENTS (
    doc_id VARCHAR(64) DEFAULT UUID_STRING(),
    title VARCHAR(500),
    content TEXT,
    source_type VARCHAR(50),
    created_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
    metadata VARIANT
);
-- Source: Document metadata
CREATE OR REPLACE TABLE METADATA (
    doc_id VARCHAR(64),
    category VARCHAR(100),
    tags ARRAY,
    author VARCHAR(200),
    published_date DATE,
    confidence_score FLOAT
);
-- Source: User feedback for reinforcement
CREATE OR REPLACE TABLE USER_FEEDBACK (
    feedback_id VARCHAR(64) DEFAULT UUID_STRING(),
    query TEXT,
    response TEXT,
    rating INT,
    corrections TEXT,
    timestamp TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

Dynamic Table Pipeline (Incremental Refresh)
-- Layer 1: Raw document normalization
-- Incremental refresh: only processes new/changed rows
CREATE OR REPLACE DYNAMIC TABLE RAW_DOCUMENTS
    TARGET_LAG = '1 minute'
    WAREHOUSE = ADAPTIVE_KAG_WH
    REFRESH_MODE = INCREMENTAL
AS
SELECT
    d.doc_id,
    d.title,
    d.content,
    d.source_type,
    d.created_at,
    m.category,
    m.tags,
    m.author,
    m.confidence_score,
    LENGTH(d.content) AS content_length,
    HASH(d.content) AS content_hash
FROM DOCUMENTS d
LEFT JOIN METADATA m ON d.doc_id = m.doc_id;


-- Layer 2: Entity extraction using Cortex AI
-- Incremental refresh: only extracts from new documents
CREATE OR REPLACE DYNAMIC TABLE EXTRACTED_ENTITIES
    TARGET_LAG = '5 minutes'
    WAREHOUSE = ADAPTIVE_KAG_WH
    REFRESH_MODE = INCREMENTAL
AS
SELECT
    doc_id,
    title,
    content,
    category,
    SNOWFLAKE.CORTEX.AI_EXTRACT(
        content,
        '{
            "entities": [
                {"name": "entity_name", "type": "string", "description": "Name of person, org, concept"},
                {"name": "entity_type", "type": "string", "description": "Type: PERSON, ORG, CONCEPT, LOCATION, EVENT"},
                {"name": "properties", "type": "object", "description": "Key attributes of the entity"}
            ]
        }'
    ) AS extracted_data,
    content_hash,
    created_at
FROM RAW_DOCUMENTS
WHERE content_length > 50;


-- Layer 3: Entity relationships extraction
-- Incremental refresh for relationship triples
CREATE OR REPLACE DYNAMIC TABLE ENTITY_RELATIONS
    TARGET_LAG = '10 minutes'
    WAREHOUSE = ADAPTIVE_KAG_WH
    REFRESH_MODE = INCREMENTAL
AS
SELECT
    doc_id,
    title,
    SNOWFLAKE.CORTEX.AI_EXTRACT(
        content,
        '{
            "relations": [
                {"name": "subject", "type": "string", "description": "Source entity"},
                {"name": "predicate", "type": "string", "description": "Relationship type"},
                {"name": "object", "type": "string", "description": "Target entity"},
                {"name": "confidence", "type": "number", "description": "Confidence 0-1"}
            ]
        }'
    ) AS relation_data,
    category,
    created_at
FROM EXTRACTED_ENTITIES;


-- Layer 4: Embedding generation for vector search
-- DOWNSTREAM lag: refreshes only when upstream changes
CREATE OR REPLACE DYNAMIC TABLE ENTITY_EMBEDDINGS
    TARGET_LAG = DOWNSTREAM
    WAREHOUSE = ADAPTIVE_KAG_WH
    REFRESH_MODE = AUTO
AS
SELECT
    doc_id,
    f.value:entity_name::VARCHAR AS entity_name,
    f.value:entity_type::VARCHAR AS entity_type,
    SNOWFLAKE.CORTEX.EMBED_TEXT_1024('e5-base-v2',
        entity_name || ' ' || COALESCE(f.value:properties::VARCHAR, '')
    ) AS embedding_vector,
    created_at
FROM EXTRACTED_ENTITIES,
    LATERAL FLATTEN(input => extracted_data:entities) f
WHERE f.value:entity_name IS NOT NULL;


-- Layer 5: Knowledge Graph sync queue
-- Fast refresh for near-real-time Neo4j sync
CREATE OR REPLACE DYNAMIC TABLE KG_SYNC_QUEUE
    TARGET_LAG = '1 minute'
    WAREHOUSE = ADAPTIVE_KAG_WH
    REFRESH_MODE = INCREMENTAL
AS
SELECT
    r.doc_id,
    f.value:subject::VARCHAR AS subject,
    f.value:predicate::VARCHAR AS predicate,
    f.value:object::VARCHAR AS object_entity,
    f.value:confidence::FLOAT AS confidence,
    r.category AS domain,
    r.created_at,
    'PENDING' AS sync_status
FROM ENTITY_RELATIONS r,
    LATERAL FLATTEN(input => r.relation_data:relations) f
WHERE f.value:confidence::FLOAT > 0.7;

Mise en place du graphe de connaissances Neo4j

Pour la configuration du graphe de connaissances Neo4j, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute exécution partielle silencieuse. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations upsert prévisibles, plutôt que de créer des doublons en masse.

--Graph Schema (Cypher)
// Constraints and indexes
CREATE CONSTRAINT entity_name IF NOT EXISTS
FOR (e:Entity) REQUIRE e.name IS UNIQUE;

CREATE CONSTRAINT doc_id IF NOT EXISTS
FOR (d:Document) REQUIRE d.doc_id IS UNIQUE;
CREATE INDEX entity_type_idx IF NOT EXISTS
FOR (e:Entity) ON (e.type);
CREATE INDEX entity_domain_idx IF NOT EXISTS
FOR (e:Entity) ON (e.domain);
// Vector index for semantic search
CREATE VECTOR INDEX entity_embedding IF NOT EXISTS
FOR (e:Entity)
ON (e.embedding)
OPTIONS {indexConfig: {
    `vector.dimensions`: 1024,
    `vector.similarity_function`: 'cosine'
}};

Graph Data Model
// Entity node structure
(:Entity {
    name: STRING,
    type: STRING,       // PERSON, ORG, CONCEPT, LOCATION, EVENT
    domain: STRING,     // category from source
    properties: MAP,
    embedding: LIST<FLOAT>,
    created_at: DATETIME,
    source_doc_id: STRING
})

// Document node
(:Document {
    doc_id: STRING,
    title: STRING,
    category: STRING,
    created_at: DATETIME
})

// Relationships
(:Entity)-[:RELATES_TO {predicate: STRING, confidence: FLOAT, doc_id: STRING}]->(:Entity)
(:Entity)-[:MENTIONED_IN]->(:Document)
(:Document)-[:BELONGS_TO]->(:Category)

Load Sample Data
-- Insert sample documents
INSERT INTO KAG_DB.PIPELINE.DOCUMENTS (title, content, source_type) VALUES
('Renewable Energy Report 2024',
 'SolarTech Inc. partnered with GreenGrid Corp to deploy 500MW of solar capacity across Texas. CEO Jane Smith announced the partnership at CleanTech Summit 2024. The project received $2B funding from CleanEnergy Fund.',
 'report'),
('AI in Healthcare',
 'MedAI Corp developed a diagnostic tool using deep learning for early cancer detection. Dr. John Chen leads the research team. The FDA approved the tool in March 2024 after clinical trials at Mayo Clinic.',
 'research'),
('Supply Chain Analysis',
 'GlobalLogistics partnered with TechManufacture for semiconductor distribution. The partnership reduced delivery times by 40%. ChipDesign Corp supplies the core components manufactured in Taiwan.',
 'analysis');

INSERT INTO KAG_DB.PIPELINE.METADATA (doc_id, category, tags, author)
SELECT doc_id, 'energy', ARRAY_CONSTRUCT('solar', 'partnership', 'funding'), 'Analyst Team'
FROM DOCUMENTS WHERE title LIKE '%Renewable%';

Exemples de requêtes

Pour les exemples de requêtes, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché.

# Example 1: Knowledge Graph query
answer = run_kag_query(
    "What companies are connected to renewable energy and what are their relationships?"
)

# Example 2: SQL Analytics query
answer = run_kag_query(
    "How many entities were extracted in the last 24 hours grouped by type?"
)
# Example 3: Hybrid query
answer = run_kag_query(
    "Which organizations have the most connections and what is their average confidence score?"
)

def route_query(state: KAGState) -> KAGState: ...
def kg_agent(state: KAGState) -> KAGState: ...
def sql_agent(state: KAGState) -> KAGState: ...
def validation_agent(state: KAGState) -> KAGState: ...
def synthesis_agent(state: KAGState) -> KAGState: ...

Conclusion

Liste de contrôle opérationnelle