Каг з многамасовымі агентамі на дынамічных таблетках Snowflake з LangGraph

У гэтым карыце парадоксальнае шлях ад сыр'ёў да рабочай сістэмы для: [2026] Стварэння мульты-агентных KAG-сістэм з дынамічнымі табламі Snowflake, LangGraph і Neo4j. Акцэнт ставяцца на практычныя крокі, чысткія перакананні і код, які можна проста дадзіць у рэпазітарый без неабяснення меты.

“Фэшаніст” інжынерыі дадзеных

Для “Фэшаніста” інжынерыі дадзеных неабяснены ўвескі, адпаведальны за крок і крытэрыя завершэння раней, чым змяніцца код. Аператары должны магчымае перадзеяць крок з вядомай точкі контролю без неабяснення схованага стану. Спрыятлівайце гэты ўражак як даговор межаў увескаў і падтвердзілых выходаў. Назвайце артыфакты, неабяснены крытэрыі успеху і адмовіцеся ад тыхоўскага частковага завершэння. Іспользуйце абмежэнні і індэксы перад масовым запоўненням. Унікальнасць на бізнес-клучах ператварае пазнейшыя з'еднанні на прыведзенныя upsert-операцыі заместа бурэй дуплікатаў.

Чаму гэта важна для сістэм 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: ...

Вывык

Чэрніця аперацыйных дзеянняў