AaronPo/the-biergarten-app
1
0
Fork 0
mirror of https://github.com/aaronpo97/the-biergarten-app.git synced 2026-07-16 17:47:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

format

Browse Source
This commit is contained in:
Aaron Po
2026-06-21 13:13:02 -04:00
parent 4de0ea6638
commit ad97b0ea6c
27 changed files with 313 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
tooling/pipeline/includes/biergarten_pipeline_orchestrator.h
View File

@@ -17,7 +17,6 @@
#include "data_model/generated_models.h" #include "data_model/generated_models.h"
#include "services/database/export_service.h" #include "services/database/export_service.h"
#include "services/enrichment/enrichment_service.h" #include "services/enrichment/enrichment_service.h"
#include "services/logging/logger.h" #include "services/logging/logger.h"
/** /**
@@ -28,14 +27,15 @@
*/ */
class BiergartenPipelineOrchestrator { class BiergartenPipelineOrchestrator {
public: public:
/** /**
* @brief Constructs the orchestrator with injected pipeline dependencies. * @brief Constructs the orchestrator with injected pipeline dependencies.
* *
* @param context_service Provides regional context for locations. * @param context_service Provides regional context for locations.
* @param generator Implementation (Llama or Mock) for brewery/user generation. * @param generator Implementation (Llama or Mock) for brewery/user
* @param exporter Database backend for persisting generated records. * generation.
* @param application_options CLI configuration and paths. * @param exporter Database backend for persisting generated records.
*/ * @param application_options CLI configuration and paths.
*/
BiergartenPipelineOrchestrator( BiergartenPipelineOrchestrator(
std::shared_ptr<ILogger> logger, std::shared_ptr<ILogger> logger,
std::unique_ptr<IEnrichmentService> context_service, std::unique_ptr<IEnrichmentService> context_service,
@@ -52,10 +52,11 @@ class BiergartenPipelineOrchestrator {
* 3. Generate brewery data for sampled cities * 3. Generate brewery data for sampled cities
* *
* @note STRUCTURAL CONCURRENCY REQUIREMENT: * @note STRUCTURAL CONCURRENCY REQUIREMENT:
* When transitioned to a multithreaded design, this method MUST structurally * When transitioned to a multithreaded design, this method MUST
* enforce that all deployed worker threads are joined before returning (e.g. * structurally enforce that all deployed worker threads are joined before
* by using std::jthread or a structured concurrency primitive). This ensures * returning (e.g. by using std::jthread or a structured concurrency
* workers do not attempt to log to a closed channel during application teardown. * primitive). This ensures workers do not attempt to log to a closed
* channel during application teardown.
* *
* @return true if successful, false if not * @return true if successful, false if not
*/ */
@@ -71,11 +72,11 @@ class BiergartenPipelineOrchestrator {
/// @brief Generator dependency selected in the composition root. /// @brief Generator dependency selected in the composition root.
std::unique_ptr<DataGenerator> generator_; std::unique_ptr<DataGenerator> generator_;
/// @brief Storage backend for generated brewery records. /// @brief Storage backend for generated brewery records.
std::unique_ptr<IExportService> exporter_; std::unique_ptr<IExportService> exporter_;
/// @brief CLI configuration: paths, model settings, generation parameters. /// @brief CLI configuration: paths, model settings, generation parameters.
ApplicationOptions application_options_; ApplicationOptions application_options_;
/** /**
* @brief Load locations from JSON and sample cities. * @brief Load locations from JSON and sample cities.
@@ -95,9 +96,9 @@ class BiergartenPipelineOrchestrator {
* @brief Generate users grounded in sampled names and personas for * @brief Generate users grounded in sampled names and personas for
* enriched cities. * enriched cities.
* *
* Loads personas.json / forenames-by-country.json / surnames-by-country.json * Loads personas.json / forenames-by-country.json /
* itself, mirroring how QueryCitiesWithCountries() owns its own * surnames-by-country.json itself, mirroring how QueryCitiesWithCountries()
* locations.json load. * owns its own locations.json load.
* *
* @param cities Span of enriched city data. * @param cities Span of enriched city data.
*/ */

6
tooling/pipeline/includes/concurrency/bounded_channel.h
View File

@@ -9,10 +9,12 @@
/** /**
* @file bounded_channel.h * @file bounded_channel.h
* @brief Thread-safe, bounded multi-producer/multi-consumer synchronous channel. * @brief Thread-safe, bounded multi-producer/multi-consumer synchronous
* channel.
* *
* Intent: Enables asynchronous inter-thread communication with backpressure. * Intent: Enables asynchronous inter-thread communication with backpressure.
* Models a synchronous channel where producers/consumers block on capacity limits. * Models a synchronous channel where producers/consumers block on capacity
* limits.
*/ */
/** /**

6
tooling/pipeline/includes/data_generation/mock_generator.h
View File

@@ -57,8 +57,7 @@ class MockGenerator final : public DataGenerator {
* @return Deterministic hash value. * @return Deterministic hash value.
*/ */
static size_t DeterministicHash(const Location& location, static size_t DeterministicHash(const Location& location,
const UserPersona& persona, const UserPersona& persona, const Name& name);
const Name& name);
// Hash stride constants for deterministic distribution across fixed-size // Hash stride constants for deterministic distribution across fixed-size
// arrays. These coprime strides spread hash values uniformly without // arrays. These coprime strides spread hash values uniformly without
@@ -142,7 +141,8 @@ class MockGenerator final : public DataGenerator {
"Visits breweries for the stories, stays for the flagship pours.", "Visits breweries for the stories, stays for the flagship pours.",
"Craft beer fan mapping tasting notes and favorite brew routes.", "Craft beer fan mapping tasting notes and favorite brew routes.",
"Always ready to trade recommendations for underrated local breweries.", "Always ready to trade recommendations for underrated local breweries.",
"Keeping a running list of must-try collab releases and tap takeovers."}; "Keeping a running list of must-try collab releases and tap "
"takeovers."};
}; };
#endif // BIERGARTEN_PIPELINE_INCLUDES_DATA_GENERATION_MOCK_GENERATOR_H_ #endif // BIERGARTEN_PIPELINE_INCLUDES_DATA_GENERATION_MOCK_GENERATOR_H_

4
tooling/pipeline/includes/data_model/generated_models.h
View File

@@ -3,8 +3,8 @@
/** /**
* @file data_model/generated_models.h * @file data_model/generated_models.h
* @brief Generated output models from the pipeline: brewery/user results, enriched data, * @brief Generated output models from the pipeline: brewery/user results,
* and complete generation results. * enriched data, and complete generation results.
*/ */
#include <string> #include <string>

6
tooling/pipeline/includes/data_model/models.h
View File

@@ -148,8 +148,6 @@ struct GeneratorOptions {
/// @brief Use mocked generator instead of actual LLM inference. /// @brief Use mocked generator instead of actual LLM inference.
bool use_mocked = false; bool use_mocked = false;
/// @brief Specific sampling parameters for this generator. /// @brief Specific sampling parameters for this generator.
/// If nullopt, the application should use global defaults. /// If nullopt, the application should use global defaults.
std::optional<SamplingOptions> sampling; std::optional<SamplingOptions> sampling;
@@ -186,7 +184,7 @@ struct ApplicationOptions {
// Function Declarations // Function Declarations
// ============================================================================ // ============================================================================
std::optional<ApplicationOptions> ParseArguments(const int argc, char** argv, std::optional<ApplicationOptions> ParseArguments(
std::shared_ptr<ILogger> logger = nullptr); const int argc, char** argv, std::shared_ptr<ILogger> logger = nullptr);
#endif // BIERGARTEN_PIPELINE_INCLUDES_DATA_MODEL_MODELS_H_ #endif // BIERGARTEN_PIPELINE_INCLUDES_DATA_MODEL_MODELS_H_

9
tooling/pipeline/includes/data_model/names_by_country.h
View File

@@ -25,11 +25,10 @@
*/ */
class NamesByCountry { class NamesByCountry {
public: public:
NamesByCountry( NamesByCountry(std::unordered_map<std::string, std::vector<ForenameEntry>>
std::unordered_map<std::string, std::vector<ForenameEntry>> forenames_by_country,
forenames_by_country, std::unordered_map<std::string, std::vector<std::string>>
std::unordered_map<std::string, std::vector<std::string>> surnames_by_country);
surnames_by_country);
/** /**
* @brief Samples a first/last name pair for the given country. * @brief Samples a first/last name pair for the given country.

2
tooling/pipeline/includes/services/database/sqlite_export_service.h
View File

@@ -11,8 +11,8 @@
#include <string> #include <string>
#include <unordered_map> #include <unordered_map>
#include "data_model/models.h"
#include "../datetime/date_time_provider.h" #include "../datetime/date_time_provider.h"
#include "data_model/models.h"
#include "export_service.h" #include "export_service.h"
#include "sqlite_export_service_helpers.h" #include "sqlite_export_service_helpers.h"

11
tooling/pipeline/includes/services/logging/log_entry.h
View File

@@ -34,9 +34,9 @@ enum class LogLevel {
* pipeline that emitted it. * pipeline that emitted it.
*/ */
enum class PipelinePhase { enum class PipelinePhase {
Startup, ///< Initialization and validation. Startup, ///< Initialization and validation.
Enrichment, ///< Location/context enrichment (e.g. Wikipedia lookups). Enrichment, ///< Location/context enrichment (e.g. Wikipedia lookups).
UserGeneration, ///< User profile generation. UserGeneration, ///< User profile generation.
BreweryAndBeerGeneration, ///< Brewery and beer data generation. BreweryAndBeerGeneration, ///< Brewery and beer data generation.
CheckinGeneration, ///< Checkin (visit) record generation. CheckinGeneration, ///< Checkin (visit) record generation.
RatingGeneration, ///< Rating and review generation. RatingGeneration, ///< Rating and review generation.
@@ -46,7 +46,8 @@ enum class PipelinePhase {
/** /**
* @struct LogDTO * @struct LogDTO
* @brief User-provided subset of log fields. Used to capture call-site info transparently. * @brief User-provided subset of log fields. Used to capture call-site info
* transparently.
*/ */
struct LogDTO { struct LogDTO {
LogLevel level; LogLevel level;
@@ -74,7 +75,6 @@ struct LogEntry {
/// @brief Thread responsible for emitting the log. /// @brief Thread responsible for emitting the log.
std::thread::id thread_id{}; std::thread::id thread_id{};
/// @brief Severity level of this entry. /// @brief Severity level of this entry.
LogLevel level; LogLevel level;
@@ -83,7 +83,6 @@ struct LogEntry {
/// @brief Log message text. /// @brief Log message text.
std::string message; std::string message;
}; };
#endif // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_ #endif // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_

3
tooling/pipeline/includes/services/logging/logger.h
View File

@@ -39,7 +39,8 @@ class ILogger {
*/ */
void Log(LogDTO payload, void Log(LogDTO payload,
std::source_location origin = std::source_location::current(), std::source_location origin = std::source_location::current(),
std::chrono::system_clock::time_point timestamp = std::chrono::system_clock::now(), std::chrono::system_clock::time_point timestamp =
std::chrono::system_clock::now(),
std::thread::id thread_id = std::this_thread::get_id()) { std::thread::id thread_id = std::this_thread::get_id()) {
LogEntry entry; LogEntry entry;
entry.timestamp = timestamp; entry.timestamp = timestamp;

19
tooling/pipeline/includes/web_client/http_web_client.h
View File

@@ -1,19 +1,18 @@
/** /**
* @file web_client/http_web_client.h * @file web_client/http_web_client.h
* @brief cpp-httplib implementation of the WebClient interface. * @brief cpp-httplib implementation of the WebClient interface.
*/ */
#ifndef BIERGARTEN_PIPELINE_INCLUDES_WEB_CLIENT_HTTP_WEB_CLIENT_H_ #ifndef BIERGARTEN_PIPELINE_INCLUDES_WEB_CLIENT_HTTP_WEB_CLIENT_H_
#define BIERGARTEN_PIPELINE_INCLUDES_WEB_CLIENT_HTTP_WEB_CLIENT_H_ #define BIERGARTEN_PIPELINE_INCLUDES_WEB_CLIENT_HTTP_WEB_CLIENT_H_
#include "web_client/web_client.h"
#include "services/logging/logger.h"
#include <memory> #include <memory>
#include <string> #include <string>
#include <utility> #include <utility>
#include "services/logging/logger.h"
#include "web_client/web_client.h"
/** /**
* @brief WebClient implementation backed by cpp-httplib. * @brief WebClient implementation backed by cpp-httplib.
* *
@@ -26,7 +25,7 @@
* bound to a single origin at construction time. * bound to a single origin at construction time.
*/ */
class HttpWebClient final : public WebClient { class HttpWebClient final : public WebClient {
public: public:
explicit HttpWebClient(std::shared_ptr<ILogger> logger) explicit HttpWebClient(std::shared_ptr<ILogger> logger)
: logger_(std::move(logger)) {} : logger_(std::move(logger)) {}
~HttpWebClient() override = default; ~HttpWebClient() override = default;
@@ -34,7 +33,8 @@ public:
/** /**
* @brief Executes a blocking HTTP/HTTPS GET request against a full URL. * @brief Executes a blocking HTTP/HTTPS GET request against a full URL.
* *
* @param url Fully-qualified URL, e.g. "https://en.wikipedia.org/api/rest_v1/page/summary/Berlin" * @param url Fully-qualified URL, e.g.
* "https://en.wikipedia.org/api/rest_v1/page/summary/Berlin"
* @return Response body on HTTP 2xx; throws std::runtime_error otherwise. * @return Response body on HTTP 2xx; throws std::runtime_error otherwise.
*/ */
std::string Get(const std::string& url) override; std::string Get(const std::string& url) override;
@@ -52,5 +52,4 @@ public:
std::shared_ptr<ILogger> logger_; std::shared_ptr<ILogger> logger_;
}; };
#endif #endif

47
tooling/pipeline/src/application_options/parse_arguments.cc
View File

@@ -51,7 +51,8 @@ std::optional<ApplicationOptions> ParseArguments(
prog_opts::value<std::string>()->default_value("pipeline.log"), prog_opts::value<std::string>()->default_value("pipeline.log"),
"Path for application logs"); "Path for application logs");
opt("prompt-dir", prog_opts::value<std::string>()->default_value(""), opt("prompt-dir", prog_opts::value<std::string>()->default_value(""),
"Directory containing named prompt files (e.g. BREWERY_GENERATION.md)." "Directory containing named prompt files (e.g. "
"BREWERY_GENERATION.md)."
" Required when not using --mocked."); " Required when not using --mocked.");
opt("location-count", prog_opts::value<uint32_t>()->default_value(10)); opt("location-count", prog_opts::value<uint32_t>()->default_value(10));
}; };
@@ -70,11 +71,11 @@ std::optional<ApplicationOptions> ParseArguments(
})(); })();
if (logger) { if (logger) {
logger->Log(LogDTO{.level = LogLevel::Info, logger->Log(LogDTO{.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = title}); .message = title});
logger->Log(LogDTO{.level = LogLevel::Info, logger->Log(LogDTO{.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = usage}); .message = usage});
} }
return std::nullopt; return std::nullopt;
} }
@@ -89,8 +90,8 @@ std::optional<ApplicationOptions> ParseArguments(
help_stream << "\n" << desc; help_stream << "\n" << desc;
if (logger) { if (logger) {
logger->Log(LogDTO{.level = LogLevel::Info, logger->Log(LogDTO{.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = help_stream.str()}); .message = help_stream.str()});
} }
return std::nullopt; return std::nullopt;
} }
@@ -106,14 +107,16 @@ std::optional<ApplicationOptions> ParseArguments(
const std::string model_path = var_map["model"].as<std::string>(); const std::string model_path = var_map["model"].as<std::string>();
const int n_gpu_layers = var_map["n-gpu-layers"].as<int>(); const int n_gpu_layers = var_map["n-gpu-layers"].as<int>();
// Enforce mutual exclusivity before any further configuration is applied. // Enforce mutual exclusivity before any further configuration is
// applied.
if (use_mocked && !model_path.empty()) { if (use_mocked && !model_path.empty()) {
const std::string msg = const std::string msg =
"Invalid arguments: --mocked and --model are mutually exclusive"; "Invalid arguments: --mocked and --model are mutually "
"exclusive";
if (logger) { if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error, logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = msg}); .message = msg});
} else { } else {
std::cerr << msg << std::endl; std::cerr << msg << std::endl;
} }
@@ -122,11 +125,12 @@ std::optional<ApplicationOptions> ParseArguments(
if (!use_mocked && model_path.empty()) { if (!use_mocked && model_path.empty()) {
const std::string msg = const std::string msg =
"Invalid arguments: either --mocked or --model must be specified"; "Invalid arguments: either --mocked or --model must be "
"specified";
if (logger) { if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error, logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = msg}); .message = msg});
} else { } else {
std::cerr << msg << std::endl; std::cerr << msg << std::endl;
} }
@@ -137,7 +141,8 @@ std::optional<ApplicationOptions> ParseArguments(
// generator has no use for it and should not require it to be present. // generator has no use for it and should not require it to be present.
if (!use_mocked && options.pipeline.prompt_dir.empty()) { if (!use_mocked && options.pipeline.prompt_dir.empty()) {
const std::string msg = const std::string msg =
"Invalid arguments: --prompt-dir is required when not using --mocked"; "Invalid arguments: --prompt-dir is required when not using "
"--mocked";
if (logger) { if (logger) {
logger->Log({.level = LogLevel::Error, logger->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
@@ -168,8 +173,8 @@ std::optional<ApplicationOptions> ParseArguments(
"Sampling parameters are ignored when using --mocked"; "Sampling parameters are ignored when using --mocked";
if (logger) { if (logger) {
logger->Log(LogDTO{.level = LogLevel::Warn, logger->Log(LogDTO{.level = LogLevel::Warn,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = msg}); .message = msg});
} else { } else {
std::cerr << msg << std::endl; std::cerr << msg << std::endl;
} }
@@ -194,8 +199,8 @@ std::optional<ApplicationOptions> ParseArguments(
exception.what(); exception.what();
if (logger) { if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error, logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = msg}); .message = msg});
} }
return std::nullopt; return std::nullopt;
} catch (...) { } catch (...) {
@@ -203,8 +208,8 @@ std::optional<ApplicationOptions> ParseArguments(
"Failed to parse command-line arguments: unknown error"; "Failed to parse command-line arguments: unknown error";
if (logger) { if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error, logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = msg}); .message = msg});
} }
return std::nullopt; return std::nullopt;
} }

