1 Commits

Author SHA1 Message Date
Aaron Po
93aabf230b update pipeline docs 2026-06-20 22:59:28 -04:00
8 changed files with 367 additions and 113 deletions

View File

@@ -245,7 +245,7 @@ but is semantically incoherent. This comes from limited training-data coverage
rather than prompt engineering. rather than prompt engineering.
Output sample: Output sample:
[./out-sample/french-cities.example](out-sample/french-cities.example) [./french-cities.example](french-cities.example)
#### Proposed Mitigations #### Proposed Mitigations

View File

@@ -94,23 +94,26 @@ Each run writes a fresh dated SQLite file such as
./biergarten-pipeline \ ./biergarten-pipeline \
--model ../models/google_gemma-4-E4B-it-Q6_K.gguf \ --model ../models/google_gemma-4-E4B-it-Q6_K.gguf \
--prompt-dir prompts \ --prompt-dir prompts \
--location-count 25 \
--temperature 1.0 --top-p 0.95 --top-k 64 --n-ctx 8192 --seed -1 --temperature 1.0 --top-p 0.95 --top-k 64 --n-ctx 8192 --seed -1
``` ```
#### CLI Flags #### CLI Flags
| Flag | Purpose | | Flag | Purpose |
| --------------- | ---------------------------------------------------------------------------------------------------- | | ------------------ | ---------------------------------------------------------------------------------------------------- |
| `--mocked` | Deterministic mock generator, no model required. | | `--mocked` | Deterministic mock generator, no model required. |
| `--model, -m` | Path to a GGUF file. Required unless `--mocked` is set. | | `--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. | | `--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`. | | `--output, -o` | Directory for generated SQLite artifacts. Default: `output`. |
| `--log-path` | Path for application logs. Default: `pipeline.log`. | | `--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`. | | `--temperature` | Sampling temperature. Default: `1.0`. |
| `--top-p` | Nucleus sampling. Default: `0.95`. | | `--top-p` | Nucleus sampling. Default: `0.95`. |
| `--top-k` | Top-k sampling. Default: `64`. | | `--top-k` | Top-k sampling. Default: `64`. |
| `--n-ctx` | Context window size. Default: `8192`. | | `--n-ctx` | Context window size. Default: `8192`. |
| `--seed` | Random seed. Default: `-1` (random at runtime). | | `--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. | | `--help, -h` | Print usage and exit. |
`--mocked` and `--model` are mutually exclusive. Omitting both exits with an `--mocked` and `--model` are mutually exclusive. Omitting both exits with an
@@ -130,17 +133,20 @@ NVIDIA GPU.
### How it works ### How it works
The container uses a two-stage build. The first stage pulls prebuilt The container uses a two-stage build. The builder stage installs CMake/Ninja,
`libllama`, `libggml`, and backend plugin libraries (including `libggml-cuda.so` clones the matching llama.cpp release tag for its headers only (installed into
and the CPU variant plugins) from `ghcr.io/ggml-org/llama.cpp:full-cuda`. The `/usr/local/include`), and copies prebuilt shared libraries (`libllama`,
second stage copies those libraries into `/usr/local/lib` and runs `ldconfig` so `libggml`, and CUDA/CPU backend plugins) from `ghcr.io/ggml-org/llama.cpp:full-cuda`
the dynamic linker and `dlopen` calls from `ggml_backend_load_all()` can resolve into `/usr/local/lib`. With both headers and libraries present, CMake's
the CUDA backend plugin at runtime. llama.cpp headers are cloned at the matching system-library detection (see [Build](#build) above) finds them and skips the
tag and installed into `/usr/local/include`. CMake auto-detects both and skips FetchContent source build, keeping image build times short.
the FetchContent source build entirely, keeping image build times short.
`GGML_BACKEND_PATH` is set to `/usr/local/lib` so llama.cpp knows where to scan The runtime stage copies the compiled binary, the same prebuilt shared
for backend plugins. 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 ### Build the image
@@ -165,44 +171,46 @@ docker build \
Look for `[biergarten] Found system llama.cpp — skipping FetchContent` in the Look for `[biergarten] Found system llama.cpp — skipping FetchContent` in the
output to confirm the fast path was taken. 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 The container always runs the model-backed path; there is no `--mocked`
export path. container mode (use a native build for that — see [Quick Start](#quick-start)).
The entrypoint, `runpod/start.sh`, downloads the GGUF model automatically if
```bash it is not already present at the configured path.
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.
```bash ```bash
docker run --rm \ docker run --rm \
--runtime=nvidia \ --runtime=nvidia \
-e BIERGARTEN_MODE=live \
-e GGML_BACKEND_PATH="/usr/local/lib/libggml-cuda.so" \
-v "$PWD/models:/workspace/models" \ -v "$PWD/models:/workspace/models" \
-v "$PWD/output:/workspace/output" \ -v "$PWD/output:/workspace/output" \
-v "$PWD/logs:/workspace/logs" \ -v "$PWD/logs:/workspace/logs" \
biergarten-pipeline:latest biergarten-pipeline:latest
``` ```
The model must be present at `./models/google_gemma-4-E4B-it-Q6_K.gguf` on the By default this downloads `google_gemma-4-E4B-it-Q6_K.gguf` to
host. See [Model](#model) above for the download command. `./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 ### RunPod deployment
Use a GPU pod template. Mount persistent storage for `/workspace/models`, Use a GPU pod template. Mount persistent storage for `/workspace/models`,
`/workspace/output`, and `/workspace/logs`. Set `BIERGARTEN_MODE=live` in the `/workspace/output`, and `/workspace/logs`. See
template environment. See `tooling/pipeline/runpod/pod-template.yaml` for a `tooling/pipeline/runpod/pod-template.yaml` for a starter template — set the
starter template. environment variables listed above to match your run.
--- ---
@@ -213,8 +221,8 @@ starter template.
| Stage | Implementation | | Stage | Implementation |
| -------- | --------------------------------------------------------------------------------------------------------------------------------------- | | -------- | --------------------------------------------------------------------------------------------------------------------------------------- |
| Load | `JsonLoader::LoadLocations()` reads `locations.json` into typed `Location` records. | | Load | `JsonLoader::LoadLocations()` reads `locations.json` into typed `Location` records. |
| Sample | `BiergartenDataGenerator::QueryCitiesWithCountries()` samples up to 50 locations per run. | | Sample | `BiergartenPipelineOrchestrator::QueryCitiesWithCountries()` samples `--location-count` locations per run (default `10`). |
| Enrich | `WikipediaService` fetches city and beer context. Keeps going when a lookup fails. | | 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. | | 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. | | 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. | | 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. - `src/main.cc` — argument parsing and Boost.DI composition root.
- `JsonLoader` — validates curated location input. - `JsonLoader` — validates curated location input.
- `WikipediaService` — queries Wikipedia extracts, caches results, returns empty - `WikipediaEnrichmentService` — queries Wikipedia extracts, caches results,
context on failure. 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 - `LlamaGenerator` — formats prompts for Gemma 4, validates JSON output, retries
malformed responses up to three times. If output looks truncated, the retry malformed responses up to three times with corrective feedback in the
raises the token budget before trying again. 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 - `MockGenerator` — stable hash-based output so the same city input always
produces the same brewery. produces the same brewery.
- `SqliteExportService` — creates a dated SQLite file per run and persists each - `SqliteExportService` — creates a dated SQLite file per run and persists each
@@ -240,18 +250,18 @@ pipeline continues.
### Runtime Behaviour ### Runtime Behaviour
`WikipediaService` queries city, country, and beer-related Wikipedia extracts `WikipediaEnrichmentService` fetches two Wikipedia extracts per city: a
using its configured lookup, then caches the first successful response per query generic "brewing" extract and a "beer in `{country}`" extract. It does not
string. The fetched extract text is included in the prompt as context for currently query a city- or region-specific page. Each query string is cached
generation. after its first successful (or empty) lookup.
`GetLocationContext()` returns an empty string when the web client is `GetLocationContext()` returns an empty string when the web client is
unavailable or when lookup/parsing fails. unavailable or when lookup/parsing fails.
`LlamaGenerator` validates model output as structured JSON. The retry path `LlamaGenerator` validates model output as structured JSON. On validation
exists as a safety hatch for cases where the reasoning block consumes available failure it retries up to three times, replaying the previous error message in
token budget and compresses the JSON output space. All runs to date have the next prompt so the model can self-correct. All runs to date have produced
produced valid output on the first pass; the path is kept for resilience. valid output on the first pass; the retry path is kept for resilience.
`MockGenerator` uses stable hashes for repeatable output in demos and Storybook `MockGenerator` uses stable hashes for repeatable output in demos and Storybook
runs. runs.
@@ -292,7 +302,6 @@ code, latitude, and longitude for each entry.
| `local_languages` | Locale-aware copy selection | | `local_languages` | Locale-aware copy selection |
| `name_en`, `description_en` | Default English display content | | `name_en`, `description_en` | Default English display content |
| `name_local`, `description_local` | Local-language display content | | `name_local`, `description_local` | Local-language display content |
| `region_context` | Richer copy for cards and detail pages |
--- ---
@@ -369,28 +378,30 @@ Silicon; CUDA or HIP/ROCm is detected on Linux when the toolkit is present.
## Repo Layout ## Repo Layout
| Path | Purpose | | Path | Purpose |
| ---------------------------- | -------------------------------------------------- | | -------------------------------------- | -------------------------------------------------- |
| `includes/` | Public headers and shared models. | | `tooling/pipeline/includes/` | Public headers and shared models. |
| `src/` | Implementation files. | | `tooling/pipeline/src/` | Implementation files. |
| `locations.json` | Curated city input copied into the build tree. | | `tooling/pipeline/locations.json` | Curated city input copied into the build tree. |
| `prompts/` | System prompts used by the model-backed path. | | `tooling/pipeline/prompts/` | System prompts used by the model-backed path. |
| `diagrams/` | Architecture and pipeline diagrams. |
| `tooling/pipeline/runpod/` | Dockerfile, launcher, and RunPod pod template. | | `tooling/pipeline/runpod/` | Dockerfile, launcher, and RunPod pod template. |
| `ETHICS-AND-KNOWN-ISSUES.md` | Ethics, bias, hallucination analysis, mitigations. | | `docs/pipeline/diagrams/` | Architecture and pipeline diagrams. |
| `docs/pipeline/ETHICS-AND-KNOWN-ISSUES.md` | Ethics, bias, hallucination analysis, mitigations. |
--- ---
## Code Tour ## Code Tour
Paths below are relative to `tooling/pipeline/`.
- `src/main.cc` — argument parsing and DI composition root. - `src/main.cc` — argument parsing and DI composition root.
- `src/biergarten_data_generator/` — orchestration, sampling, logging, and - `src/biergarten_pipeline_orchestrator/` — orchestration, sampling, logging,
export. and export.
- `src/services/wikipedia/` — enrichment service and cache. - `src/services/enrichment/wikipedia/` — enrichment service and cache.
- `src/services/sqlite/` — SQLite export implementation. - `src/services/sqlite/` — SQLite export implementation.
- `src/data_generation/llama/` — local inference, prompt loading, output - `src/data_generation/llama/` — local inference, prompt loading, output
validation. validation.
- `src/data_generation/mock/` — deterministic fallback. - `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 The pipeline currently produces city-aware brewery records and dated SQLite
exports. The next passes add additional fixture types so the app can exercise 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 ### Testing — Very High Priority

175
docs/pipeline/ROADMAP.md Normal file
View File

@@ -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<LogEntry>` 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<T>` 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.

View File

@@ -18,23 +18,40 @@ skinparam ActivityBarColor #628A5B
skinparam SwimlaneBorderColor #547461 skinparam SwimlaneBorderColor #547461
skinparam SwimlaneBorderThickness 0.3 skinparam SwimlaneBorderThickness 0.3
title The Biergarten Data Pipeline (Streaming Architecture) title The Biergarten Data Pipeline (Current — Synchronous Data Path)
|#F2F6F0|main.cc| |#F2F6F0|main.cc|
start start
:Create BoundedChannel<LogEntry> 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); :ParseArguments(argc, argv);
if (Are arguments valid?) then (no) if (Are arguments valid?) then (no)
:spdlog::error usage info; :Log error / usage info;
stop stop
else (yes) else (yes)
endif endif
:Init OpenSSL global state & LlamaBackendState; :Init OpenSSL global state & LlamaBackendState;
:di::make_injector(...); :di::make_injector(...);
:injector.create<std::unique_ptr<BiergartenDataGenerator>>(); note right
:BiergartenDataGenerator::Run(); Binds ILogger, IEnrichmentService, DataGenerator,
IExportService, IPromptFormatter, WebClient based
on --mocked vs. model-backed mode.
end note
:injector.create<std::unique_ptr<BiergartenPipelineOrchestrator>>();
: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; :Initialize SQLite export;
|#E0EAE0|SqliteExportService| |#E0EAE0|SqliteExportService|
@@ -48,23 +65,35 @@ note right
Begins Transaction Begins Transaction
end note end note
|#EAF0E8|BiergartenDataGenerator| |#EAF0E8|BiergartenPipelineOrchestrator|
:QueryCitiesWithCountries(); :QueryCitiesWithCountries();
|#E2EBDC|JsonLoader| |#E2EBDC|JsonLoader|
:JsonLoader::LoadLocations("locations.json"); :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) while (For each sampled Location?) is (Remaining cities)
|#DCE8D8|WikipediaService| |#DCE8D8|IEnrichmentService|
:GetLocationContext(loc); :GetLocationContext(loc);
:FetchExtracts(City, Country, Beer); note right
|#EAF0E8|BiergartenDataGenerator| 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}; :Store EnrichedCity{Location, region_context};
endif
endwhile (Done) endwhile (Done)
|#EAF0E8|BiergartenDataGenerator| |#EAF0E8|BiergartenPipelineOrchestrator|
:GenerateBreweries(enriched_cities); :GenerateBreweries(enriched_cities);
|#E5EDE1|DataGenerator| |#E5EDE1|DataGenerator|
@@ -73,7 +102,10 @@ while (For each EnrichedCity?) is (Remaining cities)
:DeterministicHash & Format; :DeterministicHash & Format;
else (LlamaGenerator) else (LlamaGenerator)
:PrepareRegionContext; :PrepareRegionContext;
:LoadBrewerySystemPrompt("prompts/system.md"); :prompt_directory_->Load("BREWERY_GENERATION");
note right
Resolves to BREWERY_GENERATION.md inside --prompt-dir.
end note
repeat repeat
:Infer(system_prompt, user_prompt, max_tokens, kBreweryJsonGrammar); :Infer(system_prompt, user_prompt, max_tokens, kBreweryJsonGrammar);
:ValidateBreweryJson(raw, brewery); :ValidateBreweryJson(raw, brewery);
@@ -81,11 +113,16 @@ while (For each EnrichedCity?) is (Remaining cities)
break break
else (no) else (no)
:Attempt++; :Attempt++;
:Append validation error to retry prompt;
endif endif
repeat while (Attempt < 3?) is (yes) repeat while (Attempt < 3?) is (yes)
note right
max_tokens is fixed across retries —
it is not raised on truncation.
end note
endif endif
|#EAF0E8|BiergartenDataGenerator| |#EAF0E8|BiergartenPipelineOrchestrator|
if (Generation successful?) then (yes) if (Generation successful?) then (yes)
|#E0EAE0|SqliteExportService| |#E0EAE0|SqliteExportService|
:ProcessRecord(GeneratedBrewery); :ProcessRecord(GeneratedBrewery);
@@ -97,8 +134,8 @@ while (For each EnrichedCity?) is (Remaining cities)
:Insert Brewery (FK: location_id); :Insert Brewery (FK: location_id);
if (Exception caught during insert?) then (yes) if (Exception caught during insert?) then (yes)
|#EAF0E8|BiergartenDataGenerator| |#EAF0E8|BiergartenPipelineOrchestrator|
:spdlog::warn "Failed to stream record to SQLite export"; :Log warning "SQLite export failed";
note right note right
Data loss is prevented per-record. Data loss is prevented per-record.
The pipeline continues running. The pipeline continues running.
@@ -106,7 +143,7 @@ while (For each EnrichedCity?) is (Remaining cities)
else (no) else (no)
endif endif
else (no) else (no)
:spdlog::warn "Generation failed, skipping..."; :Log warning "Generation failed, skipping...";
endif endif
|#E5EDE1|DataGenerator| |#E5EDE1|DataGenerator|
endwhile (Done) endwhile (Done)
@@ -119,6 +156,8 @@ note right
end note end note
|#F2F6F0|main.cc| |#F2F6F0|main.cc|
:Close log_channel;
:Join log dispatcher thread;
:Return 0; :Return 0;
stop stop

View File

@@ -25,11 +25,12 @@ skinparam note {
title The Biergarten Data Pipeline - Class Diagram title The Biergarten Data Pipeline - Class Diagram
class BiergartenDataGenerator { class BiergartenPipelineOrchestrator {
- logger_ : std::shared_ptr<ILogger> - logger_ : std::shared_ptr<ILogger>
- context_service_ : std::unique_ptr<IEnrichmentService> - context_service_ : std::unique_ptr<IEnrichmentService>
- generator_ : std::unique_ptr<DataGenerator> - generator_ : std::unique_ptr<DataGenerator>
- exporter_ : std::unique_ptr<IExportService> - exporter_ : std::unique_ptr<IExportService>
- application_options_ : ApplicationOptions
- generated_breweries_ : std::vector<GeneratedBrewery> - generated_breweries_ : std::vector<GeneratedBrewery>
+ Run() : bool + Run() : bool
- QueryCitiesWithCountries() : std::vector<Location> - QueryCitiesWithCountries() : std::vector<Location>
@@ -54,21 +55,29 @@ class PipelinePhase <<enumeration>> {
Teardown Teardown
} }
struct LogEntry { struct LogDTO {
+ timestamp : std::chrono::system_clock::time_point + 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 + level : LogLevel
+ phase : PipelinePhase + phase : PipelinePhase
+ message : std::string + message : std::string
+ worker : std::optional<std::string>
} }
interface ILogger <<interface>> { interface ILogger <<interface>> {
+ Log(entry : const LogEntry&) : void + Log(payload : LogDTO) : void
- {abstract} DoLog(entry : LogEntry) : void
} }
class LogProducer { class LogProducer {
- channel_ : BoundedChannel<LogEntry>& - channel_ : BoundedChannel<LogEntry>&
+ Log(entry : const LogEntry&) : void - DoLog(entry : LogEntry) : void
} }
class LogDispatcher { class LogDispatcher {
@@ -81,7 +90,11 @@ interface IEnrichmentService <<interface>> {
+ GetLocationContext(loc : const Location&) : std::string + GetLocationContext(loc : const Location&) : std::string
} }
class WikipediaService { class MockEnrichmentService {
+ GetLocationContext(loc : const Location&) : std::string
}
class WikipediaEnrichmentService {
- client_ : std::unique_ptr<WebClient> - client_ : std::unique_ptr<WebClient>
- extract_cache_ : std::unordered_map<std::string, std::string> - extract_cache_ : std::unordered_map<std::string, std::string>
+ GetLocationContext(loc : const Location&) : std::string + GetLocationContext(loc : const Location&) : std::string
@@ -113,13 +126,13 @@ class LlamaGenerator {
- model_ : ModelHandle - model_ : ModelHandle
- context_ : ContextHandle - context_ : ContextHandle
- prompt_formatter_ : std::unique_ptr<IPromptFormatter> - prompt_formatter_ : std::unique_ptr<IPromptFormatter>
- prompt_directory_ : std::unique_ptr<IPromptDirectory>
- rng_ : std::mt19937 - rng_ : std::mt19937
+ GenerateBrewery(...) : BreweryResult + GenerateBrewery(...) : BreweryResult
+ GenerateUser(...) : UserResult + GenerateUser(...) : UserResult
- Load(model_path : const std::string&) : void - Load(model_path : const std::string&) : void
- Infer(...) : std::string - Infer(...) : std::string
- InferFormatted(...) : std::string - InferFormatted(...) : std::string
- LoadBrewerySystemPrompt(...) : std::string
} }
interface IPromptFormatter <<interface>> { interface IPromptFormatter <<interface>> {
@@ -130,6 +143,16 @@ class Gemma4JinjaPromptFormatter {
+ Format(system_prompt : std::string_view, user_prompt : std::string_view) : std::string + Format(system_prompt : std::string_view, user_prompt : std::string_view) : std::string
} }
interface IPromptDirectory <<interface>> {
+ Load(key : std::string_view) : std::string
}
class PromptDirectory {
- prompt_dir_ : std::filesystem::path
- cache_ : std::unordered_map<std::string, std::string>
+ Load(key : std::string_view) : std::string
}
class JsonLoader { class JsonLoader {
+ {static} LoadLocations(filepath : const std::filesystem::path&) : std::vector<Location> + {static} LoadLocations(filepath : const std::filesystem::path&) : std::vector<Location>
} }
@@ -164,19 +187,22 @@ class SystemDateTimeProvider {
} }
' Structural Relationships / Dependency Injection ' Structural Relationships / Dependency Injection
BiergartenDataGenerator *-- ILogger : owns BiergartenPipelineOrchestrator *-- ILogger : owns
BiergartenDataGenerator *-- IEnrichmentService : owns BiergartenPipelineOrchestrator *-- IEnrichmentService : owns
BiergartenDataGenerator *-- DataGenerator : owns BiergartenPipelineOrchestrator *-- DataGenerator : owns
BiergartenDataGenerator *-- IExportService : owns BiergartenPipelineOrchestrator *-- IExportService : owns
LogEntry *-- LogLevel LogEntry *-- LogLevel
LogEntry *-- PipelinePhase LogEntry *-- PipelinePhase
LogDTO *-- LogLevel
LogDTO *-- PipelinePhase
ILogger <|.. LogProducer : implements ILogger <|.. LogProducer : implements
LogProducer ..> LogEntry : emits LogProducer ..> LogEntry : emits
LogDispatcher ..> LogEntry : consumes LogDispatcher ..> LogEntry : consumes
IEnrichmentService <|.. WikipediaService : implements IEnrichmentService <|.. WikipediaEnrichmentService : implements
WikipediaService *-- WebClient : owns IEnrichmentService <|.. MockEnrichmentService : implements
WikipediaEnrichmentService *-- WebClient : owns
WebClient <|.. HttpWebClient : implements WebClient <|.. HttpWebClient : implements
@@ -184,10 +210,12 @@ DataGenerator <|.. MockGenerator : implements
DataGenerator <|.. LlamaGenerator : implements DataGenerator <|.. LlamaGenerator : implements
LlamaGenerator *-- IPromptFormatter : uses LlamaGenerator *-- IPromptFormatter : uses
LlamaGenerator *-- IPromptDirectory : uses
IPromptFormatter <|.. Gemma4JinjaPromptFormatter : implements IPromptFormatter <|.. Gemma4JinjaPromptFormatter : implements
IPromptDirectory <|.. PromptDirectory : implements
BiergartenDataGenerator ..> JsonLoader : uses BiergartenPipelineOrchestrator ..> JsonLoader : uses
IExportService <|.. SqliteExportService : implements IExportService <|.. SqliteExportService : implements
SqliteExportService *-- IDateTimeProvider : owns SqliteExportService *-- IDateTimeProvider : owns

File diff suppressed because one or more lines are too long

Before

Width:  |  Height:  |  Size: 44 KiB

After

Width:  |  Height:  |  Size: 55 KiB

File diff suppressed because one or more lines are too long

Before

Width:  |  Height:  |  Size: 44 KiB

After

Width:  |  Height:  |  Size: 71 KiB

View File

@@ -10,11 +10,10 @@ dockerStartCmd: []
isPublic: false isPublic: false
isServerless: false isServerless: false
env: env:
BIERGARTEN_MODE: live
BIERGARTEN_MODEL_PATH: /workspace/models/google_gemma-4-E4B-it-Q6_K.gguf 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_OUTPUT_DIR: /workspace/output
BIERGARTEN_LOG_PATH: /workspace/logs/pipeline.log BIERGARTEN_LOG_PATH: /workspace/logs/pipeline.log
BIERGARTEN_GL_LAYERS: "40"
BIERGARTEN_TEMPERATURE: "1.0" BIERGARTEN_TEMPERATURE: "1.0"
BIERGARTEN_TOP_P: "0.95" BIERGARTEN_TOP_P: "0.95"
BIERGARTEN_TOP_K: "64" BIERGARTEN_TOP_K: "64"