Многопроцессный 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: ...

Заключение

Чек-лист операций