Багатоагентний KAG на динамічних таблицях Snowflake з LangGraph

У цьому посібнику описано процес створення системи від сировини до готового продукту для проекту: [2026] Створення багатоагентних KAG-систем за допомогою динамічних таблиць Snowflake, LangGraph та Neo4j. Основна увага приділяється конкретним крокам виконання, чітким перевіркам та коду, який можна без проблем додати до репозиторію, не здогадуючись про його призначення.

„Фешніст“ у сфері інженерії даних

Для проекту „Фешніст“ у сфері інженерії даних необхідно перед зміною коду визначити вхідні дані, відповідального за крок та критерії завершення. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не здогадуючись про прихований стан системи. Вважайте цей етап контрактом між вхідними даними та перевіреними результатами. Позначте всі елементи проекту, визначте критерії успіху та не допускайте мовчазного часткового завершення роботи. Використовуйте обмеження та індекси перед масовим введенням даних. Унікальність бізнес-ключів дозволяє здійснювати пізнішу об’єднання даних у передбачуваний спосіб, а не створювати купу дублікатів.

Чому це важливо для багатоагентних систем KAG

Щоб зрозуміти, чому це важливо для багатоагентних систем KAG, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Відображення витрат на ранньому етапі запобігає несподіваним рахункам під час переходу від демо-середовищ до спільних. Використовуйте обмеження та індекси перед масовим завантаженням даних. Унікальність за бізнес-ключами перетворює подальшу об’єднання даних на прогнозовані операції додавання чи заміни, а не на ситуації з дублікатами.

Архітектура

Для архітектури необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Використовуйте обмеження та індекси перед масовим введенням даних. Унікальність бізнес-ключів перетворює подальшу об’єднання даних на прогнозовані операції додавання/заміни замість ситуацій з дублікатами.

Попередні вимоги

Щодо передумов, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Необхідно документувати як шлях успішного виконання, так і шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Використовуйте обмеження та індекси перед масовим введенням даних. Унікальність за бізнес-клучами перетворює подальшу об’єднання даних на прогнозовані операції додавання чи заміни, а не на ситуації з дублікатами.

# 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;

Створення таблиць джерела

Для створення вихідних таблиць необхідно спочатку визначити параметри входу, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Краще використовувати невеликі, перевірювані одиниці коду замість об’ємних скриптів. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на складну структуру процесу. Перед масовим ввезенням даних слід використовувати обмеження та індекси. Унікальність бізнес-ключів дозволяє перетворювати подальшу об’єднання даних на прогнозовані операції додавання чи заміни, а не на ситуації з дублікатами.

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;

Налаштування графа знань Neo4j

Для налаштування графа знань Neo4j необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Призначте назви елементів, визначте критерії успіху та не допускайте мовчазного часткового завершення. Використовуйте обмеження та індекси перед масовим введенням даних. Унікальність бізнес-ключів дозволяє перетворювати подальшу об’єднання даних на прогнозовані операції оновлення замість ситуацій з дублікатами.

--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%';

Приклади запитів

Для прикладів запитів необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан.

# 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: ...

Висновки

Чек-лист для експлуатації