2
tooling/pipeline/src/biergarten_pipeline_orchestrator/biergarten_pipeline_orchestrator.cc
View File

@@ -12,7 +12,7 @@ BiergartenPipelineOrchestrator::BiergartenPipelineOrchestrator(
std::unique_ptr<IEnrichmentService> context_service, std::unique_ptr<IEnrichmentService> context_service,
std::unique_ptr<DataGenerator> generator, std::unique_ptr<DataGenerator> generator,
std::unique_ptr<IExportService> exporter, std::unique_ptr<IExportService> exporter,
const ApplicationOptions &app_options) const ApplicationOptions& app_options)
: logger_(std::move(logger)), : logger_(std::move(logger)),
context_service_(std::move(context_service)), context_service_(std::move(context_service)),
generator_(std::move(generator)), generator_(std::move(generator)),

37
tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_breweries.cc
View File

@@ -36,33 +36,38 @@ void BiergartenPipelineOrchestrator::GenerateBreweries(
logger_->Log( logger_->Log(
{.level = LogLevel::Warn, {.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration, .phase = PipelinePhase::BreweryAndBeerGeneration,
.message = .message = std::format(
std::format("[Pipeline] Generated brewery for '{}' ({}) but SQLite export failed: {}", "[Pipeline] Generated brewery for '{}' ({}) "
"but SQLite export failed: {}",
location.city, location.country, export_exception.what())}); location.city, location.country, export_exception.what())});
} }
} catch (const std::exception& e) { } catch (const std::exception& e) {
++skipped_count; ++skipped_count;
logger_->Log({.level = LogLevel::Warn, logger_->Log(
.phase = PipelinePhase::BreweryAndBeerGeneration, {.level = LogLevel::Warn,
.message = std::format("[Pipeline] Skipping city '{}' ({}): brewery generation failed: {}", .phase = PipelinePhase::BreweryAndBeerGeneration,
location.city, location.country, e.what())}); .message = std::format("[Pipeline] Skipping city '{}' ({}): brewery "
"generation failed: {}",
location.city, location.country, e.what())});
} }
} }
if (skipped_count > 0) { if (skipped_count > 0) {
logger_->Log({.level = LogLevel::Warn, logger_->Log(
.phase = PipelinePhase::BreweryAndBeerGeneration, {.level = LogLevel::Warn,
.message = std::format( .phase = PipelinePhase::BreweryAndBeerGeneration,
"[Pipeline] Skipped {} city/cities due to generation errors", .message = std::format(
skipped_count)}); "[Pipeline] Skipped {} city/cities due to generation errors",
skipped_count)});
} }
if (export_failed_count > 0) { if (export_failed_count > 0) {
logger_->Log({.level = LogLevel::Warn, logger_->Log(
.phase = PipelinePhase::Teardown, {.level = LogLevel::Warn,
.message = std::format( .phase = PipelinePhase::Teardown,
"[Pipeline] Failed to export {} generated brewery/breweries to SQLite", .message = std::format("[Pipeline] Failed to export {} generated "
export_failed_count)}); "brewery/breweries to SQLite",
export_failed_count)});
} }
} }

