added verbosity to pipeline

src/pipeline/pipeline.mjs

@@ -11,21 +11,29 @@ import { runReward } from '../reward/reward_core.mjs';
 
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = path.dirname(__filename);
-
 const PROJECT_ROOT = path.join(__dirname, '..', '..');
 
 const DEFAULT_SEEDS_PATH = path.join(
   PROJECT_ROOT,
   'test_samples',
-  'seed_questions.jsonl'
+  'seed_questions.jsonl',
 );
 
 const DEFAULT_OUT_PATH = path.join(
   PROJECT_ROOT,
   'gold',
-  'pipeline_gold.jsonl'
+  'pipeline_gold.jsonl',
 );
 
+function preview(value, max = 400) {
+  if (value == null) return '';
+  let str = typeof value === 'string' ? value : JSON.stringify(value, null, 2);
+  if (str.length > max) {
+    return str.slice(0, max) + `… [truncated ${str.length - max} chars]`;
+  }
+  return str;
+}
+
 /**
  * Load JSONL seed questions.
  * Each line may be:
@@ -59,72 +67,164 @@ export function seedToQuestion(seed) {
  *
  * Returns a structured result:
  *   {
- *     status: 'accepted' | 'invalid_question'
- *                      | 'generator_failed'
- *                      | 'verifier_rejected'
- *                      | 'reward_rejected',
+ *     status: 'accepted' | 'invalid_question' | 'retrieval_failed'
+ *             | 'generator_failed' | 'verifier_rejected'
+ *             | 'reward_rejected'  | 'verifier_error' | 'reward_error',
  *     question,
  *     context,
  *     gen,
  *     ver,
  *     rew,
+ *     error? // optional message
  *   }
  */
 export async function runPipelineStep({
   question,
   retrievalMode = process.env.RETRIEVAL_MODE || 'hybrid',
   k = Number(process.env.RETRIEVAL_K || '6'),
-  providers
+  generatorProvider,
+  verifierProvider,
+  rewardProvider,
+  verbose = false,
+  logger = console,
 } = {}) {
+  const log = logger?.log?.bind(logger) || console.log;
+  const errLog = logger?.error?.bind(logger) || console.error;
+
   if (!question || !question.trim()) {
+    if (verbose) log('   [pipeline] empty / invalid question, skipping');
     return { status: 'invalid_question', question };
   }
 
-  // Let provider.mjs decide env mapping; we just say which stage.
-  const generatorProvider =
-    providers?.generator || loadProviderFor('generator');
-  const verifierProvider =
-    providers?.verifier || loadProviderFor('verifier');
-  const rewardProvider =
-    providers?.reward || loadProviderFor('reward');
+  const genProv = generatorProvider || loadProviderFor('generator');
+  const verProv = verifierProvider || loadProviderFor('verifier');
+  const rewProv = rewardProvider || loadProviderFor('reward');
 
   // --- Retrieval ---
   let context = [];
-  if (retrievalMode === 'hybrid' || !retrievalMode) {
-    context = await hybridSearch(question, k);
-  } else {
-    // Additional modes could be added later
+  try {
+    if (verbose) log(`   [retrieval] mode=${retrievalMode} k=${k}`);
     context = await hybridSearch(question, k);
+    if (verbose) {
+      log(`   [retrieval] got ${context.length} chunks`);
+      if (context.length > 0) {
+        const first = context[0]?.content ?? '';
+        log('   [retrieval] first chunk:');
+        log('   ' + preview(first, 200).replace(/\n/g, '\n   '));
+      }
+    }
+  } catch (e) {
+    const msg = e?.message || String(e);
+    if (verbose) errLog('   [retrieval] ERROR:', msg);
+    return {
+      status: 'retrieval_failed',
+      question,
+      error: msg,
+    };
   }
 
   // --- Generator ---
-  const gen = await runGenerator(question, context, generatorProvider);
-
-  // If generator decisively failed or emitted obviously bad JSON,
-  // treat as failure.
-  if (!gen || gen.parsed?.error === 'invalid_json') {
-    return { status: 'generator_failed', question, context, gen };
+  let gen;
+  try {
+    if (verbose) log('   [generator] calling model…');
+    // NOTE: runGenerator(query, contextChunks, provider)
+    gen = await runGenerator(question, context, genProv);
+    if (verbose) {
+      log('   [generator] raw:');
+      log('   ' + preview(gen.raw ?? '', 400).replace(/\n/g, '\n   '));
+      log('   [generator] parsed:');
+      log('   ' + preview(gen.parsed, 400).replace(/\n/g, '\n   '));
+    }
+  } catch (e) {
+    const msg = e?.message || String(e);
+    if (verbose) errLog('   [generator] ERROR:', msg);
+    return {
+      status: 'generator_failed',
+      question,
+      context,
+      error: msg,
+    };
   }
 
   // --- Verifier ---
-  const ver = await runVerifier(gen, verifierProvider);
-  if (!ver || ver.ok === false) {
-    return { status: 'verifier_rejected', question, context, gen, ver };
+  let ver;
+  try {
+    if (verbose) log('   [verifier] calling model…');
+    // NOTE: runVerifier(sample, provider)
+    ver = await runVerifier(gen, verProv);
+    if (verbose) {
+      log('   [verifier] parsed:');
+      log('   ' + preview(ver.parsed, 400).replace(/\n/g, '\n   '));
+      log(`   [verifier] ok=${ver.ok === true}`);
+    }
+  } catch (e) {
+    const msg = e?.message || String(e);
+    if (verbose) errLog('   [verifier] ERROR:', msg);
+    return {
+      status: 'verifier_error',
+      question,
+      context,
+      gen,
+      error: msg,
+    };
+  }
+
+  if (!ver || ver.ok !== true) {
+    if (verbose) log('   [verifier] rejected sample');
+    return {
+      status: 'verifier_rejected',
+      question,
+      context,
+      gen,
+      ver,
+    };
   }
 
   // --- Reward ---
-  const rew = await runReward(gen, rewardProvider);
-  if (!rew || rew.ok === false) {
-    return { status: 'reward_rejected', question, context, gen, ver, rew };
+  let rew;
+  try {
+    if (verbose) log('   [reward] calling model…');
+    // NOTE: runReward(sample, provider)
+    rew = await runReward(gen, rewProv);
+    if (verbose) {
+      log('   [reward] parsed:');
+      log('   ' + preview(rew.parsed, 400).replace(/\n/g, '\n   '));
+      log(`   [reward] score=${rew.score} ok=${rew.ok}`);
+    }
+  } catch (e) {
+    const msg = e?.message || String(e);
+    if (verbose) errLog('   [reward] ERROR:', msg);
+    return {
+      status: 'reward_error',
+      question,
+      context,
+      gen,
+      ver,
+      error: msg,
+    };
+  }
+
+  if (!rew || rew.ok !== true) {
+    if (verbose) log('   [reward] rejected sample');
+    return {
+      status: 'reward_rejected',
+      question,
+      context,
+      gen,
+      ver,
+      rew,
+    };
   }
 
+  if (verbose) log('   [pipeline] accepted ✅');
+
   return {
     status: 'accepted',
     question,
     context,
     gen,
     ver,
-    rew
+    rew,
   };
 }
 
@@ -145,45 +245,69 @@ export async function appendGoldRecord(outPath, record) {
  *   - seedsPath: JSONL of seeds (defaults to test_samples/seed_questions.jsonl)
  *   - outPath:   output JSONL (defaults to gold/pipeline_gold.jsonl)
  *   - limit:     max number of seeds to process
+ *   - verbose:   extra per-stage logging
+ *   - logger:    optional logger (defaults to console)
  *
  * Returns:
- *   { total, processed, accepted, outPath }
+ *   { total, processed, accepted, outPath, statusCounts }
  */
 export async function runPipelineBatch({
   seedsPath = DEFAULT_SEEDS_PATH,
   outPath = DEFAULT_OUT_PATH,
-  limit
+  limit,
+  verbose = false,
+  logger = console,
 } = {}) {
+  const log = logger?.log?.bind(logger) || console.log;
+  const errLog = logger?.error?.bind(logger) || console.error;
+
   const seeds = await loadSeedQuestions(seedsPath);
   const max = typeof limit === 'number' ? limit : seeds.length;
 
-  // Shared provider instances for the whole batch
-  const providers = {
-    generator: loadProviderFor('generator'),
-    verifier: loadProviderFor('verifier'),
-    reward: loadProviderFor('reward')
-  };
-
   let processed = 0;
   let accepted = 0;
+  const statusCounts = {};
 
-  for (const seed of seeds.slice(0, max)) {
+  for (let idx = 0; idx < max; idx++) {
+    const seed = seeds[idx];
     const question = seedToQuestion(seed);
-    const result = await runPipelineStep({ question, providers });
+    const label = `[${idx + 1}/${max}]`;
 
-    processed += 1;
+    log(`→ ${label} Running pipeline for: "${question}"`);
 
-    if (result.status === 'accepted') {
-      const record = {
+    try {
+      const result = await runPipelineStep({
         question,
-        context: result.context,
-        sample: result.gen, // treat generator result as the core sample
-        verifier: result.ver,
-        reward: result.rew
-      };
-
-      await appendGoldRecord(outPath, record);
-      accepted += 1;
+        verbose,
+        logger,
+      });
+
+      processed += 1;
+      statusCounts[result.status] =
+        (statusCounts[result.status] || 0) + 1;
+
+      if (verbose) {
+        log(`   ↳ status: ${result.status}`);
+      }
+
+      if (result.status === 'accepted') {
+        const record = {
+          question,
+          context: result.context,
+          sample: result.gen, // generator output
+          verifier: result.ver,
+          reward: result.rew,
+        };
+
+        await appendGoldRecord(outPath, record);
+        accepted += 1;
+      }
+    } catch (e) {
+      const msg = e?.message || String(e);
+      processed += 1;
+      statusCounts.pipeline_error =
+        (statusCounts.pipeline_error || 0) + 1;
+      errLog('   [pipeline] ERROR:', msg);
     }
   }
 
@@ -191,6 +315,7 @@ export async function runPipelineBatch({
     total: seeds.length,
     processed,
     accepted,
-    outPath
+    outPath,
+    statusCounts,
   };
 }

src/pipeline/pipeline_cli.js

@@ -1,117 +1,130 @@
 #!/usr/bin/env node
 // src/pipeline/pipeline_cli.js
-
-import { fileURLToPath } from 'url';
 import path from 'path';
-import fs from 'fs/promises';
-
-import { loadProvider } from '../providers/provider.mjs';
-import {
-  loadSeedQuestions,
-  seedToQuestion,
-  runPipelineStep,
-  appendGoldRecord,
-  runPipelineBatch
-} from './pipeline.mjs';
+import { fileURLToPath } from 'url';
+import { runPipelineBatch } from './pipeline.mjs';
 
-//
-// ---- CLI Helpers ----
-//
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = path.dirname(__filename);
-
-function parseArgs() {
-  const args = process.argv.slice(2);
-  const out = {};
+const PROJECT_ROOT = path.join(__dirname, '..', '..');
+
+const DEFAULT_SEEDS = path.join(
+  PROJECT_ROOT,
+  'test_samples',
+  'seed_questions.jsonl',
+);
+const DEFAULT_OUT = path.join(
+  PROJECT_ROOT,
+  'gold',
+  'pipeline_gold.jsonl',
+);
+
+function parseArgs(argv) {
+  const args = argv.slice(2);
+  let limit;
+  let seedsPath;
+  let outPath;
+  let verbose = false;
 
   for (let i = 0; i < args.length; i++) {
     const a = args[i];
-
-    if (a === '--seeds' || a === '-s') {
-      out.seeds = args[++i];
-    } else if (a === '--out' || a === '-o') {
-      out.out = args[++i];
-    } else if (a === '--limit' || a === '-l') {
-      out.limit = Number(args[++i]);
-    } else if (a === '--provider' || a === '-p') {
-      out.provider = args[++i];
-    } else if (a === '--help' || a === '-h') {
-      out.help = true;
+    if (a === '--limit' || a === '-n') {
+      const v = Number(args[i + 1]);
+      if (!Number.isNaN(v)) limit = v;
+      i++;
+    } else if (a === '--seeds') {
+      seedsPath = args[i + 1];
+      i++;
+    } else if (a === '--out') {
+      outPath = args[i + 1];
+      i++;
+    } else if (a === '--verbose' || a === '-v') {
+      verbose = true;
     }
   }
 
-  return out;
-}
-
-function showHelp() {
-  console.log(`
-distill-pipeline — Full Distillation Cycle Runner
-
-Usage:
-  node pipeline_cli.js [options]
-
-Options:
-  --seeds <file>     Path to seed JSONL file (default: test_samples/seed_questions.jsonl)
-  --out <file>       Path to output JSONL      (default: gold/pipeline_gold.jsonl)
-  --limit <n>        Max number of seeds to process
-  --provider <name>  Provider override: ollama | openai | http
-  -h, --help         Show this help
+  // Also honour env var PIPELINE_VERBOSE=1
+  if (!verbose && process.env.PIPELINE_VERBOSE) {
+    const v = process.env.PIPELINE_VERBOSE.toLowerCase();
+    if (v === '1' || v === 'true' || v === 'yes') {
+      verbose = true;
+    }
+  }
 
-Examples:
-  node pipeline_cli.js --limit 25
-  node pipeline_cli.js -s custom_seeds.jsonl -o gold/round1.jsonl
-`);
+  return {
+    limit,
+    seedsPath: seedsPath || DEFAULT_SEEDS,
+    outPath: outPath || DEFAULT_OUT,
+    verbose,
+  };
 }
 
-//
-// ---- MAIN ----
-//
 async function main() {
-  const args = parseArgs();
-
-  if (args.help) {
-    showHelp();
-    process.exit(0);
-  }
-
-  // ---- Resolve paths ----
-  const seedsPath = args.seeds
-    ? path.resolve(args.seeds)
-    : path.resolve(__dirname, '../../test_samples/seed_questions.jsonl');
+  const { limit, seedsPath, outPath, verbose } = parseArgs(process.argv);
 
-  const outPath = args.out
-    ? path.resolve(args.out)
-    : path.resolve(__dirname, '../../gold/pipeline_gold.jsonl');
+  const generatorProvider = process.env.GENERATOR_PROVIDER || 'ollama';
+  const verifierProvider = process.env.VERIFIER_PROVIDER || generatorProvider;
+  const rewardProvider = process.env.REWARD_PROVIDER || generatorProvider;
 
-  const limit = args.limit ?? undefined;
+  const generatorModel =
+    process.env.GENERATOR_MODEL || process.env.OLLAMA_MODEL || 'qwen3-vl:8b-thinking';
+  const verifierModel =
+    process.env.VERIFIER_MODEL || generatorModel;
+  const rewardModel =
+    process.env.REWARD_MODEL || verifierModel;
 
-  // ---- Provider override ----
-  if (args.provider) {
-    process.env.PROVIDER_TYPE = args.provider;
-  }
-
-  // ---- Announce run ----
-  console.log(`\n🚀 Starting Distillation Pipeline`);
+  console.log('');
+  console.log('🚀 Starting Distillation Pipeline');
   console.log(`   Seeds:    ${seedsPath}`);
   console.log(`   Output:   ${outPath}`);
-  console.log(`   Provider: ${process.env.PROVIDER_TYPE || 'ollama (default)'}`);
-  console.log(`   Limit:    ${limit ?? 'none'}\n`);
-
-  // ---- Run batch ----
-  const result = await runPipelineBatch({
-    seedsPath,
-    outPath,
-    limit,
-  });
+  console.log(`   Providers:`);
+  console.log(
+    `     generator: ${generatorProvider} (${generatorModel})`,
+  );
+  console.log(
+    `     verifier:  ${verifierProvider} (${verifierModel})`,
+  );
+  console.log(
+    `     reward:    ${rewardProvider} (${rewardModel})`,
+  );
+  console.log(`   Limit:    ${limit ?? 'all'}`);
+  console.log(`   Verbose:  ${verbose ? 'yes' : 'no'}`);
+  console.log('');
+
+  try {
+    const result = await runPipelineBatch({
+      seedsPath,
+      outPath,
+      limit,
+      verbose,
+      logger: console,
+    });
+
+    console.log('');
+    console.log('🎉 Pipeline completed');
+    console.log(`   Total Seeds:    ${result.total}`);
+    console.log(`   Processed:      ${result.processed}`);
+    console.log(`   Accepted:       ${result.accepted}`);
+    console.log(`   Written to:     ${result.outPath}`);
+
+    if (result.statusCounts) {
+      console.log('');
+      console.log('   Status breakdown:');
+      for (const [status, count] of Object.entries(
+        result.statusCounts,
+      )) {
+        console.log(`     ${status.padEnd(17)} ${count}`);
+      }
+    }
 
-  console.log(`\n🎉 Pipeline completed`);
-  console.log(`   Total Seeds:    ${result.total}`);
-  console.log(`   Processed:      ${result.processed}`);
-  console.log(`   Accepted:       ${result.accepted}`);
-  console.log(`   Written to:     ${result.outPath}\n`);
+    console.log('');
+  } catch (err) {
+    console.error('❌ Pipeline error:', err);
+    process.exit(1);
+  }
 }
 
 main().catch((err) => {
-  console.error(`❌ Pipeline error:`, err);
+  console.error('❌ Fatal error:', err);
   process.exit(1);
 });