Add gazetteer to analyze worker
Pass a `Gazetteer` to the analyze worker to enable proper-name correction. The gazetteer is loaded from disk at application startup. If the directory is missing or unreadable, the worker will start without proper-name correction capabilities.
This commit is contained in:
@@ -7,18 +7,26 @@ use tracing::{error, info, warn};
|
|||||||
|
|
||||||
use super::{llm, prompt, AnalysisInput, AnalyzeReceiver, ANALYSIS_INPUT_FILE, DOCUMENT_FILE};
|
use super::{llm, prompt, AnalysisInput, AnalyzeReceiver, ANALYSIS_INPUT_FILE, DOCUMENT_FILE};
|
||||||
use crate::config::Config;
|
use crate::config::Config;
|
||||||
|
use crate::gazetteer::Gazetteer;
|
||||||
use crate::{BusyGuard, WorkerBusy};
|
use crate::{BusyGuard, WorkerBusy};
|
||||||
|
|
||||||
/// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
|
/// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
|
||||||
/// but sequential processing keeps the architecture simple and shields the
|
/// but sequential processing keeps the architecture simple and shields the
|
||||||
/// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`.
|
/// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`.
|
||||||
|
///
|
||||||
|
/// `_vocab` is the gazetteer for proper-name correction. Currently unused;
|
||||||
|
/// wired into `process()` in a follow-up step.
|
||||||
pub async fn run(
|
pub async fn run(
|
||||||
mut rx: AnalyzeReceiver,
|
mut rx: AnalyzeReceiver,
|
||||||
config: Arc<Config>,
|
config: Arc<Config>,
|
||||||
client: reqwest::Client,
|
client: reqwest::Client,
|
||||||
worker_busy: WorkerBusy,
|
worker_busy: WorkerBusy,
|
||||||
|
_vocab: Arc<Gazetteer>,
|
||||||
) {
|
) {
|
||||||
info!("Analyze worker started");
|
info!(
|
||||||
|
vocab_entries = _vocab.len(),
|
||||||
|
"Analyze worker started"
|
||||||
|
);
|
||||||
let timeout = Duration::from_secs(config.llm_timeout_seconds);
|
let timeout = Duration::from_secs(config.llm_timeout_seconds);
|
||||||
|
|
||||||
while let Some(job) = rx.recv().await {
|
while let Some(job) = rx.recv().await {
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ use tracing_subscriber::EnvFilter;
|
|||||||
|
|
||||||
use doctate_server::analyze;
|
use doctate_server::analyze;
|
||||||
use doctate_server::config::Config;
|
use doctate_server::config::Config;
|
||||||
|
use doctate_server::gazetteer::Gazetteer;
|
||||||
use doctate_server::transcribe;
|
use doctate_server::transcribe;
|
||||||
use doctate_server::AppState;
|
use doctate_server::AppState;
|
||||||
|
|
||||||
@@ -64,6 +65,27 @@ async fn main() {
|
|||||||
.build()
|
.build()
|
||||||
.expect("reqwest client build failed");
|
.expect("reqwest client build failed");
|
||||||
|
|
||||||
|
// Gazetteer for pre-LLM proper-name correction. Missing / unreadable
|
||||||
|
// vocab directory is non-fatal — analysis falls back to plain prompts.
|
||||||
|
let vocab = Arc::new(match Gazetteer::load_dir(&config.vocab_dir).await {
|
||||||
|
Ok(g) => {
|
||||||
|
info!(
|
||||||
|
dir = %config.vocab_dir.display(),
|
||||||
|
entries = g.len(),
|
||||||
|
"Gazetteer loaded"
|
||||||
|
);
|
||||||
|
g
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
warn!(
|
||||||
|
dir = %config.vocab_dir.display(),
|
||||||
|
error = %e,
|
||||||
|
"Gazetteer not available; running without proper-name correction"
|
||||||
|
);
|
||||||
|
Gazetteer::empty()
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
// Transcription pipeline: channel + worker + recovery scan.
|
// Transcription pipeline: channel + worker + recovery scan.
|
||||||
let (transcribe_tx, transcribe_rx) = transcribe::channel();
|
let (transcribe_tx, transcribe_rx) = transcribe::channel();
|
||||||
let transcribe_busy: doctate_server::WorkerBusy =
|
let transcribe_busy: doctate_server::WorkerBusy =
|
||||||
@@ -96,6 +118,7 @@ async fn main() {
|
|||||||
config.clone(),
|
config.clone(),
|
||||||
http_client.clone(),
|
http_client.clone(),
|
||||||
analyze_busy.clone(),
|
analyze_busy.clone(),
|
||||||
|
vocab.clone(),
|
||||||
));
|
));
|
||||||
{
|
{
|
||||||
let tx = analyze_tx.clone();
|
let tx = analyze_tx.clone();
|
||||||
|
|||||||
@@ -332,7 +332,8 @@ async fn analyze_worker_writes_document_via_wiremock() {
|
|||||||
let (tx, rx) = analyze::channel();
|
let (tx, rx) = analyze::channel();
|
||||||
let client = reqwest::Client::new();
|
let client = reqwest::Client::new();
|
||||||
let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||||
let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy));
|
let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty());
|
||||||
|
let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy, vocab));
|
||||||
|
|
||||||
tx.send(analyze::AnalyzeJob {
|
tx.send(analyze::AnalyzeJob {
|
||||||
case_dir: case_dir.clone(),
|
case_dir: case_dir.clone(),
|
||||||
@@ -863,7 +864,8 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() {
|
|||||||
let (tx, rx) = analyze::channel();
|
let (tx, rx) = analyze::channel();
|
||||||
let client = reqwest::Client::new();
|
let client = reqwest::Client::new();
|
||||||
let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||||
let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy));
|
let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty());
|
||||||
|
let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy, vocab));
|
||||||
|
|
||||||
tx.send(analyze::AnalyzeJob {
|
tx.send(analyze::AnalyzeJob {
|
||||||
case_dir: case_dir.clone(),
|
case_dir: case_dir.clone(),
|
||||||
|
|||||||
Reference in New Issue
Block a user