63
tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc
View File

@@ -24,8 +24,8 @@ std::string Sanitize(std::string_view value) {
out.reserve(value.size()); out.reserve(value.size());
for (const char character : value) { for (const char character : value) {
if (std::isalnum(static_cast<unsigned char>(character)) != 0) { if (std::isalnum(static_cast<unsigned char>(character)) != 0) {
out.push_back( out.push_back(static_cast<char>(
static_cast<char>(std::tolower(static_cast<unsigned char>(character)))); std::tolower(static_cast<unsigned char>(character))));
} }
} }
return out; return out;
@@ -62,8 +62,7 @@ std::string GenerateDateOfBirth(std::mt19937& rng) {
sys_days{birth_year_anchor} - days{day_offset_dist(rng)}; sys_days{birth_year_anchor} - days{day_offset_dist(rng)};
const year_month_day birth_ymd{birth_date}; const year_month_day birth_ymd{birth_date};
return std::format("{:04}-{:02}-{:02}", return std::format("{:04}-{:02}-{:02}", static_cast<int>(birth_ymd.year()),
static_cast<int>(birth_ymd.year()),
static_cast<unsigned>(birth_ymd.month()), static_cast<unsigned>(birth_ymd.month()),
static_cast<unsigned>(birth_ymd.day())); static_cast<unsigned>(birth_ymd.day()));
} }
@@ -71,7 +70,8 @@ std::string GenerateDateOfBirth(std::mt19937& rng) {
std::string GenerateRandomPassword(std::mt19937& rng) { std::string GenerateRandomPassword(std::mt19937& rng) {
constexpr size_t kPasswordLength = 32; constexpr size_t kPasswordLength = 32;
constexpr std::string_view kCharset = constexpr std::string_view kCharset =
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!@#$%^&*"; "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!@#$%^&"
"*";
std::uniform_int_distribution<size_t> char_dist(0, kCharset.size() - 1); std::uniform_int_distribution<size_t> char_dist(0, kCharset.size() - 1);
@@ -115,14 +115,14 @@ void BiergartenPipelineOrchestrator::GenerateUsers(
if (!sampled_name.has_value()) { if (!sampled_name.has_value()) {
++skipped_count; ++skipped_count;
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::UserGeneration,
.phase = PipelinePhase::UserGeneration, .message = std::format(
.message = std::format( "[Pipeline] Skipping city '{}' ({}): no names "
"[Pipeline] Skipping city '{}' ({}): no names available for " "available for "
"country '{}'", "country '{}'",
city.location.city, city.location.country, city.location.city, city.location.country,
city.location.iso3166_1)}); city.location.iso3166_1)});
continue; continue;
} }
@@ -147,22 +147,22 @@ void BiergartenPipelineOrchestrator::GenerateUsers(
} catch (const std::exception& export_exception) { } catch (const std::exception& export_exception) {
++export_failed_count; ++export_failed_count;
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::UserGeneration,
.phase = PipelinePhase::UserGeneration, .message = std::format(
.message = std::format( "[Pipeline] Generated user for '{}' ({}) but "
"[Pipeline] Generated user for '{}' ({}) but SQLite export failed: {}", "SQLite export failed: {}",
city.location.city, city.location.country, city.location.city, city.location.country,
export_exception.what())}); export_exception.what())});
} }
} catch (const std::exception& e) { } catch (const std::exception& e) {
++skipped_count; ++skipped_count;
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::UserGeneration,
.phase = PipelinePhase::UserGeneration, .message = std::format(
.message = std::format( "[Pipeline] Skipping city '{}' ({}): "
"[Pipeline] Skipping city '{}' ({}): user generation failed: {}", "user generation failed: {}",
city.location.city, city.location.country, e.what())}); city.location.city, city.location.country, e.what())});
} }
} }
@@ -176,11 +176,10 @@ void BiergartenPipelineOrchestrator::GenerateUsers(
} }
if (export_failed_count > 0) { if (export_failed_count > 0) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Teardown,
.phase = PipelinePhase::Teardown, .message = std::format("[Pipeline] Failed to export {} "
.message = std::format( "generated user/users to SQLite",
"[Pipeline] Failed to export {} generated user/users to SQLite", export_failed_count)});
export_failed_count)});
} }
} }

