Usługi AI

Platforma DataFlow AI dostarcza dwie usługi Python 3.12 / FastAPI, które wnoszą możliwości dużych modeli językowych do zasobu danych: Copilot do konwersacyjnej i generatywnej asysty oraz Migration Engine do konwersji starszych ETL na potoki DataFlow.

Przegląd

Obie usługi AI znajdują się pod backend/ai-services/, są budowane za pomocą Poetry i działają na FastAPI. Współdzielą instancję PostgreSQL platformy (Copilot używa również jej rozszerzenia pgvector dla RAG).

Definiującą właściwością obu usług jest płynna degradacja — awarie RAG, katalogu i LLM wracają do czatu bez RAG, gotowych schematów lub jawnych znaczników llm_unavailable, zamiast ujawniać fałszywki. Drugą jest wszechobecne ocenianie pewności — każdy wynik AI niesie ze sobą ocenę pewności w zakresie 0–1, a Migration Engine używa jej jako twardej bramki wydania.

Usługa Copilot

Struktura aplikacji

copilot/copilot/main.py buduje aplikację FastAPI DataFlow AI Copilot v0.3.0.

• Potrójne montowanie tras — ten sam zestaw routerów jest montowany pod trzema prefiksami, aby był osiągalny niezależnie od tego, jak adresuje go brama lub frontend:
  • /api/v1/ai/** — kanoniczne montowanie (api/router.py).
  • /api/v1/copilot/** — baseURL=/api/v1 frontendu plus /copilot/chat.
  • /api/copilot/** — dostęp bezpośredni, gdy VITE_API_URL=/api.
• Cykl życia — przy starcie usługa inicjalizuje pulę pgvector RAGService i zapewnia schemat embeddingów, degradując płynnie do rag_mode: "unavailable", jeśli pgvector jest nieosiągalny; pula jest zamykana przy wyłączeniu.
• Endpointy operacyjne — /health, /ready (zwraca 503 z przyczynami pgvector_unavailable / llm_unavailable, gdy zdegradowane) oraz /metrics (minimalna ekspozycja tekstowa Prometheus: informacje o usłudze, czas działania, licznik żądań).
• CORS udostępnia X-Request-Id i X-Copilot-Confidence.

Konfiguracja

Konfiguracja używa pydantic-settings i jest sterowana zmiennymi środowiskowymi (copilot/config.py). Godne uwagi wartości domyślne:

Zależności (pyproject.toml): fastapi, uvicorn, anthropic ^0.39, asyncpg, pgvector, httpx, pyyaml, numpy. sentence-transformers ^3.3 to opcjonalny dodatek (grupa ml) — gdy nie jest zainstalowany, wyszukiwanie wektorowe RAG ulega degradacji.

Fabryka dostawców LLM

Podłączalna warstwa dostawców znajduje się pod copilot/providers/ (base.py, factory.py, trzy implementacje). ABC LLMProvider definiuje complete() i stream(), zwracając znormalizowane CompletionResponse (treść, model, tokeny prompt/completion, latency_ms, finish_reason) oraz delty StreamChunk. LLMProviderError to jednolity typ błędu.

factory.py odczytuje zmienną środowiskową LLM_PROVIDER (anthropic | openrouter | local; domyślnie anthropic), buforuje bezpieczny wątkowo singleton i oferuje reset_provider() do przeładowania. Starszy shim ClaudeClient (clients/claude_client.py) jest utrzymywany dla wstecznej kompatybilności — każda usługa Copilota wywołuje self._claude.generate/stream/complete, ale shim teraz kieruje przez factory.get_provider(). Przełączenie LLM_PROVIDER przerabia więc cały stos Copilota bez zmian po stronie wywołujących. CatalogClient (clients/catalog_client.py) to asynchroniczny klient httpx do katalogu zgodnego z OpenMetadata (search_tables(), get_table(), get_lineage()).

Endpointy API

Wszystkie endpointy są osiągalne pod każdym z /api/v1/ai, /api/v1/copilot oraz /api/copilot.

CopilotService — centralny orkiestrator

services/copilot_service.py to centralny orkiestrator. Jego przepływ chat() to:

1. Routing intencji — wykrywanie słów kluczowych (_SQL_INTENT_KEYWORDS, _ERROR_INTENT_INDICATORS) kieruje pytania SQL do NL-to-SQL, a zgłoszenia błędów do diagnostyki błędów.
2. Wzbogacenie RAG — RAGService.get_context_for_query() działa w trybie best-effort, wracając do czatu bez RAG przy awarii.
3. Buduje wzbogacony prompt systemowy (build_chat_system_prompt) wstrzykujący kontekst potoku plus wyniki RAG.
4. Wywołuje Claude poprzez ClaudeClient.generate().
5. _parse_response() wyodrębnia bloki [ACTION type=... label=... payload={...}] do strukturalnych obiektów Action.
6. Stosuje heurystykę pewności — baza 0,7, +0,15 za RAG, +0,05 za kontekst potoku.

chat_stream() produkuje tokeny poprzez strumieniowanie dostawcy. get_insights() używa wewnątrzprocesowego bufora TTL (300s) kluczowanego na (pipeline_id, limit) — wywołanie LLM jest wolne (~50s), a pulpit odpytuje co 60s, ze statycznym _fallback_insights() przy awarii LLM. Dialekt SQL jest automatycznie wykrywany z tekstu komunikatu (Teradata / Snowflake / Postgres / SAP HANA / MSSQL).

NL-to-SQL

services/nl_to_sql.py dostarcza NLToSQLService — generowanie, wyjaśnianie, optymalizację i walidację SQL w pięciu dialektach (enum SQLDialect: snowflake, teradata, postgresql, sap_hana, mssql). Posiada bogate per-dialektowe bloki podpowiedzi (_DIALECT_HINTS) oraz cztery prompty systemowe. Przepływ pracy:

1. Zbuduj kontekst schematu z RAG search_with_scores() (encje tabel/kolumn).
2. Opcjonalnie zastosuj wzbogacenie warstwą semantyczną (domyślnie włączone).
3. Sformatuj per-dialektowy prompt systemowy i wywołaj Claude (temperatura 0,1).
4. Wyodrębnij SQL z bloku w ogrodzeniu; _extract_table_references() poprzez regex; heurystyka _estimate_confidence(); _detect_warnings() oznacza SELECT *, DELETE/UPDATE bez WHERE, DROP/TRUNCATE oraz nieprawidłowe dla dialektu LIMIT.

Generowanie potoków

services/pipeline_generator.py — PipelineGenerator.generate() pobiera schemat katalogu dla żądanych source_systems, pobiera podobne szablony RAG w trybie best-effort, wywołuje Claude z _GENERATOR_SYSTEM_PROMPT (specyfikacja DSL YAML DataFlow), parsuje YAML, wyodrębnia PipelineNode / PipelineEdge dla grafu frontendu i stosuje heurystykę pewności względem strukturalnej jakości YAML.

RAG i magazyn wektorowy

services/rag_service.py oraz services/embedding_pipeline.py implementują RAG nad podobieństwem kosinusowym pgvector.

• Embeddingi używają sentence-transformers all-MiniLM-L6-v2 (EMBEDDING_DIM = 384), ładowanym leniwie.
• Tabela catalog_embeddings przechowuje entity_type, entity_id, entity_name, workspace_id, content, metadata JSONB oraz kolumnę vector(384), z indeksem kosinusowym ivfflat (lists=100) plus indeksami btree workspace i entity_type.
• search() / search_with_scores() stosują rag_similarity_threshold (0,72) oraz opcjonalne filtry workspace_id / entity_type, sortując BY embedding <=>.
• Gdy pgvector jest niedostępny, usługa degraduje do pustych wyników (mode="unavailable") — nie ma cichego powrotu do wyszukiwania słów kluczowych.
• Operacje CRUD: index_catalog() upsert, delete_by_entity/all/stale, get_stats.

EmbeddingPipeline indeksuje metadane katalogu do magazynu wektorowego. Dostarcza wbudowany korpus 6 fragmentów dokumentów (Pipeline YAML DSL Reference, Connector Configuration Guide, Polkomtel Data Landscape, ETL Best Practices, Common Error Patterns, SQL Dialect Reference) oraz 5 szablonów potoków. run_full_index() pobiera tabele i kolumny z 6 usług Polkomtela (teradata-dwh-mona, sap-hana, snowflake-analytics, databricks-lakehouse, oracle-legacy, mssql-departmental) z rozmiarem partii 64.

Agentowy builder

services/agent_service.py dostarcza AgentService — autonomiczny builder potoków używający Claude tool_use (agentowy poziom 4). Pętla działa do 15 iteracji. Sześć narzędzi jest zdefiniowanych w AGENT_TOOLS:

_ToolBackend wykonuje operacje względem prawdziwych API platformy i katalogu, z fallbackami opartymi na wiedzy domenowej (np. gotowe schematy CDR/billing Polkomtela, gdy katalog jest niedostępny). Stan per-sesja jest przechowywany w słowniku AgentSession w pamięci. Bramka zatwierdzeń wstrzymuje destrukcyjne narzędzia jako PendingAction; approve() wznawia konwersację. Agent może kierować żądania w formacie Anthropic przez /v1/messages OpenRouter, gdy skonfigurowany jest tylko OPENROUTER_API_KEY.

Agent RCA

services/rca_agent.py (FIX-052) dostarcza RcaAgent.analyze() do analizy przyczyn źródłowych incydentów. Wykonuje 4 prawdziwe wywołania narzędzi HTTP — ostatnie alerty i uruchomienia (monitor-service), logi uruchomień oraz oś czasu zmian schematu (metadata-service) — a następnie syntetyzuje poprzez skonfigurowany LLMProvider (temperatura 0,2) w strukturalne RcaResponse: podsumowanie, oś czasu, 2–4 uszeregowane wpisy Hypothesis (każdy z pewnością, dowodami i suggestedAction) oraz citedAssets. Niesie prawdziwe model, tokenCount, latencyMs oraz toolCallCount, używa tolerancyjnego parsowania JSON i nigdy nie wymyśla treści.

Pozostałe usługi Copilota

• Dopasowywacz schematów (services/schema_matcher.py) — SchemaMatcherService.match_schemas() wykonuje dopasowanie kolumn przez LLM dla migracji. Z opcjonalnym wzbogaceniem RAG wywołuje Claude (temperatura 0,2) i parsuje JSON do SchemaMatchResult wpisów ColumnMatch sklasyfikowanych jako EXACT / SEMANTIC / TYPE_COERCE / TRANSFORM_NEEDED, każdy z pewnością, transform_expression i uzasadnieniem. Wyniki są przechowywane w pamięci w _match_store; apply_overrides() obsługuje akcje użytkownika accept/reject/modify. Prompt domeny polskiej telekomunikacji obsługuje polskie nazwy kolumn.
• Generator reguł jakości (services/quality_rule_generator.py) — QualityRuleGenerator.generate_rules() zamienia NL na strukturalne reguły w 10 typach RuleType (NOT_NULL, UNIQUE, RANGE, REGEX, ROW_COUNT, CUSTOM_SQL, FRESHNESS, COMPLETENESS, SCHEMA_MATCH, STATISTICAL). Post-walidacja obniża pewność dla kolumn nieobecnych w kontekście schematu. Jego prompt osadza kontekst Polkomtela (PESEL ^\d{11}$, MSISDN ^48\d{9}$, CDR, przychód-nigdy-ujemny).
• Warstwa semantyczna (services/semantic_layer.py) — SemanticLayerService to warstwa wzbogacania glosariuszem biznesowym przed NL-to-SQL. Jego wbudowany glosariusz Polkomtela zawiera ~11 metryk (wskaźnik churn, ARPU, MoU, zużycie danych, net adds, przychód), ~14 wymiarów i 7 szablonów filtrów. resolve_terms() skanuje pytanie (najdłuższe dopasowanie), enrich_prompt() wstrzykuje formuły i mapowania, a następnie deleguje do NLToSQLService.generate_sql().

Strategia promptów

Katalog copilot/prompts/ koduje wiedzę domenową platformy:

• chat_prompt.py — duży CHAT_SYSTEM_PROMPT kodujący pełny krajobraz danych Polkomtela (DWH-MONA Teradata, SAP HANA, Snowflake, Databricks, Oracle, MSSQL, Kafka, S3), glosariusz telekomunikacyjny (CDR, MSISDN, IMSI, ARPU, churn), DSL YAML potoków, konektory, najlepsze praktyki ETL oraz format bloku ACTION.
• suggest_prompt.py — prompty uzupełnień specyficzne dla języka (konfiguracja YAML/SQL/połączenia) z formatem wyjścia confidence|text.
• insights_prompt.py — 6 kategorii wglądu, wyjście jako tablica JSON.
• error_diagnosis_prompt.py — 8 kategorii błędów ze specyficzną dla Polkomtela wiedzą o błędach (Teradata TPT/PERM, SAP HANA CDC), wyjście jako obiekt JSON.
• schema_match_prompt.py — 4 poziomy reguł dopasowania z pasmami pewności.

Modele danych Copilota

Modele Pydantic pod copilot/models/ obejmują chat.py (Role, Message, PipelineContext, ChatRequest, Action/ActionType, ChatResponse), suggestion.py, insight.py, pipeline.py, sql.py (SQLDialect, SQLRequest/Result/ValidationResult, modele diagnostyki błędów), quality_rule.py, schema_match.py, semantic.py oraz rca.py (RcaRequest/Response, TimelineEvent, Hypothesis, CitedAsset). Wszystkie niosą wartość zmiennoprzecinkową confidence, gdzie ma to zastosowanie.

Migration Engine

Struktura aplikacji

Migration Engine to aplikacja FastAPI DataFlow Migration Engine v0.2.0 na porcie 8091, montowana pod /api/v1/migration oraz /api/migration. Konwertuje starsze skrypty ETL na YAML DataFlow, używając dopasowywania wzorców opartego na regułach plus tłumaczenia wspomaganego LLM. Konfiguracja: anthropic_api_key, anthropic_model = claude-sonnet-4-20250514, max_upload_size_mb = 50, upload_dir = /tmp/migration-uploads.

Co migruje

Enum MigrationType obejmuje cztery starsze narzędzia ETL (nie dialekty SQL):

Przesyłanie obecnie akceptuje tylko .xml, .yxmd i .dtsx; parser DataStage istnieje i jest podpięty do silnika reguł, ale .dsx nie znajduje się w _ALLOWED_EXTENSIONS.

Endpointy API

Potok parsuj → konwertuj → waliduj

Cykl życia MigrationStatus to: uploaded → parsing → parsed → converting → validating → validated → completed | completed_with_warnings | failed. Endpoint /upload uruchamia wszystkie etapy w trybie inline.

   upload file (.xml / .yxmd / .dtsx)
        │
        ▼
┌───────────────────┐   Stage 1 — PARSE
│  parsers/<tool>   │   tool-specific parser → common WorkflowAST
└─────────┬─────────┘   (WorkflowAST → MappingAST → SourceDefinition /
          │              Transformation / TransformationField / Connector)
          ▼
┌───────────────────┐   Stage 2 — CONVERT
│ RuleEngine.convert│   TRANSFORM_RULES registry handles known types
│  + Expression     │   deterministically; ExpressionTranslator turns
│    Translator     │   PowerCenter expression language into SQL
│  → LLMConverter   │   sub-threshold / unknown types fall back to LLM
│  → YamlGenerator  │   assembles the final DataFlow YAML
└─────────┬─────────┘
          ▼
┌───────────────────┐   Stage 3 — VALIDATE
│ PipelineValidator │   6 checks on the generated YAML
└─────────┬─────────┘
          ▼
   release gate: completed only if confidence ≥ 0.85

Etap 1 — Parsowanie. Parser dla każdego narzędzia źródłowego zamienia plik we wspólny WorkflowAST. powercenter_parser.py i alteryx_parser.py używają lxml; ssis_parser.py odczytuje XML DTS:Executable (zadania Data Flow / Control Flow, Connection Managers, Variables, osadzony SQL); datastage_parser.py odczytuje zastrzeżony format tekstowy (bloki BEGIN DSJOB … END DSJOB). Wszystkie udostępniają metodę klasową get_confidence(type) i rozszerzają bazowy parsers/base.py.

Etap 2 — Konwersja. converters/rule_engine.py — RuleEngine.convert() przechodzi przez każdą transformację. Rejestr TRANSFORM_RULES podklas TransformRule deterministycznie obsługuje 13 typów PowerCenter:

Etapy Alteryx (Filter/Formula/Summarize/Join/Sort/Union/AlteryxSelect) i DataStage (PxFilter/PxAggregator/PxJoin/PxLookup/PxSort/PxMerge/PxChangeCapture/PxSCD/CTransformerStage/PxModify/ContainerView) mają własne wbudowane bloki konwersji. _LLM_FALLBACK_THRESHOLD = 0,60 — typy bez reguły lub z pewnością poniżej progu spadają do konwertera LLM.

ExpressionTranslator (converters/expression_translator.py) tłumaczy język wyrażeń PowerCenter na SQL. Rekurencyjnie zstępujący tokenizer/parser mapuje ~60 funkcji poprzez PC_TO_SQL_FUNCTIONS, ze specjalizowanymi handlerami (_fn_iif → CASE WHEN, _fn_decode → CASE, NVL → COALESCE, SUBSTR → SUBSTRING, ADD_TO_DATE → INTERVAL, mapowanie tokenów formatu daty). Obsługuje zmienne $$parameter; nieznane funkcje przechodzą dalej z ostrzeżeniem i spadkiem pewności.

LLMConverter (converters/llm_converter.py) to fallback Claude dla złożonych transformacji. Używa SDK anthropic bezpośrednio (anthropic.AsyncAnthropic) — nie fabryki dostawców Copilota. Trzy specjalizowane prompty systemowe obejmują przypadek ogólny, procedury składowane oraz logikę transformacji niestandardowej/Java. Przy błędzie uwierzytelnienia API, limitu szybkości lub innym zwraca _unavailable_result() z conversion_source="llm_unavailable", pewnością 0,0 oraz requires_manual_review=True — odróżniając autentyczny wynik LLM od fallbacku awarii. Enum ConversionSource to rule_engine | llm | llm_unavailable | unsupported.

YamlGenerator (converters/yaml_generator.py) składa końcowy YAML DataFlow — klasyfikuje węzły na źródła/transformacje/cele, buduje powiązanie depends_on z krawędzi, wstrzykuje 3 domyślne sprawdzenia jakości (row_count, null_percentage, duplicate) plus per-ujściowe sprawdzenia walidacji schematu i sprawdzenie low_confidence_reconciliation dla mapowań poniżej 0,80 oraz dodaje metadane (generated_at, overall_confidence, warnings_count).

Etap 3 — Walidacja. validators/pipeline_validator.py — PipelineValidator.validate() uruchamia 6 sprawdzeń wygenerowanego YAML: składnia YAML, wymagane klucze potoku (name, nodes), schemat węzła (id/type, prawidłowe typy, prawidłowe konektory), odwołania krawędzi rozwiązujące się do id węzłów oraz skan niebezpiecznego SQL (DROP TABLE/DATABASE, TRUNCATE, ALTER TABLE, EXEC, xp_cmdshell). Zwraca ValidationResult z flagą is_valid plus problemami z prefiksem ERROR/WARNING.

Bramkowanie pewności

Bramka wydania (api/upload.py::collect_blocking_conversion_issues) oznacza zadanie jako completed tylko wtedy, gdy spełnione są wszystkie poniższe warunki:

• Ogólna pewność ≥ 0,85.
• Brak obiektów poniżej pewności 0,80.
• Brak obiektów wymagających ręcznego przeglądu.
• Brak problemów walidacji.

W przeciwnym razie zadanie jest oznaczane jako failed.

Usługi i trwałość

• services/migration_service.py — MigrationService opakowuje magazyn zadań. list_jobs() filtruje/sortuje/stronicuje; validate_output() uruchamia bogatszy zestaw 6 sprawdzeń (składnia YAML, bezpieczeństwo SQL, pokrycie transformacji ≥ 50%, pewność ≥ 60%, kompletność węzłów dla źródła+ujścia, problemy z mapowaniem); start_conversion() obsługuje endpoint ponownej konwersji.
• services/report_generator.py — produkuje wskaźniki MappingSummary i EffortEstimate przy użyciu modelu osobogodzin: 0,25h bazowo na mapowanie, 2,0h na ręczny przegląd, 1,0h na konwersję wspomaganą LLM, 0,5h testowania, 4,0h integracji; progi pewności 0,85 / 0,60 / 0,40.

Trwałość zadań to słownik _jobs w pamięci, który jest również utrwalany do jobs-store.json w katalogu przesyłania (persist_job() / save_jobs_store() / _load_jobs_store()) — przetrwa restarty dzięki plikowi JSON. Nie ma prawdziwej bazy danych.

Modele danych migracji

migration/models/migration.py definiuje MigrationType, MigrationStatus, MigrationJob, UploadResponse, MigrationReport (total_objects / auto_converted / needs_review / manual_required), PaginatedJobsResponse, ValidationCheckResult, ValidationResponse oraz ConvertResponse. migration/models/mapping.py definiuje ConversionSource, ObjectMapping, ConversionResult oraz dataclasy AST SourceDefinition, TransformationField, Transformation, Connector, MappingAST i WorkflowAST.

Kluczowe obserwacje

• Dwie odrębne ścieżki dostępu do LLM — Copilot używa podłączalnej fabryki LLMProvider (anthropic/openrouter/local) kierowanej przez shim ClaudeClient, podczas gdy LLMConverter Migration Engine oraz AgentService Copilota wywołują SDK anthropic bezpośrednio.
• Embeddingi używają all-MiniLM-L6-v2 (384-wymiarowe) z indeksem kosinusowym ivfflat pgvector — opcjonalna zależność, której brak powodował błędy 500 w czacie.
• Płynna degradacja wszędzie — awarie RAG, katalogu i LLM wracają do czatu bez RAG, gotowych schematów lub znaczników llm_unavailable; endpoint /ready Copilota uczciwie raportuje degraded.
• Pewność jest wszechobecna — każdy wynik AI (czat, SQL, reguły, dopasowania, konwersje, hipotezy RCA) niesie ocenę pewności w zakresie 0–1; Migration Engine używa jej jako twardej bramki wydania przy ≥ 0,85.
• Silne zakodowanie domeny Polkomtel/telekomunikacja — krajobraz danych, glosariusz CDR/MSISDN/ARPU, regexy PESEL/MSISDN oraz obsługa polskojęzycznych nazw kolumn są wbudowane w prompty i glosariusz semantyczny.