From 93aabf230bebf098ecf414746a7b43ec9638d212 Mon Sep 17 00:00:00 2001 From: Aaron Po Date: Sat, 20 Jun 2026 18:03:39 -0400 Subject: [PATCH] update pipeline docs --- docs/pipeline/ETHICS-AND-KNOWN-ISSUES.md | 2 +- docs/pipeline/README.md | 163 ++++++++-------- docs/pipeline/ROADMAP.md | 175 ++++++++++++++++++ docs/pipeline/diagrams/current/activity.puml | 75 ++++++-- docs/pipeline/diagrams/current/class.puml | 58 ++++-- .../diagrams/current/output/activity.svg | 2 +- .../diagrams/current/output/class.svg | 2 +- tooling/pipeline/runpod/pod-template.yaml | 3 +- 8 files changed, 367 insertions(+), 113 deletions(-) create mode 100644 docs/pipeline/ROADMAP.md diff --git a/docs/pipeline/ETHICS-AND-KNOWN-ISSUES.md b/docs/pipeline/ETHICS-AND-KNOWN-ISSUES.md index 2c1686a..0b5fced 100644 --- a/docs/pipeline/ETHICS-AND-KNOWN-ISSUES.md +++ b/docs/pipeline/ETHICS-AND-KNOWN-ISSUES.md @@ -245,7 +245,7 @@ but is semantically incoherent. This comes from limited training-data coverage rather than prompt engineering. Output sample: -[./out-sample/french-cities.example](out-sample/french-cities.example) +[./french-cities.example](french-cities.example) #### Proposed Mitigations diff --git a/docs/pipeline/README.md b/docs/pipeline/README.md index f08757c..37c9ef8 100644 --- a/docs/pipeline/README.md +++ b/docs/pipeline/README.md @@ -94,24 +94,27 @@ Each run writes a fresh dated SQLite file such as ./biergarten-pipeline \ --model ../models/google_gemma-4-E4B-it-Q6_K.gguf \ --prompt-dir prompts \ + --location-count 25 \ --temperature 1.0 --top-p 0.95 --top-k 64 --n-ctx 8192 --seed -1 ``` #### CLI Flags -| Flag | Purpose | -| --------------- | ---------------------------------------------------------------------------------------------------- | -| `--mocked` | Deterministic mock generator, no model required. | -| `--model, -m` | Path to a GGUF file. Required unless `--mocked` is set. | -| `--prompt-dir` | Directory containing prompt files (e.g. `BREWERY_GENERATION.md`). Required unless `--mocked` is set. | -| `--output, -o` | Directory for generated SQLite artifacts. Default: `output`. | -| `--log-path` | Path for application logs. Default: `pipeline.log`. | -| `--temperature` | Sampling temperature. Default: `1.0`. | -| `--top-p` | Nucleus sampling. Default: `0.95`. | -| `--top-k` | Top-k sampling. Default: `64`. | -| `--n-ctx` | Context window size. Default: `8192`. | -| `--seed` | Random seed. Default: `-1` (random at runtime). | -| `--help, -h` | Print usage and exit. | +| Flag | Purpose | +| ------------------ | ---------------------------------------------------------------------------------------------------- | +| `--mocked` | Deterministic mock generator, no model required. | +| `--model, -m` | Path to a GGUF file. Required unless `--mocked` is set. | +| `--prompt-dir` | Directory containing prompt files (e.g. `BREWERY_GENERATION.md`). Required unless `--mocked` is set. | +| `--output, -o` | Directory for generated SQLite artifacts. Default: `output`. | +| `--log-path` | Path for application logs. Default: `pipeline.log`. | +| `--location-count` | Number of cities to sample from `locations.json` per run. Default: `10`. | +| `--temperature` | Sampling temperature. Default: `1.0`. | +| `--top-p` | Nucleus sampling. Default: `0.95`. | +| `--top-k` | Top-k sampling. Default: `64`. | +| `--n-ctx` | Context window size. Default: `8192`. | +| `--seed` | Random seed. Default: `-1` (random at runtime). | +| `--n-gpu-layers` | Number of model layers to offload to GPU. Default: `0`. | +| `--help, -h` | Print usage and exit. | `--mocked` and `--model` are mutually exclusive. Omitting both exits with an error before the pipeline starts. Sampling flags are ignored when `--mocked` is @@ -130,17 +133,20 @@ NVIDIA GPU. ### How it works -The container uses a two-stage build. The first stage pulls prebuilt -`libllama`, `libggml`, and backend plugin libraries (including `libggml-cuda.so` -and the CPU variant plugins) from `ghcr.io/ggml-org/llama.cpp:full-cuda`. The -second stage copies those libraries into `/usr/local/lib` and runs `ldconfig` so -the dynamic linker and `dlopen` calls from `ggml_backend_load_all()` can resolve -the CUDA backend plugin at runtime. llama.cpp headers are cloned at the matching -tag and installed into `/usr/local/include`. CMake auto-detects both and skips -the FetchContent source build entirely, keeping image build times short. +The container uses a two-stage build. The builder stage installs CMake/Ninja, +clones the matching llama.cpp release tag for its headers only (installed into +`/usr/local/include`), and copies prebuilt shared libraries (`libllama`, +`libggml`, and CUDA/CPU backend plugins) from `ghcr.io/ggml-org/llama.cpp:full-cuda` +into `/usr/local/lib`. With both headers and libraries present, CMake's +system-library detection (see [Build](#build) above) finds them and skips the +FetchContent source build, keeping image build times short. -`GGML_BACKEND_PATH` is set to `/usr/local/lib` so llama.cpp knows where to scan -for backend plugins. +The runtime stage copies the compiled binary, the same prebuilt shared +libraries, and config/prompt assets into a slim CUDA runtime image. It sets +`LD_LIBRARY_PATH=/usr/local/lib:$LD_LIBRARY_PATH` so the dynamic linker +resolves `libllama`/`libggml` at startup, and also co-locates +`libggml-cuda.so` and the CPU backend plugins next to the binary for +`ggml_backend_load_all()`'s `dlopen` scan. ### Build the image @@ -165,44 +171,46 @@ docker build \ Look for `[biergarten] Found system llama.cpp — skipping FetchContent` in the output to confirm the fast path was taken. -### Run in mocked mode +### Run the container -No model or GPU required. Useful for validating the pipeline logic and SQLite -export path. - -```bash -docker run --rm \ - -e BIERGARTEN_MODE=mocked \ - -v "$PWD/output:/workspace/output" \ - -v "$PWD/logs:/workspace/logs" \ - biergarten-pipeline:latest -``` - -### Run in live mode - -Mount your GGUF model before starting. The container validates the model path -before launching the binary. +The container always runs the model-backed path; there is no `--mocked` +container mode (use a native build for that — see [Quick Start](#quick-start)). +The entrypoint, `runpod/start.sh`, downloads the GGUF model automatically if +it is not already present at the configured path. ```bash docker run --rm \ --runtime=nvidia \ - -e BIERGARTEN_MODE=live \ - -e GGML_BACKEND_PATH="/usr/local/lib/libggml-cuda.so" \ -v "$PWD/models:/workspace/models" \ -v "$PWD/output:/workspace/output" \ -v "$PWD/logs:/workspace/logs" \ biergarten-pipeline:latest ``` -The model must be present at `./models/google_gemma-4-E4B-it-Q6_K.gguf` on the -host. See [Model](#model) above for the download command. +By default this downloads `google_gemma-4-E4B-it-Q6_K.gguf` to +`./models/` on first run if it isn't already there. To use a pre-downloaded +model, place it at that path first — see [Model](#model) above. + +#### Environment variables + +| Variable | Purpose | +| ------------------------- | ----------------------------------------------------------------- | +| `BIERGARTEN_MODEL_PATH` | GGUF model path. Default: `/workspace/models/google_gemma-4-E4B-it-Q6_K.gguf`. | +| `BIERGARTEN_OUTPUT_DIR` | SQLite output directory. Default: `/workspace/output`. | +| `BIERGARTEN_LOG_PATH` | Log file path. Default: `/workspace/logs/pipeline.log`. | +| `BIERGARTEN_GL_LAYERS` | GPU layers to offload (`--n-gpu-layers`). Default: `40`. | +| `BIERGARTEN_TEMPERATURE`, `BIERGARTEN_TOP_P`, `BIERGARTEN_TOP_K`, `BIERGARTEN_N_CTX`, `BIERGARTEN_SEED` | Optional sampling overrides, unset by default (binary defaults apply). | +| `BIERGARTEN_EXTRA_ARGS` | Additional raw CLI args appended verbatim. | + +`--prompt-dir` is hardcoded to `/app/prompts` inside the container and is not +configurable via environment variable. ### RunPod deployment Use a GPU pod template. Mount persistent storage for `/workspace/models`, -`/workspace/output`, and `/workspace/logs`. Set `BIERGARTEN_MODE=live` in the -template environment. See `tooling/pipeline/runpod/pod-template.yaml` for a -starter template. +`/workspace/output`, and `/workspace/logs`. See +`tooling/pipeline/runpod/pod-template.yaml` for a starter template — set the +environment variables listed above to match your run. --- @@ -213,8 +221,8 @@ starter template. | Stage | Implementation | | -------- | --------------------------------------------------------------------------------------------------------------------------------------- | | Load | `JsonLoader::LoadLocations()` reads `locations.json` into typed `Location` records. | -| Sample | `BiergartenDataGenerator::QueryCitiesWithCountries()` samples up to 50 locations per run. | -| Enrich | `WikipediaService` fetches city and beer context. Keeps going when a lookup fails. | +| Sample | `BiergartenPipelineOrchestrator::QueryCitiesWithCountries()` samples `--location-count` locations per run (default `10`). | +| Enrich | `WikipediaEnrichmentService` fetches brewing and beer-related context. Keeps going when a lookup fails. `--mocked` runs use `MockEnrichmentService` instead and skip Wikipedia entirely. | | Generate | `MockGenerator` or `LlamaGenerator` produces brewery names and descriptions in English and the local language. | | Store | `SqliteExportService` writes each successful brewery into a fresh dated `.sqlite` database with normalized location and brewery tables. | | Log | `spdlog` writes results and warnings to the console. | @@ -226,11 +234,13 @@ pipeline continues. - `src/main.cc` — argument parsing and Boost.DI composition root. - `JsonLoader` — validates curated location input. -- `WikipediaService` — queries Wikipedia extracts, caches results, returns empty - context on failure. +- `WikipediaEnrichmentService` — queries Wikipedia extracts, caches results, + returns empty context on failure. `MockEnrichmentService` is the no-op + substitute used in `--mocked` runs. - `LlamaGenerator` — formats prompts for Gemma 4, validates JSON output, retries - malformed responses up to three times. If output looks truncated, the retry - raises the token budget before trying again. + malformed responses up to three times with corrective feedback in the + retry prompt. The token budget is fixed across attempts; it is not raised + automatically on truncation. - `MockGenerator` — stable hash-based output so the same city input always produces the same brewery. - `SqliteExportService` — creates a dated SQLite file per run and persists each @@ -240,18 +250,18 @@ pipeline continues. ### Runtime Behaviour -`WikipediaService` queries city, country, and beer-related Wikipedia extracts -using its configured lookup, then caches the first successful response per query -string. The fetched extract text is included in the prompt as context for -generation. +`WikipediaEnrichmentService` fetches two Wikipedia extracts per city: a +generic "brewing" extract and a "beer in `{country}`" extract. It does not +currently query a city- or region-specific page. Each query string is cached +after its first successful (or empty) lookup. `GetLocationContext()` returns an empty string when the web client is unavailable or when lookup/parsing fails. -`LlamaGenerator` validates model output as structured JSON. The retry path -exists as a safety hatch for cases where the reasoning block consumes available -token budget and compresses the JSON output space. All runs to date have -produced valid output on the first pass; the path is kept for resilience. +`LlamaGenerator` validates model output as structured JSON. On validation +failure it retries up to three times, replaying the previous error message in +the next prompt so the model can self-correct. All runs to date have produced +valid output on the first pass; the retry path is kept for resilience. `MockGenerator` uses stable hashes for repeatable output in demos and Storybook runs. @@ -292,7 +302,6 @@ code, latitude, and longitude for each entry. | `local_languages` | Locale-aware copy selection | | `name_en`, `description_en` | Default English display content | | `name_local`, `description_local` | Local-language display content | -| `region_context` | Richer copy for cards and detail pages | --- @@ -368,29 +377,31 @@ Silicon; CUDA or HIP/ROCm is detected on Linux when the toolkit is present. ## Repo Layout -| Path | Purpose | -| ---------------------------- | -------------------------------------------------- | -| `includes/` | Public headers and shared models. | -| `src/` | Implementation files. | -| `locations.json` | Curated city input copied into the build tree. | -| `prompts/` | System prompts used by the model-backed path. | -| `diagrams/` | Architecture and pipeline diagrams. | -| `tooling/pipeline/runpod/` | Dockerfile, launcher, and RunPod pod template. | -| `ETHICS-AND-KNOWN-ISSUES.md` | Ethics, bias, hallucination analysis, mitigations. | +| Path | Purpose | +| -------------------------------------- | -------------------------------------------------- | +| `tooling/pipeline/includes/` | Public headers and shared models. | +| `tooling/pipeline/src/` | Implementation files. | +| `tooling/pipeline/locations.json` | Curated city input copied into the build tree. | +| `tooling/pipeline/prompts/` | System prompts used by the model-backed path. | +| `tooling/pipeline/runpod/` | Dockerfile, launcher, and RunPod pod template. | +| `docs/pipeline/diagrams/` | Architecture and pipeline diagrams. | +| `docs/pipeline/ETHICS-AND-KNOWN-ISSUES.md` | Ethics, bias, hallucination analysis, mitigations. | --- ## Code Tour +Paths below are relative to `tooling/pipeline/`. + - `src/main.cc` — argument parsing and DI composition root. -- `src/biergarten_data_generator/` — orchestration, sampling, logging, and - export. -- `src/services/wikipedia/` — enrichment service and cache. +- `src/biergarten_pipeline_orchestrator/` — orchestration, sampling, logging, + and export. +- `src/services/enrichment/wikipedia/` — enrichment service and cache. - `src/services/sqlite/` — SQLite export implementation. - `src/data_generation/llama/` — local inference, prompt loading, output validation. - `src/data_generation/mock/` — deterministic fallback. -- `tooling/pipeline/runpod/` — container build and runtime launcher. +- `runpod/` — container build and runtime launcher. --- @@ -398,7 +409,9 @@ Silicon; CUDA or HIP/ROCm is detected on Linux when the toolkit is present. The pipeline currently produces city-aware brewery records and dated SQLite exports. The next passes add additional fixture types so the app can exercise -the full brewery domain without live data. +the full brewery domain without live data. For the detailed engineering +breakdown of what's needed to reach the architecture in +[`diagrams/planned/`](./diagrams/planned/), see [ROADMAP.md](./ROADMAP.md). ### Testing — Very High Priority diff --git a/docs/pipeline/ROADMAP.md b/docs/pipeline/ROADMAP.md new file mode 100644 index 0000000..a583aeb --- /dev/null +++ b/docs/pipeline/ROADMAP.md @@ -0,0 +1,175 @@ +# Pipeline Roadmap — Reaching the Planned Architecture + +This is the engineering breakdown for closing the gap between the **current** +pipeline and the **planned** architecture documented in +[`diagrams/planned/class.puml`](./diagrams/planned/class.puml) and +[`diagrams/planned/activity.puml`](./diagrams/planned/activity.puml). Nothing +in `diagrams/planned/` is implemented yet — this file tracks what it would +take to get there. For the current implementation, see +[README.md](./README.md). + +Items are grouped by layer and roughly ordered by dependency: later groups +build on types and services introduced earlier. + +The only concurrency in the current pipeline is the log dispatcher thread +(`main.cc` spawns it to drain a `BoundedChannel` into spdlog for the +whole run). Sampling, enrichment, generation, and export all happen +synchronously on the main thread inside `BiergartenPipelineOrchestrator::Run()` +— §6 below is what introduces the additional worker threads the planned +architecture needs. + +--- + +## 1. Domain Models + +`tooling/pipeline/includes/data_model/generated_models.h` and `models.h`. + +- [ ] Add `Completeness` enum (`Full`, `Partial`, `Absent`) and a + `LocationContext` struct (`text`, `completeness`, `char_count`). + Replace `EnrichedCity::region_context` (currently a plain + `std::string`) with `context : LocationContext`. +- [ ] Add `BeerStyle` (`name`, `description`, `min_abv`, `max_abv`, + `min_ibu`, `max_ibu`). +- [ ] Add `BeerResult`, `CheckinResult`, `RatingResult` result payloads. +- [ ] Add `GenerationMetadata` (`generation_id`, `generated_time`, + `context_provided`, `generated_with`). +- [ ] Add `activity_weight` to `UserResult` (currently just `username`, + `bio`). +- [ ] Extend `GeneratedBrewery` with `brewery_id`, `context_completeness`, + `metadata` (currently just `{location, brewery}`). +- [ ] Add `GeneratedBeer`, `GeneratedUser`, `GeneratedCheckin`, + `GeneratedRating`, `GeneratedFollow` aggregate structs. +- [ ] Add `UserPersona` (`name`, `description`, `style_affinities`). + +## 2. Data Preloading + +`tooling/pipeline/includes/json_handling/`, fixture files. + +- [ ] Extract a `DataPreloader` interface; have `JsonLoader` implement it + instead of exposing only a static `LoadLocations()`. +- [ ] Add `LoadBeerStyles()`. `beer-styles.json` already exists in the repo + and is already copied into the Docker image + (`runpod/Dockerfile`), but no loader reads it yet, and the native + CMake build doesn't copy it into `build/` at all — `CMakeLists.txt`'s + "Runtime Assets" step only copies `locations.json` and `prompts/`. +- [ ] Add `LoadPersonas()` and author `personas.json` (doesn't exist yet). +- [ ] Add `LoadNamesByCountry()` and author `names-by-country.json` (doesn't + exist yet). + +## 3. Policy / Strategy Layer + +Entirely new — no `includes/policy/` (or equivalent) directory exists today. + +- [ ] `ContextStrategy` interface + `BreweryContextStrategy` / + `BeerContextStrategy`. Today `WikipediaEnrichmentService::GetLocationContext` + hardcodes a generic `"brewing"` query and a `"beer in {country}"` + query directly — no per-phase strategy selection. +- [ ] `SamplingStrategy` interface + `UniformSamplingStrategy`, replacing + the inline `std::ranges::sample(...)` call in + `BiergartenPipelineOrchestrator::QueryCitiesWithCountries()`. +- [ ] `BeerSelectionStrategy` interface + `RandomBeerSelectionStrategy`, to + pick styles per brewery from the `BeerStyle` palette (depends on §1 + and §2). +- [ ] `CheckinDistributionStrategy` interface + `JCurveCheckinStrategy` / + `RandomCheckinStrategy` — this is the "Check-In System" item already + called out in README.md's Next Steps, made concrete. +- [ ] `FollowGenerationStrategy` interface + `RandomFollowStrategy` / + `ActivityWeightedFollowStrategy`. + +## 4. Data Generation + +`tooling/pipeline/includes/data_generation/`, `src/data_generation/`. + +- [ ] Extend the `DataGenerator` interface with `GenerateBeer`, + `GenerateCheckin`, `GenerateRating` (today: only `GenerateBrewery` and + `GenerateUser`). +- [ ] Implement `LlamaGenerator::GenerateUser` for real. It currently + returns a hardcoded `{"test_user", "This is a test user profile from + {locale}."}` regardless of input — see the `// TODO` at the top of + `src/data_generation/llama/generate_user.cc`. +- [ ] Implement `MockGenerator::GenerateBeer` / `GenerateCheckin` / + `GenerateRating`. +- [ ] Add `IPromptFormatter::ExpectedArchitecture()` and + `LlamaGenerator::ValidateModelArchitecture()` so loading a GGUF that + doesn't match the configured chat template fails fast instead of + silently producing degraded output. +- [ ] Add prompt template files for beer/checkin/rating generation next to + the existing `prompts/BREWERY_GENERATION.md`. + +## 5. Export Service + +`tooling/pipeline/includes/services/database/`, `src/services/sqlite/`. + +- [ ] Extend `IExportService` with `ProcessBeer`, `ProcessUser`, + `ProcessCheckin`, `ProcessRating`, `ProcessFollow` (today only + `ProcessRecord(const GeneratedBrewery&)` exists). +- [ ] Add `beers`, `users`, `checkins`, `ratings`, `follows` tables to + `SqliteExportService::InitializeSchema()` — see + `kCreateLocationsTableSql` / `kCreateBreweriesTableSql` in + `includes/services/database/sqlite_statement_helpers.h` for the + existing pattern to follow. +- [ ] Add a `brewery_cache_` alongside the existing `location_cache_`, and + move from one open transaction for the whole run to per-phase batched + commits (`BEGIN` / `COMMIT & BEGIN` on a batch-size threshold), as + shown in the planned activity diagram. + +## 6. Concurrency & Orchestration + +`tooling/pipeline/includes/concurrency/`, +`src/biergarten_pipeline_orchestrator/`. + +- [ ] `BoundedChannel` already exists and is production-tested — but + it's only wired to the single log channel today. Stand up the + per-phase producer/consumer channels (`loc_ch`, `exp_ch`, etc.) the + planned activity diagram describes, each with a dedicated LLM worker + thread and SQLite worker thread. +- [ ] Rewrite `BiergartenPipelineOrchestrator::Run()` from one synchronous + pass over `generated_breweries_` into phased methods — + `RunUserPhase` → `RunBreweryAndBeerPhase` (with a `RunBeerPhase` + sub-step once `brewery_pool_` is populated) → `RunCheckinPhase` / + `RunFollowPhase` (these two can run in parallel) → `RunRatingPhase` — + backed by `user_pool_`, `brewery_pool_`, `beer_pool_`, + `checkin_pool_`, `follow_pool_`. +- [ ] Honor the structural-concurrency requirement already called out in a + comment on `BiergartenPipelineOrchestrator::Run()` in + `biergarten_pipeline_orchestrator.h`: once real worker threads exist, + they must be structurally joined (e.g. via `std::jthread`) before + `Run()` returns, so no worker logs to a closed channel during + teardown. + +## 7. Enrichment + +- [ ] Decide whether to restore the city/region-specific Wikipedia query + that's currently commented out in + `src/services/enrichment/wikipedia/get_summary.cc` + (`GetLocationContext`). The `ContextStrategy` work in §3 is a natural + place to reintroduce it via `BreweryContextStrategy::QueriesFor()`. +- [ ] Pre-warm caches at startup (`PreWarmBeerStyleCache`, + `PreWarmLocationCache` in the planned activity diagram) instead of + fetching lazily per record, so the streaming phases never block on a + cold cache mid-run. + +## 8. Fixtures / Build + +- [ ] Author `personas.json` and `names-by-country.json` (see §2). +- [ ] Fix the `beer-styles.json` build-tree gap: add it to the "Runtime + Assets" `configure_file` step in `CMakeLists.txt` so native builds and + the Docker image agree on what's available at runtime. + +--- + +## Suggested build order + +The planned activity diagram's own phase notes ("brewery*pool* is now fully +populated. Phase 1b may begin.", etc.) imply a dependency order. Roughly: + +1. Domain models (§1) and data preloading (§2) — nothing else compiles + without these. +2. Export schema (§5) — so every later generation phase has somewhere to + land its output. +3. Policy/strategy layer (§3) and generator interface (§4). +4. Concurrency/orchestration rewrite (§6), which is the only piece that + actually wires §1–§5 into the phased, parallel pipeline shown in the + planned diagrams. +5. Enrichment cache pre-warming and the city/region query decision (§7) — + useful at any point, but most valuable once phases run concurrently. diff --git a/docs/pipeline/diagrams/current/activity.puml b/docs/pipeline/diagrams/current/activity.puml index 6a4d8c2..e3559d6 100644 --- a/docs/pipeline/diagrams/current/activity.puml +++ b/docs/pipeline/diagrams/current/activity.puml @@ -18,23 +18,40 @@ skinparam ActivityBarColor #628A5B skinparam SwimlaneBorderColor #547461 skinparam SwimlaneBorderThickness 0.3 -title The Biergarten Data Pipeline (Streaming Architecture) +title The Biergarten Data Pipeline (Current — Synchronous Data Path) |#F2F6F0|main.cc| start +:Create BoundedChannel log_channel; +:Spawn log dispatcher thread\n(LogDispatcher::Run()); +note right + The only concurrency in the current pipeline: + a dedicated thread drains log_channel and + forwards entries to spdlog for the entire run. + Joined during shutdown after log_channel closes. +end note :ParseArguments(argc, argv); if (Are arguments valid?) then (no) - :spdlog::error usage info; + :Log error / usage info; stop else (yes) endif :Init OpenSSL global state & LlamaBackendState; :di::make_injector(...); -:injector.create>(); -:BiergartenDataGenerator::Run(); +note right + Binds ILogger, IEnrichmentService, DataGenerator, + IExportService, IPromptFormatter, WebClient based + on --mocked vs. model-backed mode. +end note +:injector.create>(); +:BiergartenPipelineOrchestrator::Run(); +note right + Run() itself is synchronous — sampling, enrichment, + generation, and export all happen on this thread. +end note -|#EAF0E8|BiergartenDataGenerator| +|#EAF0E8|BiergartenPipelineOrchestrator| :Initialize SQLite export; |#E0EAE0|SqliteExportService| @@ -48,23 +65,35 @@ note right Begins Transaction end note -|#EAF0E8|BiergartenDataGenerator| +|#EAF0E8|BiergartenPipelineOrchestrator| :QueryCitiesWithCountries(); |#E2EBDC|JsonLoader| :JsonLoader::LoadLocations("locations.json"); -:std::ranges::sample(all_locations, 50); +:std::ranges::sample(all_locations, --location-count); +note right + --location-count defaults to 10. +end note -|#EAF0E8|BiergartenDataGenerator| +|#EAF0E8|BiergartenPipelineOrchestrator| while (For each sampled Location?) is (Remaining cities) - |#DCE8D8|WikipediaService| + |#DCE8D8|IEnrichmentService| :GetLocationContext(loc); - :FetchExtracts(City, Country, Beer); - |#EAF0E8|BiergartenDataGenerator| - :Store EnrichedCity{Location, region_context}; + note right + WikipediaEnrichmentService fetches a generic + "brewing" extract and a "beer in {country}" extract + (not a city/region-specific page). MockEnrichmentService + (--mocked) returns an empty string instead. + end note + |#EAF0E8|BiergartenPipelineOrchestrator| + if (Lookup failed?) then (yes) + :Log warning, skip city; + else (no) + :Store EnrichedCity{Location, region_context}; + endif endwhile (Done) -|#EAF0E8|BiergartenDataGenerator| +|#EAF0E8|BiergartenPipelineOrchestrator| :GenerateBreweries(enriched_cities); |#E5EDE1|DataGenerator| @@ -73,7 +102,10 @@ while (For each EnrichedCity?) is (Remaining cities) :DeterministicHash & Format; else (LlamaGenerator) :PrepareRegionContext; - :LoadBrewerySystemPrompt("prompts/system.md"); + :prompt_directory_->Load("BREWERY_GENERATION"); + note right + Resolves to BREWERY_GENERATION.md inside --prompt-dir. + end note repeat :Infer(system_prompt, user_prompt, max_tokens, kBreweryJsonGrammar); :ValidateBreweryJson(raw, brewery); @@ -81,11 +113,16 @@ while (For each EnrichedCity?) is (Remaining cities) break else (no) :Attempt++; + :Append validation error to retry prompt; endif repeat while (Attempt < 3?) is (yes) + note right + max_tokens is fixed across retries — + it is not raised on truncation. + end note endif - |#EAF0E8|BiergartenDataGenerator| + |#EAF0E8|BiergartenPipelineOrchestrator| if (Generation successful?) then (yes) |#E0EAE0|SqliteExportService| :ProcessRecord(GeneratedBrewery); @@ -97,8 +134,8 @@ while (For each EnrichedCity?) is (Remaining cities) :Insert Brewery (FK: location_id); if (Exception caught during insert?) then (yes) - |#EAF0E8|BiergartenDataGenerator| - :spdlog::warn "Failed to stream record to SQLite export"; + |#EAF0E8|BiergartenPipelineOrchestrator| + :Log warning "SQLite export failed"; note right Data loss is prevented per-record. The pipeline continues running. @@ -106,7 +143,7 @@ while (For each EnrichedCity?) is (Remaining cities) else (no) endif else (no) - :spdlog::warn "Generation failed, skipping..."; + :Log warning "Generation failed, skipping..."; endif |#E5EDE1|DataGenerator| endwhile (Done) @@ -119,6 +156,8 @@ note right end note |#F2F6F0|main.cc| +:Close log_channel; +:Join log dispatcher thread; :Return 0; stop diff --git a/docs/pipeline/diagrams/current/class.puml b/docs/pipeline/diagrams/current/class.puml index 0bfb7f2..76c3b74 100644 --- a/docs/pipeline/diagrams/current/class.puml +++ b/docs/pipeline/diagrams/current/class.puml @@ -25,11 +25,12 @@ skinparam note { title The Biergarten Data Pipeline - Class Diagram -class BiergartenDataGenerator { +class BiergartenPipelineOrchestrator { - logger_ : std::shared_ptr - context_service_ : std::unique_ptr - generator_ : std::unique_ptr - exporter_ : std::unique_ptr + - application_options_ : ApplicationOptions - generated_breweries_ : std::vector + Run() : bool - QueryCitiesWithCountries() : std::vector @@ -54,21 +55,29 @@ class PipelinePhase <> { Teardown } -struct LogEntry { - + timestamp : std::chrono::system_clock::time_point +struct LogDTO { + + level : LogLevel + + phase : PipelinePhase + + message : std::string +} + +struct LogEntry { + + timestamp : std::chrono::system_clock::time_point + + origin : std::source_location + + thread_id : std::thread::id + level : LogLevel + phase : PipelinePhase + message : std::string - + worker : std::optional } interface ILogger <> { - + Log(entry : const LogEntry&) : void + + Log(payload : LogDTO) : void + - {abstract} DoLog(entry : LogEntry) : void } class LogProducer { - channel_ : BoundedChannel& - + Log(entry : const LogEntry&) : void + - DoLog(entry : LogEntry) : void } class LogDispatcher { @@ -81,7 +90,11 @@ interface IEnrichmentService <> { + GetLocationContext(loc : const Location&) : std::string } -class WikipediaService { +class MockEnrichmentService { + + GetLocationContext(loc : const Location&) : std::string +} + +class WikipediaEnrichmentService { - client_ : std::unique_ptr - extract_cache_ : std::unordered_map + GetLocationContext(loc : const Location&) : std::string @@ -113,13 +126,13 @@ class LlamaGenerator { - model_ : ModelHandle - context_ : ContextHandle - prompt_formatter_ : std::unique_ptr + - prompt_directory_ : std::unique_ptr - rng_ : std::mt19937 + GenerateBrewery(...) : BreweryResult + GenerateUser(...) : UserResult - Load(model_path : const std::string&) : void - Infer(...) : std::string - InferFormatted(...) : std::string - - LoadBrewerySystemPrompt(...) : std::string } interface IPromptFormatter <> { @@ -130,6 +143,16 @@ class Gemma4JinjaPromptFormatter { + Format(system_prompt : std::string_view, user_prompt : std::string_view) : std::string } +interface IPromptDirectory <> { + + Load(key : std::string_view) : std::string +} + +class PromptDirectory { + - prompt_dir_ : std::filesystem::path + - cache_ : std::unordered_map + + Load(key : std::string_view) : std::string +} + class JsonLoader { + {static} LoadLocations(filepath : const std::filesystem::path&) : std::vector } @@ -164,19 +187,22 @@ class SystemDateTimeProvider { } ' Structural Relationships / Dependency Injection -BiergartenDataGenerator *-- ILogger : owns -BiergartenDataGenerator *-- IEnrichmentService : owns -BiergartenDataGenerator *-- DataGenerator : owns -BiergartenDataGenerator *-- IExportService : owns +BiergartenPipelineOrchestrator *-- ILogger : owns +BiergartenPipelineOrchestrator *-- IEnrichmentService : owns +BiergartenPipelineOrchestrator *-- DataGenerator : owns +BiergartenPipelineOrchestrator *-- IExportService : owns LogEntry *-- LogLevel LogEntry *-- PipelinePhase +LogDTO *-- LogLevel +LogDTO *-- PipelinePhase ILogger <|.. LogProducer : implements LogProducer ..> LogEntry : emits LogDispatcher ..> LogEntry : consumes -IEnrichmentService <|.. WikipediaService : implements -WikipediaService *-- WebClient : owns +IEnrichmentService <|.. WikipediaEnrichmentService : implements +IEnrichmentService <|.. MockEnrichmentService : implements +WikipediaEnrichmentService *-- WebClient : owns WebClient <|.. HttpWebClient : implements @@ -184,10 +210,12 @@ DataGenerator <|.. MockGenerator : implements DataGenerator <|.. LlamaGenerator : implements LlamaGenerator *-- IPromptFormatter : uses +LlamaGenerator *-- IPromptDirectory : uses IPromptFormatter <|.. Gemma4JinjaPromptFormatter : implements +IPromptDirectory <|.. PromptDirectory : implements -BiergartenDataGenerator ..> JsonLoader : uses +BiergartenPipelineOrchestrator ..> JsonLoader : uses IExportService <|.. SqliteExportService : implements SqliteExportService *-- IDateTimeProvider : owns diff --git a/docs/pipeline/diagrams/current/output/activity.svg b/docs/pipeline/diagrams/current/output/activity.svg index 9b63f89..f827bab 100644 --- a/docs/pipeline/diagrams/current/output/activity.svg +++ b/docs/pipeline/diagrams/current/output/activity.svg @@ -1 +1 @@ -The Biergarten Data Pipeline (Streaming Architecture)The Biergarten Data Pipeline (Streaming Architecture)ParseArguments(argc, argv)spdlog::error usage infonoAre arguments valid?yesInit CurlGlobalState & LlamaBackendStatedi::make_injector(...)injector.create<std::unique_ptr<BiergartenDataGenerator>>()BiergartenDataGenerator::Run()Return 0Initialize SQLite exportQueryCitiesWithCountries()Store EnrichedCity{Location, region_context}Remaining citiesFor each sampled Location?DoneGenerateBreweries(enriched_cities)Generation successful?yesnoData loss is prevented per-record.The pipeline continues running.spdlog::warn "Failed to stream record to SQLite export"spdlog::warn "Generation failed, skipping..."GetUtcTimestamp() from SystemDateTimeProviderBuilds a fresh biergarten_seed_<UTC datetime>.sqlite filenameAppends a numeric suffix if the timestamp already existsOpens DB ConnectionExecutes Schema DDLBegins TransactionInitialize()ProcessRecord(GeneratedBrewery)Location in cache?yesnoReuse location_idInsert Location & Cache IDInsert Brewery (FK: location_id)yesException caught during insert?noCommits TransactionCloses Database ConnectionFinalize()JsonLoader::LoadLocations("locations.json")std::ranges::sample(all_locations, 50)GetLocationContext(loc)FetchExtracts(City, Country, Beer)Generator ModeMockGeneratorLlamaGeneratorDeterministicHash & FormatPrepareRegionContextLoadBrewerySystemPrompt("prompts/system.md")Infer(system_prompt, user_prompt, max_tokens, kBreweryJsonGrammar)ValidateBreweryJson(raw, brewery)Is JSON Valid?yesnoAttempt++Attempt < 3?yesRemaining citiesFor each EnrichedCity?Donemain.ccBiergartenDataGeneratorSqliteExportServiceJsonLoaderWikipediaServiceDataGenerator \ No newline at end of file +The Biergarten Data Pipeline (Current — Synchronous Data Path)Create BoundedChannel<LogEntry> log_channelThe only concurrency in the current pipeline:a dedicated thread drains log_channel andforwards entries to spdlog for the entire run.Joined during shutdown after log_channel closes.Spawn log dispatcher thread(LogDispatcher::Run())ParseArguments(argc, argv)Log error / usage infonoAre arguments valid?yesInit OpenSSL global state & LlamaBackendStateBinds ILogger, IEnrichmentService, DataGenerator,IExportService, IPromptFormatter, WebClient basedon --mocked vs. model-backed mode.di::make_injector(...)injector.create<std::unique_ptr<BiergartenPipelineOrchestrator>>()Run() itself is synchronous — sampling, enrichment,generation, and export all happen on this thread.BiergartenPipelineOrchestrator::Run()Close log_channelJoin log dispatcher threadReturn 0Initialize SQLite exportQueryCitiesWithCountries()Lookup failed?yesnoLog warning, skip cityStore EnrichedCity{Location, region_context}Remaining citiesFor each sampled Location?DoneGenerateBreweries(enriched_cities)Generation successful?yesnoData loss is prevented per-record.The pipeline continues running.Log warning "SQLite export failed"Log warning "Generation failed, skipping..."GetUtcTimestamp() from SystemDateTimeProviderBuilds a fresh biergarten_seed_<UTC datetime>.sqlite filenameAppends a numeric suffix if the timestamp already existsOpens DB ConnectionExecutes Schema DDLBegins TransactionInitialize()ProcessRecord(GeneratedBrewery)Location in cache?yesnoReuse location_idInsert Location & Cache IDInsert Brewery (FK: location_id)yesException caught during insert?noCommits TransactionCloses Database ConnectionFinalize()JsonLoader::LoadLocations("locations.json")--location-count defaults to 10.std::ranges::sample(all_locations, --location-count)WikipediaEnrichmentService fetches a generic"brewing" extract and a "beer in {country}" extract(not a city/region-specific page). MockEnrichmentService(--mocked) returns an empty string instead.GetLocationContext(loc)Generator ModeMockGeneratorLlamaGeneratorDeterministicHash & FormatPrepareRegionContextResolves to BREWERY_GENERATION.md inside --prompt-dir.prompt_directory_->Load("BREWERY_GENERATION")Infer(system_prompt, user_prompt, max_tokens, kBreweryJsonGrammar)ValidateBreweryJson(raw, brewery)max_tokens is fixed across retries —it is not raised on truncation.Is JSON Valid?yesnoAttempt++Append validation error to retry promptAttempt < 3?yesRemaining citiesFor each EnrichedCity?Donemain.ccBiergartenPipelineOrchestratorSqliteExportServiceJsonLoaderIEnrichmentServiceDataGenerator \ No newline at end of file diff --git a/docs/pipeline/diagrams/current/output/class.svg b/docs/pipeline/diagrams/current/output/class.svg index a1426e2..4f7cb8e 100644 --- a/docs/pipeline/diagrams/current/output/class.svg +++ b/docs/pipeline/diagrams/current/output/class.svg @@ -1 +1 @@ -The Biergarten Data Pipeline - Class DiagramThe Biergarten Data Pipeline - Class DiagramBiergartenDataGeneratorcontext_service_ : std::unique_ptr<IEnrichmentService>generator_ : std::unique_ptr<DataGenerator>exporter_ : std::unique_ptr<IExportService>generated_breweries_ : std::vector<GeneratedBrewery>Run() : boolQueryCitiesWithCountries() : std::vector<Location>GenerateBreweries(cities : std::span<const EnrichedCity>) : voidLogResults() : void«interface»IEnrichmentServiceGetLocationContext(loc : const Location&) : std::stringWikipediaServiceclient_ : std::unique_ptr<WebClient>extract_cache_ : std::unordered_map<std::string, std::string>GetLocationContext(loc : const Location&) : std::stringFetchExtract(query : std::string_view) : std::string«interface»WebClientGet(url : const std::string&) : std::stringUrlEncode(value : const std::string&) : std::stringCURLWebClientGet(url : const std::string&) : std::stringUrlEncode(value : const std::string&) : std::string«interface»DataGeneratorGenerateBrewery(location : const Location&, region_context : const std::string&) : BreweryResultGenerateUser(locale : const std::string&) : UserResultMockGeneratorGenerateBrewery(...) : BreweryResultGenerateUser(...) : UserResultDeterministicHash(location : const Location&) : size_tLlamaGeneratormodel_ : ModelHandlecontext_ : ContextHandleprompt_formatter_ : std::unique_ptr<IPromptFormatter>rng_ : std::mt19937GenerateBrewery(...) : BreweryResultGenerateUser(...) : UserResultLoad(model_path : const std::string&) : voidInfer(...) : std::stringInferFormatted(...) : std::stringLoadBrewerySystemPrompt(...) : std::string«interface»IPromptFormatterFormat(system_prompt : std::string_view, user_prompt : std::string_view) : std::stringGemma4JinjaPromptFormatterFormat(system_prompt : std::string_view, user_prompt : std::string_view) : std::stringJsonLoaderLoadLocations(filepath : const std::filesystem::path&) : std::vector<Location>«interface»IExportServiceInitialize() : voidProcessRecord(brewery : const GeneratedBrewery&) : voidFinalize() : voidSqliteExportServicedate_time_provider_ : std::unique_ptr<IDateTimeProvider>run_timestamp_utc_ : std::stringdatabase_path_ : std::filesystem::pathdb_handle_ : sqlite3*insert_location_stmt_ : sqlite3_stmt*insert_brewery_stmt_ : sqlite3_stmt*transaction_open_ : boollocation_cache_ : std::unordered_map<std::string, sqlite3_int64>Initialize() : voidProcessRecord(brewery : const GeneratedBrewery&) : voidFinalize() : voidInitializeSchema() : void«interface»IDateTimeProviderGetUtcTimestamp() : std::stringSystemDateTimeProviderGetUtcTimestamp() : std::stringownsownsownsimplementsownsimplementsimplementsimplementsusesimplementsusesimplementsownsimplements \ No newline at end of file +The Biergarten Data Pipeline - Class DiagramBiergartenPipelineOrchestratorlogger_ : std::shared_ptr<ILogger>context_service_ : std::unique_ptr<IEnrichmentService>generator_ : std::unique_ptr<DataGenerator>exporter_ : std::unique_ptr<IExportService>application_options_ : ApplicationOptionsgenerated_breweries_ : std::vector<GeneratedBrewery>Run() : boolQueryCitiesWithCountries() : std::vector<Location>GenerateBreweries(cities : std::span<const EnrichedCity>) : voidLogResults() : void«enumeration»LogLevelDebugInfoWarnError«enumeration»PipelinePhaseStartupUserGenerationBreweryAndBeerGenerationCheckinGenerationRatingGenerationFollowGenerationTeardownLogDTOlevel : LogLevelphase : PipelinePhasemessage : std::stringLogEntrytimestamp : std::chrono::system_clock::time_pointorigin : std::source_locationthread_id : std::thread::idlevel : LogLevelphase : PipelinePhasemessage : std::string«interface»ILoggerLog(payload : LogDTO) : voidDoLog(entry : LogEntry) : voidLogProducerchannel_ : BoundedChannel<LogEntry>&DoLog(entry : LogEntry) : voidLogDispatcherchannel_ : BoundedChannel<LogEntry>&Run() : voidToSpdlogLevel(level) : spdlog::level::level_enum«interface»IEnrichmentServiceGetLocationContext(loc : const Location&) : std::stringMockEnrichmentServiceGetLocationContext(loc : const Location&) : std::stringWikipediaEnrichmentServiceclient_ : std::unique_ptr<WebClient>extract_cache_ : std::unordered_map<std::string, std::string>GetLocationContext(loc : const Location&) : std::stringFetchExtract(query : std::string_view) : std::string«interface»WebClientGet(url : const std::string&) : std::stringUrlEncode(value : const std::string&) : std::stringHttpWebClientGet(url : const std::string&) : std::stringUrlEncode(value : const std::string&) : std::string«interface»DataGeneratorGenerateBrewery(location : const Location&, region_context : const std::string&) : BreweryResultGenerateUser(locale : const std::string&) : UserResultMockGeneratorGenerateBrewery(...) : BreweryResultGenerateUser(...) : UserResultDeterministicHash(location : const Location&) : size_tLlamaGeneratormodel_ : ModelHandlecontext_ : ContextHandleprompt_formatter_ : std::unique_ptr<IPromptFormatter>prompt_directory_ : std::unique_ptr<IPromptDirectory>rng_ : std::mt19937GenerateBrewery(...) : BreweryResultGenerateUser(...) : UserResultLoad(model_path : const std::string&) : voidInfer(...) : std::stringInferFormatted(...) : std::string«interface»IPromptFormatterFormat(system_prompt : std::string_view, user_prompt : std::string_view) : std::stringGemma4JinjaPromptFormatterFormat(system_prompt : std::string_view, user_prompt : std::string_view) : std::string«interface»IPromptDirectoryLoad(key : std::string_view) : std::stringPromptDirectoryprompt_dir_ : std::filesystem::pathcache_ : std::unordered_map<std::string, std::string>Load(key : std::string_view) : std::stringJsonLoaderLoadLocations(filepath : const std::filesystem::path&) : std::vector<Location>«interface»IExportServiceInitialize() : voidProcessRecord(brewery : const GeneratedBrewery&) : voidFinalize() : voidSqliteExportServicedate_time_provider_ : std::unique_ptr<IDateTimeProvider>run_timestamp_utc_ : std::stringdatabase_path_ : std::filesystem::pathdb_handle_ : sqlite3*insert_location_stmt_ : sqlite3_stmt*insert_brewery_stmt_ : sqlite3_stmt*transaction_open_ : boollocation_cache_ : std::unordered_map<std::string, sqlite3_int64>Initialize() : voidProcessRecord(brewery : const GeneratedBrewery&) : voidFinalize() : voidInitializeSchema() : void«interface»IDateTimeProviderGetUtcTimestamp() : std::stringSystemDateTimeProviderGetUtcTimestamp() : std::stringownsownsownsownsimplementsemitsconsumesimplementsimplementsownsimplementsimplementsimplementsusesusesimplementsimplementsusesimplementsownsimplements \ No newline at end of file diff --git a/tooling/pipeline/runpod/pod-template.yaml b/tooling/pipeline/runpod/pod-template.yaml index 3b8d5c0..c24d324 100644 --- a/tooling/pipeline/runpod/pod-template.yaml +++ b/tooling/pipeline/runpod/pod-template.yaml @@ -10,11 +10,10 @@ dockerStartCmd: [] isPublic: false isServerless: false env: - BIERGARTEN_MODE: live BIERGARTEN_MODEL_PATH: /workspace/models/google_gemma-4-E4B-it-Q6_K.gguf - BIERGARTEN_PROMPT_DIR: /workspace/app/build/prompts BIERGARTEN_OUTPUT_DIR: /workspace/output BIERGARTEN_LOG_PATH: /workspace/logs/pipeline.log + BIERGARTEN_GL_LAYERS: "40" BIERGARTEN_TEMPERATURE: "1.0" BIERGARTEN_TOP_P: "0.95" BIERGARTEN_TOP_K: "64"