1
tooling/pipeline/src/biergarten_pipeline_orchestrator/log_results.cc
View File

@@ -38,7 +38,6 @@ void BiergartenPipelineOrchestrator::LogResults() const {
boost::json::array user_output; boost::json::array user_output;
for (const auto& generated_user : generated_users_) { for (const auto& generated_user : generated_users_) {
user_output.push_back(boost::json::object{ user_output.push_back(boost::json::object{
{"first_name", generated_user.user.first_name}, {"first_name", generated_user.user.first_name},

32
tooling/pipeline/src/biergarten_pipeline_orchestrator/run.cc
View File

@@ -28,21 +28,22 @@ bool BiergartenPipelineOrchestrator::Run() {
.region_context = std::move(region_context)}); .region_context = std::move(region_context)});
} catch (const std::exception& exception) { } catch (const std::exception& exception) {
++skipped_count; ++skipped_count;
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = std::format( "[Pipeline] Skipping city '{}' ({}): context "
"[Pipeline] Skipping city '{}' ({}): context lookup failed: {}", "lookup failed: {}",
city.city, city.country, exception.what())}); city.city, city.country, exception.what())});
} }
} }
if (skipped_count > 0) { if (skipped_count > 0) {
logger_->Log({.level = LogLevel::Warn, logger_->Log(
.phase = PipelinePhase::Enrichment, {.level = LogLevel::Warn,
.message = std::format( .phase = PipelinePhase::Enrichment,
"[Pipeline] Skipped {} city/cities due to context lookup errors", .message = std::format("[Pipeline] Skipped {} city/cities due "
skipped_count)}); "to context lookup errors",
skipped_count)});
} }
this->GenerateUsers(enriched); this->GenerateUsers(enriched);
@@ -51,11 +52,10 @@ bool BiergartenPipelineOrchestrator::Run() {
this->LogResults(); this->LogResults();
return true; return true;
} catch (const std::exception& e) { } catch (const std::exception& e) {
logger_->Log( logger_->Log({.level = LogLevel::Error,
{.level = LogLevel::Error, .phase = PipelinePhase::Teardown,
.phase = PipelinePhase::Teardown, .message = std::format(
.message = "Pipeline execution failed with error: {}", e.what())});
std::format("Pipeline execution failed with error: {}", e.what())});
return false; return false;
} }
} }

42
tooling/pipeline/src/data_generation/llama/generate_brewery.cc
View File

@@ -83,8 +83,8 @@ BreweryResult LlamaGenerator::GenerateBrewery(
/** /**
* RETRY LOOP with validation and error correction * RETRY LOOP with validation and error correction
* Attempts to generate valid brewery data up to 3 times, with feedback-based * Attempts to generate valid brewery data up to 3 times, with
* refinement * feedback-based refinement
*/ */
constexpr int max_attempts = 3; constexpr int max_attempts = 3;
std::string raw; std::string raw;
@@ -103,7 +103,7 @@ BreweryResult LlamaGenerator::GenerateBrewery(
{.level = LogLevel::Debug, {.level = LogLevel::Debug,
.phase = PipelinePhase::BreweryAndBeerGeneration, .phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format("LlamaGenerator: raw output (attempt {}): {}", .message = std::format("LlamaGenerator: raw output (attempt {}): {}",
attempt + 1, raw)}); attempt + 1, raw)});
} }
// Validate output: parse JSON and check required fields // Validate output: parse JSON and check required fields
@@ -119,8 +119,9 @@ BreweryResult LlamaGenerator::GenerateBrewery(
logger_->Log( logger_->Log(
{.level = LogLevel::Info, {.level = LogLevel::Info,
.phase = PipelinePhase::BreweryAndBeerGeneration, .phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format("LlamaGenerator: successfully generated brewery data on attempt {}", .message = std::format("LlamaGenerator: successfully generated "
attempt + 1)}); "brewery data on attempt {}",
attempt + 1)});
} }
return brewery; return brewery;
@@ -133,33 +134,38 @@ BreweryResult LlamaGenerator::GenerateBrewery(
logger_->Log( logger_->Log(
{.level = LogLevel::Warn, {.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration, .phase = PipelinePhase::BreweryAndBeerGeneration,
.message = .message = std::format(
std::format("LlamaGenerator: malformed brewery JSON (attempt {}): {}", "LlamaGenerator: malformed brewery JSON (attempt {}): {}",
attempt + 1, *validation_error)}); attempt + 1, *validation_error)});
} }
// Update prompt with error details to guide LLM toward correct output. // Update prompt with error details to guide LLM toward correct output.
user_prompt = std::format( user_prompt = std::format(
"Your previous response was invalid. Error: {}\nReturn the thought " "Your previous response was invalid. Error: {}\nReturn the thought "
"process before the JSON if needed, then return ONLY valid JSON with " "process before the JSON if needed, then return ONLY valid JSON "
"exactly these keys, in this exact order: {{\"name_en\": \"<English " "with "
"exactly these keys, in this exact order: {{\"name_en\": "
"\"<English "
"brewery name>\", \"description_en\": \"<English single-paragraph " "brewery name>\", \"description_en\": \"<English single-paragraph "
"description>\", \"name_local\": \"<local-language brewery name>\", " "description>\", \"name_local\": \"<local-language brewery "
"name>\", "
"\"description_local\": \"<local-language single-paragraph " "\"description_local\": \"<local-language single-paragraph "
"description>\"}}.\nDo not include markdown, comments, extra keys, or " "description>\"}}.\nDo not include markdown, comments, extra keys, "
"literal placeholder values.\n\nKeep the JSON strings concise enough " "or "
"literal placeholder values.\n\nKeep the JSON strings concise "
"enough "
"to fit within the token budget.\n\n{}", "to fit within the token budget.\n\n{}",
*validation_error, retry_location); *validation_error, retry_location);
} }
// All retry attempts exhausted: log failure and throw exception // All retry attempts exhausted: log failure and throw exception
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Error,
{.level = LogLevel::Error, .phase = PipelinePhase::BreweryAndBeerGeneration,
.phase = PipelinePhase::BreweryAndBeerGeneration, .message = std::format(
.message = std::format( "LlamaGenerator: malformed brewery "
"LlamaGenerator: malformed brewery response after {} attempts: {}", "response after {} attempts: {}",
max_attempts, last_error.empty() ? raw : last_error)}); max_attempts, last_error.empty() ? raw : last_error)});
} }
throw std::runtime_error("LlamaGenerator: malformed brewery response"); throw std::runtime_error("LlamaGenerator: malformed brewery response");
} }

52
tooling/pipeline/src/data_generation/llama/generate_user.cc
View File

@@ -41,17 +41,17 @@ UserResult LlamaGenerator::GenerateUser(const EnrichedCity& city,
const std::string system_prompt = prompt_directory_->Load("USER_GENERATION"); const std::string system_prompt = prompt_directory_->Load("USER_GENERATION");
std::string user_prompt = std::format( std::string user_prompt = std::format(
"## NAME:\n{} {}\n\n## GENDER:\n{}\n\n## CITY:\n{}\n\n## COUNTRY:\n{}\n\n" "## NAME:\n{} {}\n\n## GENDER:\n{}\n\n## CITY:\n{}\n\n## "
"COUNTRY:\n{}\n\n"
"## PERSONA:\n{}\n\n## PERSONA DESCRIPTION:\n{}\n\n## STYLE " "## PERSONA:\n{}\n\n## PERSONA DESCRIPTION:\n{}\n\n## STYLE "
"AFFINITIES:\n{}", "AFFINITIES:\n{}",
name.first_name, name.last_name, name.gender, city.location.city, name.first_name, name.last_name, name.gender, city.location.city,
city.location.country, persona.name, persona.description, city.location.country, persona.name, persona.description,
style_affinities); style_affinities);
const std::string retry_context = const std::string retry_context = std::format(
std::format("Name: {} {}\nCity: {}, {}\nPersona: {}", name.first_name, "Name: {} {}\nCity: {}, {}\nPersona: {}", name.first_name, name.last_name,
name.last_name, city.location.city, city.location.country, city.location.city, city.location.country, persona.name);
persona.name);
constexpr int max_attempts = 3; constexpr int max_attempts = 3;
std::string raw; std::string raw;
@@ -59,14 +59,13 @@ UserResult LlamaGenerator::GenerateUser(const EnrichedCity& city,
int max_tokens = kUserInitialMaxTokens; int max_tokens = kUserInitialMaxTokens;
for (int attempt = 0; attempt < max_attempts; ++attempt) { for (int attempt = 0; attempt < max_attempts; ++attempt) {
raw = this->Infer(system_prompt, user_prompt, max_tokens, raw = this->Infer(system_prompt, user_prompt, max_tokens, kUserJsonGrammar);
kUserJsonGrammar);
if (logger_) { if (logger_) {
logger_->Log( logger_->Log(
{.level = LogLevel::Debug, {.level = LogLevel::Debug,
.phase = PipelinePhase::UserGeneration, .phase = PipelinePhase::UserGeneration,
.message = std::format("LlamaGenerator: raw output (attempt {}): {}", .message = std::format("LlamaGenerator: raw output (attempt {}): {}",
attempt + 1, raw)}); attempt + 1, raw)});
} }
UserResult user; UserResult user;
@@ -78,8 +77,9 @@ UserResult LlamaGenerator::GenerateUser(const EnrichedCity& city,
logger_->Log( logger_->Log(
{.level = LogLevel::Info, {.level = LogLevel::Info,
.phase = PipelinePhase::UserGeneration, .phase = PipelinePhase::UserGeneration,
.message = std::format("LlamaGenerator: successfully generated user data on attempt {}", .message = std::format("LlamaGenerator: successfully "
attempt + 1)}); "generated user data on attempt {}",
attempt + 1)});
} }
user.first_name = name.first_name; user.first_name = name.first_name;
@@ -90,32 +90,34 @@ UserResult LlamaGenerator::GenerateUser(const EnrichedCity& city,
last_error = *validation_error; last_error = *validation_error;
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::UserGeneration,
.phase = PipelinePhase::UserGeneration, .message = std::format(
.message = "LlamaGenerator: malformed user JSON (attempt {}): {}",
std::format("LlamaGenerator: malformed user JSON (attempt {}): {}", attempt + 1, *validation_error)});
attempt + 1, *validation_error)});
} }
user_prompt = std::format( user_prompt = std::format(
"Your previous response was invalid. Error: {}\nReturn the thought " "Your previous response was invalid. Error: {}\nReturn the thought "
"process before the JSON if needed, then return ONLY valid JSON with " "process before the JSON if needed, then return ONLY valid JSON "
"exactly these keys, in this exact order: {{\"username\": \"<handle " "with "
"exactly these keys, in this exact order: {{\"username\": "
"\"<handle "
"derived from the given name>\", \"bio\": \"<first-person bio " "derived from the given name>\", \"bio\": \"<first-person bio "
"grounded in the persona>\", \"activity_weight\": <number between 0 " "grounded in the persona>\", \"activity_weight\": <number between "
"0 "
"and 1>}}.\nDo not include markdown, comments, extra keys, or " "and 1>}}.\nDo not include markdown, comments, extra keys, or "
"literal placeholder values.\n\n{}", "literal placeholder values.\n\n{}",
*validation_error, retry_context); *validation_error, retry_context);
} }
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Error,
{.level = LogLevel::Error, .phase = PipelinePhase::UserGeneration,
.phase = PipelinePhase::UserGeneration, .message = std::format(
.message = std::format( "LlamaGenerator: malformed user response "
"LlamaGenerator: malformed user response after {} attempts: {}", "after {} attempts: {}",
max_attempts, last_error.empty() ? raw : last_error)}); max_attempts, last_error.empty() ? raw : last_error)});
} }
throw std::runtime_error("LlamaGenerator: malformed user response"); throw std::runtime_error("LlamaGenerator: malformed user response");
} }

3
tooling/pipeline/src/data_model/names_by_country.cc
View File

@@ -29,8 +29,7 @@ std::optional<Name> NamesByCountry::SampleName(const std::string& iso3166_1,
const std::vector<ForenameEntry>& forenames = forenames_it->second; const std::vector<ForenameEntry>& forenames = forenames_it->second;
const std::vector<std::string>& surnames = surnames_it->second; const std::vector<std::string>& surnames = surnames_it->second;
std::uniform_int_distribution<size_t> forename_dist(0, std::uniform_int_distribution<size_t> forename_dist(0, forenames.size() - 1);
forenames.size() - 1);
std::uniform_int_distribution<size_t> surname_dist(0, surnames.size() - 1); std::uniform_int_distribution<size_t> surname_dist(0, surnames.size() - 1);
const ForenameEntry& forename = forenames[forename_dist(rng)]; const ForenameEntry& forename = forenames[forename_dist(rng)];

27
tooling/pipeline/src/json_handling/json_loader.cc
View File

@@ -6,12 +6,10 @@
#include "json_handling/json_loader.h" #include "json_handling/json_loader.h"
#include <format>
#include "services/logging/logger.h"
#include <iostream>
#include <boost/json.hpp> #include <boost/json.hpp>
#include <format>
#include <fstream> #include <fstream>
#include <iostream>
#include <sstream> #include <sstream>
#include <stdexcept> #include <stdexcept>
#include <string> #include <string>
@@ -19,6 +17,8 @@
#include <unordered_map> #include <unordered_map>
#include <utility> #include <utility>
#include "services/logging/logger.h"
static std::string ReadRequiredString(const boost::json::object& object, static std::string ReadRequiredString(const boost::json::object& object,
const char* key) { const char* key) {
const boost::json::value* value = object.if_contains(key); const boost::json::value* value = object.if_contains(key);
@@ -45,7 +45,7 @@ static std::vector<std::string> ReadRequiredStringArray(
const boost::json::value* value = object.if_contains(key); const boost::json::value* value = object.if_contains(key);
if (value == nullptr || !value->is_array()) { if (value == nullptr || !value->is_array()) {
throw std::runtime_error( throw std::runtime_error(
std::format("Missing or invalid string array field: {}", key)); std::format("Missing or invalid string array field: {}", key));
} }
const auto& array = value->as_array(); const auto& array = value->as_array();
@@ -67,8 +67,8 @@ boost::json::value ParseJsonFile(const std::filesystem::path& filepath,
const char* what) { const char* what) {
std::ifstream input(filepath); std::ifstream input(filepath);
if (!input.is_open()) { if (!input.is_open()) {
throw std::runtime_error(std::format("Failed to open {} file: {}", what, throw std::runtime_error(
filepath.string())); std::format("Failed to open {} file: {}", what, filepath.string()));
} }
std::stringstream buffer; std::stringstream buffer;
@@ -87,8 +87,7 @@ boost::json::value ParseJsonFile(const std::filesystem::path& filepath,
/// `fallback_key` if `key` is missing/empty (some source entries only have a /// `fallback_key` if `key` is missing/empty (some source entries only have a
/// "localized" form and no "romanized" form). /// "localized" form and no "romanized" form).
std::string ReadFirstOfStringArray(const boost::json::object& object, std::string ReadFirstOfStringArray(const boost::json::object& object,
const char* key, const char* key, const char* fallback_key) {
const char* fallback_key) {
for (const char* candidate_key : {key, fallback_key}) { for (const char* candidate_key : {key, fallback_key}) {
const boost::json::value* value = object.if_contains(candidate_key); const boost::json::value* value = object.if_contains(candidate_key);
if (value == nullptr || !value->is_array()) { if (value == nullptr || !value->is_array()) {
@@ -163,8 +162,7 @@ std::vector<UserPersona> JsonLoader::LoadPersonas(
personas.push_back(UserPersona{ personas.push_back(UserPersona{
.name = ReadRequiredString(object, "name"), .name = ReadRequiredString(object, "name"),
.description = ReadRequiredString(object, "description"), .description = ReadRequiredString(object, "description"),
.style_affinities = .style_affinities = ReadRequiredStringArray(object, "style_affinities"),
ReadRequiredStringArray(object, "style_affinities"),
}); });
} }
@@ -207,8 +205,8 @@ NamesByCountry JsonLoader::LoadNamesByCountry(
} }
const auto& name_object = name_value.as_object(); const auto& name_object = name_value.as_object();
entries.push_back(ForenameEntry{ entries.push_back(ForenameEntry{
.name = ReadFirstOfStringArray(name_object, "romanized", .name =
"localized"), ReadFirstOfStringArray(name_object, "romanized", "localized"),
.gender = ReadRequiredString(name_object, "gender"), .gender = ReadRequiredString(name_object, "gender"),
}); });
} }
@@ -216,8 +214,7 @@ NamesByCountry JsonLoader::LoadNamesByCountry(
forenames_by_country.emplace(country_code, std::move(entries)); forenames_by_country.emplace(country_code, std::move(entries));
} }
std::unordered_map<std::string, std::vector<std::string>> std::unordered_map<std::string, std::vector<std::string>> surnames_by_country;
surnames_by_country;
for (const auto& [country_code, name_entries] : surnames_root.as_object()) { for (const auto& [country_code, name_entries] : surnames_root.as_object()) {
if (!name_entries.is_array()) { if (!name_entries.is_array()) {
continue; continue;

49
tooling/pipeline/src/main.cc
View File

@@ -67,8 +67,8 @@ int main(const int argc, char** argv) {
const LlamaBackendState llama_backend_state; const LlamaBackendState llama_backend_state;
#endif #endif
log_producer->Log({.level = LogLevel::Info, log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "STARTING PIPELINE"}); .message = "STARTING PIPELINE"});
const std::optional<ApplicationOptions> parsed_options = const std::optional<ApplicationOptions> parsed_options =
@@ -90,8 +90,8 @@ int main(const int argc, char** argv) {
prompt_directory = std::make_unique<PromptDirectory>( prompt_directory = std::make_unique<PromptDirectory>(
options.pipeline.prompt_dir, log_producer); options.pipeline.prompt_dir, log_producer);
} catch (const std::exception& dir_error) { } catch (const std::exception& dir_error) {
log_producer->Log({.level = LogLevel::Error, log_producer->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = std::format("Invalid --prompt-dir: {}", .message = std::format("Invalid --prompt-dir: {}",
dir_error.what())}); dir_error.what())});
@@ -109,7 +109,7 @@ int main(const int argc, char** argv) {
{ {
log_producer->Log( log_producer->Log(
{.level = LogLevel::Info, {.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "Prompt formatter: none (mock mode)"}); .message = "Prompt formatter: none (mock mode)"});
} }
return std::unique_ptr<IPromptFormatter>(nullptr); return std::unique_ptr<IPromptFormatter>(nullptr);
@@ -117,7 +117,7 @@ int main(const int argc, char** argv) {
{ {
log_producer->Log( log_producer->Log(
{.level = LogLevel::Info, {.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "Prompt formatter: Gemma4JinjaPromptFormatter"}); .message = "Prompt formatter: Gemma4JinjaPromptFormatter"});
} }
return std::unique_ptr<IPromptFormatter>( return std::unique_ptr<IPromptFormatter>(
@@ -126,15 +126,15 @@ int main(const int argc, char** argv) {
di::bind<WebClient>().to([options, log_producer] { di::bind<WebClient>().to([options, log_producer] {
if (options.generator.use_mocked) { if (options.generator.use_mocked) {
{ {
log_producer->Log({.level = LogLevel::Info, log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "Web client: none (mock mode)"}); .message = "Web client: none (mock mode)"});
} }
return std::unique_ptr<WebClient>(nullptr); return std::unique_ptr<WebClient>(nullptr);
} }
{ {
log_producer->Log({.level = LogLevel::Info, log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "Web client: HttpWebClient"}); .message = "Web client: HttpWebClient"});
} }
return std::unique_ptr<WebClient>( return std::unique_ptr<WebClient>(
@@ -145,15 +145,15 @@ int main(const int argc, char** argv) {
const auto& inj) -> std::unique_ptr<IEnrichmentService> { const auto& inj) -> std::unique_ptr<IEnrichmentService> {
if (options.generator.use_mocked) { if (options.generator.use_mocked) {
{ {
log_producer->Log({.level = LogLevel::Info, log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "Enrichment: mock"}); .message = "Enrichment: mock"});
} }
return std::make_unique<MockEnrichmentService>(); return std::make_unique<MockEnrichmentService>();
} }
{ {
log_producer->Log({.level = LogLevel::Info, log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "Enrichment: Wikipedia"}); .message = "Enrichment: Wikipedia"});
} }
return std::make_unique<WikipediaEnrichmentService>( return std::make_unique<WikipediaEnrichmentService>(
@@ -165,8 +165,8 @@ int main(const int argc, char** argv) {
&log_producer](const auto& inj) -> std::unique_ptr<DataGenerator> { &log_producer](const auto& inj) -> std::unique_ptr<DataGenerator> {
if (options.generator.use_mocked) { if (options.generator.use_mocked) {
{ {
log_producer->Log({.level = LogLevel::Info, log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = "Generator: mock"}); .message = "Generator: mock"});
} }
return std::make_unique<MockGenerator>(); return std::make_unique<MockGenerator>();
@@ -174,9 +174,10 @@ int main(const int argc, char** argv) {
{ {
log_producer->Log( log_producer->Log(
{.level = LogLevel::Info, {.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = std::format( .message = std::format(
"Generator: LlamaGenerator | model={} | temp={:.2f} " "Generator: LlamaGenerator | model={} | "
"temp={:.2f} "
"top_p={:.2f} top_k={} n_ctx={} seed={}", "top_p={:.2f} top_k={} n_ctx={} seed={}",
model_path, sampling.temperature, sampling.top_p, model_path, sampling.temperature, sampling.top_p,
sampling.top_k, sampling.n_ctx, sampling.seed)}); sampling.top_k, sampling.n_ctx, sampling.seed)});
@@ -191,14 +192,14 @@ int main(const int argc, char** argv) {
injector.create<std::unique_ptr<BiergartenPipelineOrchestrator>>(); injector.create<std::unique_ptr<BiergartenPipelineOrchestrator>>();
if (!orchestrator->Run()) { if (!orchestrator->Run()) {
log_producer->Log({.level = LogLevel::Error, log_producer->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Teardown, .phase = PipelinePhase::Teardown,
.message = "Pipeline execution failed"}); .message = "Pipeline execution failed"});
return shutdown(EXIT_FAILURE); return shutdown(EXIT_FAILURE);
} }
log_producer->Log({.level = LogLevel::Info, log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Teardown, .phase = PipelinePhase::Teardown,
.message = std::format("Pipeline complete in {} ms", .message = std::format("Pipeline complete in {} ms",
timer.Elapsed())}); timer.Elapsed())});
@@ -206,8 +207,8 @@ int main(const int argc, char** argv) {
} catch (const std::exception& exception) { } catch (const std::exception& exception) {
const LogDTO log_entry{.level = LogLevel::Error, const LogDTO log_entry{.level = LogLevel::Error,
.phase = PipelinePhase::Teardown, .phase = PipelinePhase::Teardown,
.message = exception.what()}; .message = exception.what()};
if (log_producer) { if (log_producer) {
log_producer->Log(log_entry); log_producer->Log(log_entry);
} else { } else {

92
tooling/pipeline/src/services/enrichment/wikipedia/fetch_extract.cc
View File

@@ -20,9 +20,10 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (const auto cache_it = this->extract_cache_.find(cache_key); if (const auto cache_it = this->extract_cache_.find(cache_key);
cache_it != this->extract_cache_.end()) { cache_it != this->extract_cache_.end()) {
if (logger_) { if (logger_) {
logger_->Log({.level = LogLevel::Debug, logger_->Log(
.phase = PipelinePhase::Enrichment, {.level = LogLevel::Debug,
.message = std::format("Wikipedia: Cache hit for {}!", cache_key)}); .phase = PipelinePhase::Enrichment,
.message = std::format("Wikipedia: Cache hit for {}!", cache_key)});
} }
return cache_it->second; return cache_it->second;
} }
@@ -30,7 +31,8 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
const std::string encoded = this->client_->EncodeURL(cache_key); const std::string encoded = this->client_->EncodeURL(cache_key);
const std::string url = std::format( const std::string url = std::format(
"https://en.wikipedia.org/w/" "https://en.wikipedia.org/w/"
"api.php?action=query&titles={}&prop=extracts&explaintext=1&format=json", "api.php?action=query&titles={}&prop=extracts&explaintext=1&format="
"json",
encoded); encoded);
const std::string body = this->client_->Get(url); const std::string body = this->client_->Get(url);
@@ -45,11 +47,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (ec) { if (ec) {
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = std::format("WikipediaService: JSON parse error for '{}': {}", "WikipediaService: JSON parse error for '{}': {}",
std::string(query), ec.message())}); std::string(query), ec.message())});
} }
return {}; return {};
} }
@@ -58,12 +60,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
const json::object* obj = doc.if_object(); const json::object* obj = doc.if_object();
if (obj == nullptr) { if (obj == nullptr) {
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = "WikipediaService: Expected root object for '{}'",
std::format("WikipediaService: Expected root object for '{}'", std::string(query))});
std::string(query))});
} }
return {}; return {};
} }
@@ -76,12 +77,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if ((pages_ptr == nullptr) || !pages_ptr->is_object()) { if ((pages_ptr == nullptr) || !pages_ptr->is_object()) {
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = "WikipediaService: Missing query.pages for '{}'",
std::format("WikipediaService: Missing query.pages for '{}'", std::string(query))});
std::string(query))});
} }
return {}; return {};
} }
@@ -90,11 +90,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (pages.empty()) { if (pages.empty()) {
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = std::format("WikipediaService: No pages returned for '{}'", "WikipediaService: No pages returned for '{}'",
std::string(query))}); std::string(query))});
} }
this->extract_cache_.emplace(cache_key, ""); this->extract_cache_.emplace(cache_key, "");
return {}; return {};
@@ -106,12 +106,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (!page_val.is_object()) { if (!page_val.is_object()) {
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = "WikipediaService: Unexpected page format for '{}'",
std::format("WikipediaService: Unexpected page format for '{}'", std::string(query))});
std::string(query))});
} }
return {}; return {};
} }
@@ -121,10 +120,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
// Handle 404/Missing status // Handle 404/Missing status
if (page.contains("missing")) { if (page.contains("missing")) {
if (logger_) { if (logger_) {
logger_->Log({.level = LogLevel::Warn, logger_->Log(
.phase = PipelinePhase::Enrichment, {.level = LogLevel::Warn,
.message = std::format("WikipediaService: Page '{}' does not exist", .phase = PipelinePhase::Enrichment,
std::string(query))}); .message = std::format("WikipediaService: Page '{}' does not exist",
std::string(query))});
} }
this->extract_cache_.emplace(cache_key, ""); this->extract_cache_.emplace(cache_key, "");
return {}; return {};
@@ -134,12 +134,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if ((extract_ptr == nullptr) || !extract_ptr->is_string()) { if ((extract_ptr == nullptr) || !extract_ptr->is_string()) {
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Warn,
{.level = LogLevel::Warn, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = "WikipediaService: No extract string found for '{}'",
std::format("WikipediaService: No extract string found for '{}'", std::string(query))});
std::string(query))});
} }
this->extract_cache_.emplace(cache_key, ""); this->extract_cache_.emplace(cache_key, "");
return {}; return {};
@@ -148,10 +147,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
// 4. Success // 4. Success
std::string extract(extract_ptr->as_string()); std::string extract(extract_ptr->as_string());
if (logger_) { if (logger_) {
logger_->Log({.level = LogLevel::Info, logger_->Log(
.phase = PipelinePhase::Enrichment, {.level = LogLevel::Info,
.message = std::format("WikipediaService: Fetched {} chars for '{}'", .phase = PipelinePhase::Enrichment,
extract.size(), std::string(query))}); .message = std::format("WikipediaService: Fetched {} chars for '{}'",
extract.size(), std::string(query))});
} }
this->extract_cache_.insert_or_assign(cache_key, extract); this->extract_cache_.insert_or_assign(cache_key, extract);

7
tooling/pipeline/src/services/enrichment/wikipedia/get_summary.cc
View File

@@ -45,8 +45,9 @@ std::string WikipediaEnrichmentService::GetLocationContext(
if (logger_) { if (logger_) {
logger_->Log({.level = LogLevel::Info, logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Enrichment, .phase = PipelinePhase::Enrichment,
.message = std::format("Done fetching for {}. Sleeping for 10 seconds.", .message = std::format(
location_query)}); "Done fetching for {}. Sleeping for 10 seconds.",
location_query)});
} }
std::this_thread::sleep_for(10s); std::this_thread::sleep_for(10s);
@@ -56,7 +57,7 @@ std::string WikipediaEnrichmentService::GetLocationContext(
{.level = LogLevel::Debug, {.level = LogLevel::Debug,
.phase = PipelinePhase::Enrichment, .phase = PipelinePhase::Enrichment,
.message = std::format("WikipediaService lookup failed for '{}': {}", .message = std::format("WikipediaService lookup failed for '{}': {}",
location_query, e.what())}); location_query, e.what())});
} }
} }
return result; return result;

21
tooling/pipeline/src/services/prompt_directory.cc
View File

@@ -44,8 +44,8 @@ PromptDirectory::PromptDirectory(const std::filesystem::path& prompt_dir,
// Scenario 4: directory must be readable (probe with directory_iterator). // Scenario 4: directory must be readable (probe with directory_iterator).
std::filesystem::directory_iterator probe(prompt_dir_, ec); std::filesystem::directory_iterator probe(prompt_dir_, ec);
if (ec) { if (ec) {
throw std::runtime_error( throw std::runtime_error(std::format(
std::format("PromptDirectory: prompt directory is not readable: {} ({})", "PromptDirectory: prompt directory is not readable: {} ({})",
prompt_dir_.string(), ec.message())); prompt_dir_.string(), ec.message()));
} }
@@ -76,7 +76,7 @@ std::string PromptDirectory::Load(std::string_view key) {
if (!file.is_open()) { if (!file.is_open()) {
throw std::runtime_error( throw std::runtime_error(
std::format("PromptDirectory: prompt file not found for key '{}': {}", std::format("PromptDirectory: prompt file not found for key '{}': {}",
key_str, file_path.string())); key_str, file_path.string()));
} }
std::string content((std::istreambuf_iterator<char>(file)), std::string content((std::istreambuf_iterator<char>(file)),
@@ -84,15 +84,18 @@ std::string PromptDirectory::Load(std::string_view key) {
file.close(); file.close();
if (content.empty()) { if (content.empty()) {
throw std::runtime_error(std::format("PromptDirectory: prompt file for key '{}' is empty: {}", throw std::runtime_error(
key_str, file_path.string())); std::format("PromptDirectory: prompt file for key '{}' is empty: {}",
key_str, file_path.string()));
} }
if (logger_) { if (logger_) {
logger_->Log({.level = LogLevel::Info, logger_->Log(
.phase = PipelinePhase::Startup, {.level = LogLevel::Info,
.message = std::format("[PromptDirectory] Loaded prompt '{}' from '{}' ({} chars)", .phase = PipelinePhase::Startup,
key_str, file_path.string(), content.size())}); .message = std::format(
"[PromptDirectory] Loaded prompt '{}' from '{}' ({} chars)",
key_str, file_path.string(), content.size())});
} }
cache_.emplace(key_str, content); cache_.emplace(key_str, content);

6
tooling/pipeline/src/services/sqlite/initialize.cc
View File

@@ -18,9 +18,9 @@ std::filesystem::path SqliteExportService::BuildDatabasePath() const {
std::filesystem::path candidate = output_path_ / base_filename; std::filesystem::path candidate = output_path_ / base_filename;
for (int suffix = 1; std::filesystem::exists(candidate); ++suffix) { for (int suffix = 1; std::filesystem::exists(candidate); ++suffix) {
candidate = output_path_ / candidate = output_path_ / std::filesystem::path(
std::filesystem::path(std::format("biergarten_seed_{}-{}.sqlite", std::format("biergarten_seed_{}-{}.sqlite",
run_timestamp_utc_, suffix)); run_timestamp_utc_, suffix));
} }
return candidate; return candidate;

6
tooling/pipeline/src/services/sqlite/process_record.cc
View File

@@ -38,8 +38,7 @@ sqlite3_int64 SqliteExportService::ResolveLocationId(const Location& location) {
} }
const std::string local_languages_json = const std::string local_languages_json =
sqlite_export_service_internal::SerializeVector( sqlite_export_service_internal::SerializeVector(location.local_languages);
location.local_languages);
sqlite_export_service_internal::Bind( sqlite_export_service_internal::Bind(
insert_location_stmt_, insert_location_stmt_,
@@ -92,7 +91,8 @@ sqlite3_int64 SqliteExportService::ResolveLocationId(const Location& location) {
.action = "Failed to bind SQLite location longitude"}); .action = "Failed to bind SQLite location longitude"});
sqlite_export_service_internal::StepStatement( sqlite_export_service_internal::StepStatement(
db_handle_, insert_location_stmt_, "Failed to insert SQLite location row"); db_handle_, insert_location_stmt_,
"Failed to insert SQLite location row");
const sqlite3_int64 location_id = const sqlite3_int64 location_id =
sqlite_export_service_internal::LastInsertRowId(db_handle_); sqlite_export_service_internal::LastInsertRowId(db_handle_);

17
tooling/pipeline/src/web_client/http_web_client.cc
View File

@@ -48,21 +48,20 @@ std::string HttpWebClient::Get(const std::string& url) {
const httplib::Result result = client.Get(path); const httplib::Result result = client.Get(path);
if (!result) { if (!result) {
throw std::runtime_error(std::format( throw std::runtime_error(
"[HttpWebClient] Request failed for URL: {} — {}", url, std::format("[HttpWebClient] Request failed for URL: {} — {}", url,
httplib::to_string(result.error()))); httplib::to_string(result.error())));
} }
if (result->status < kSuccessMin || result->status >= kSuccessMax) { if (result->status < kSuccessMin || result->status >= kSuccessMax) {
if (logger_) { if (logger_) {
logger_->Log( logger_->Log({.level = LogLevel::Error,
{.level = LogLevel::Error, .phase = PipelinePhase::Enrichment,
.phase = PipelinePhase::Enrichment, .message = std::format(
.message = "[HttpWebClient] Request failed for URL: {}", url)});
std::format("[HttpWebClient] Request failed for URL: {}", url)});
} }
throw std::runtime_error(std::format("[HttpWebClient] HTTP {} for URL: {}", throw std::runtime_error(std::format("[HttpWebClient] HTTP {} for URL: {}",
result->status, url)); result->status, url));
} }
return result->body; return result->body;