Multimodal ingest & processing

public record PageExtract(int pageNum, String text, String tier, Map<String, Object> meta) {}

public class TieredPdfExtractor {
  static final int MIN_NATIVE_CHARS = 80;
  static final double OCR_CONF_THRESHOLD = 0.72;

  public List<PageExtract> extract(Path pdfPath, Optional<LlamaParseClient> llama) throws IOException {
    try (PDDocument doc = Loader.loadPDF(pdfPath.toFile())) {
      var out = new ArrayList<PageExtract>();
      for (int i = 0; i < doc.getNumberOfPages(); i++) {
        out.add(extractPage(doc, i + 1, llama));
      }
      return out;
    }
  }

  PageExtract extractPage(PDDocument doc, int pageNum, Optional<LlamaParseClient> llama) throws IOException {
    String nativeText = new PDFTextStripper() {{
      setStartPage(pageNum);
      setEndPage(pageNum);
    }}.getText(doc).trim();

    if (nativeText.length() >= MIN_NATIVE_CHARS) {
      return new PageExtract(pageNum, nativeText, "native", Map.of("char_count", nativeText.length()));
    }

    OcrResult ocr = tesseractOcr(doc, pageNum);
    if (ocr.confidence() >= OCR_CONF_THRESHOLD && ocr.text().length() >= MIN_NATIVE_CHARS) {
      return new PageExtract(pageNum, ocr.text(), "ocr", Map.of("ocr_confidence", ocr.confidence()));
    }

    if (llama.isPresent()) {
      String parsed = llama.get().parsePage(pdfPath, pageNum);
      if (parsed != null && parsed.length() >= MIN_NATIVE_CHARS) {
        return new PageExtract(pageNum, parsed, "llamaparse", Map.of("ocr_confidence", ocr.confidence()));
      }
    }

    return new PageExtract(pageNum, ocr.text(), "ocr_low_confidence",
        Map.of("ocr_confidence", ocr.confidence(), "needs_vision", true));
  }
}

// Tabula via tabula-java CLI wrapper or PDFBox table heuristics
public List<TableChunk> extractTables(Path pdf) throws IOException {
  var chunks = new ArrayList<TableChunk>();
  try (PDDocument doc = Loader.loadPDF(pdf.toFile())) {
    var extractor = new SpreadsheetExtractionAlgorithm();
    int pageNum = 1;
    for (PDPage page : doc.getPages()) {
      var tables = extractor.extract(page);
      int idx = 0;
      for (Table table : tables) {
        String markdown = tableToMarkdown(table);
        chunks.add(new TableChunk(markdown, pageNum, idx++, tableToCsv(table)));
      }
      pageNum++;
    }
  }
  return chunks;
}

import anthropic, base64

client = anthropic.Anthropic()

def describe_image(image_bytes: bytes, mime: str = "image/png") -> str:
    msg = client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": mime, "data": base64.standard_b64encode(image_bytes).decode()}},
                {"type": "text", "text": "Describe this figure for search retrieval. 2-3 sentences."},
            ],
        }],
    )
    return msg.content[0].text

import boto3, base64, json

bedrock = boto3.client("bedrock-runtime")

def describe_image(image_bytes: bytes) -> str:
    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 200,
        "messages": [{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": base64.b64encode(image_bytes).decode()}},
                {"type": "text", "text": "Describe this figure for search retrieval."},
            ],
        }],
    }
    resp = bedrock.invoke_model(modelId="anthropic.claude-3-5-haiku-20241022", body=json.dumps(body))
    return json.loads(resp["body"].read())["content"][0]["text"]

public String describeImage(byte[] pngBytes) {
  Media media = new Media(MimeTypeUtils.IMAGE_PNG, pngBytes);
  UserMessage msg = UserMessage.builder()
      .text("Describe this figure for search retrieval. 2-3 sentences.")
      .media(media)
      .build();
  return chatClient.prompt(new Prompt(msg))
      .options(OpenAiChatOptions.builder().withModel("gpt-4o-mini").withMaxTokens(200).build())
      .call()
      .chatResponse()
      .getResult().getOutput().getContent();
}

public String describeImage(byte[] pngBytes) {
  var model = AnthropicChatModel.builder()
      .modelName("claude-3-5-haiku-20241022")
      .maxTokens(200)
      .build();
  Media media = new Media(MimeTypeUtils.IMAGE_PNG, pngBytes);
  return model.call(new Prompt(UserMessage.builder()
      .text("Describe this figure for search retrieval.")
      .media(media)
      .build())).getResult().getOutput().getContent();
}

public String describeImage(byte[] pngBytes) {
  var chat = new BedrockAnthropicChatModel(
      BedrockAnthropicChatOptions.builder()
          .withModel("anthropic.claude-3-5-haiku-20241022")
          .withMaxTokens(200)
          .build());
  Media media = new Media(MimeTypeUtils.IMAGE_PNG, pngBytes);
  return chat.call(new Prompt(UserMessage.builder()
      .text("Describe this figure for search retrieval.")
      .media(media)
      .build())).getResult().getOutput().getContent();
}

public record DocClassification(String docType, double confidence, String rationale) {}

public class DocumentRouter {
  private final ChatClient chatClient;

  public IngestJob route(String docId, String textSample) {
    DocClassification c = classify(textSample.substring(0, Math.min(8000, textSample.length())));
    if (c.confidence() < 0.75) {
      return humanReviewQueue.enqueue(docId, c);
    }
    return switch (c.docType()) {
      case "invoice" -> invoicePipeline.submit(docId, c);
      case "contract" -> contractPipeline.submit(docId, c);
      default -> ragPipeline.submit(docId, c);
    };
  }

  DocClassification classify(String sample) {
    // Spring AI structured output → DocClassification record
    return chatClient.prompt()
        .system("Classify for ingest routing: invoice | contract | general | unknown")
        .user(sample)
        .call()
        .entity(DocClassification.class);
  }
}

import anthropic, base64

client = anthropic.Anthropic()

def vision_transcribe_page(png_bytes: bytes) -> tuple[str, int]:
    msg = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": base64.standard_b64encode(png_bytes).decode()}},
                {"type": "text", "text": "Transcribe this page to markdown. Preserve tables."},
            ],
        }],
    )
    return msg.content[0].text, msg.usage.input_tokens

import boto3, base64, json

bedrock = boto3.client("bedrock-runtime")

def vision_transcribe_page(png_bytes: bytes) -> tuple[str, int]:
    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 4096,
        "messages": [{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": base64.b64encode(png_bytes).decode()}},
                {"type": "text", "text": "Transcribe this page to markdown."},
            ],
        }],
    }
    resp = bedrock.invoke_model(
        modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
        body=json.dumps(body),
    )
    payload = json.loads(resp["body"].read())
    return payload["content"][0]["text"], payload["usage"]["input_tokens"]

public VisionPageResult transcribePage(byte[] pngBytes) {
  Media media = new Media(MimeTypeUtils.IMAGE_PNG, pngBytes);
  UserMessage msg = UserMessage.builder()
      .text("Transcribe this document page to markdown. Preserve tables.")
      .media(media)
      .build();
  ChatResponse resp = chatClient.prompt(new Prompt(msg))
      .options(OpenAiChatOptions.builder().withModel("gpt-4o").withMaxTokens(4096).build())
      .call()
      .chatResponse();
  int promptTokens = resp.getMetadata().getUsage().getPromptTokens();
  return new VisionPageResult(resp.getResult().getOutput().getContent(), promptTokens);
}

public VisionPageResult transcribePage(byte[] pngBytes) {
  var model = AnthropicChatModel.builder()
      .modelName("claude-3-5-sonnet-20241022")
      .maxTokens(4096)
      .build();
  Media media = new Media(MimeTypeUtils.IMAGE_PNG, pngBytes);
  ChatResponse resp = model.call(new Prompt(UserMessage.builder()
      .text("Transcribe this page to markdown.")
      .media(media)
      .build()));
  return new VisionPageResult(
      resp.getResult().getOutput().getContent(),
      resp.getMetadata().getUsage().getPromptTokens());
}

public VisionPageResult transcribePage(byte[] pngBytes) {
  var chat = new BedrockAnthropicChatModel(
      BedrockAnthropicChatOptions.builder()
          .withModel("anthropic.claude-3-5-sonnet-20241022-v2:0")
          .withMaxTokens(4096)
          .build());
  Media media = new Media(MimeTypeUtils.IMAGE_PNG, pngBytes);
  ChatResponse resp = chat.call(new Prompt(UserMessage.builder()
      .text("Transcribe this page to markdown.")
      .media(media)
      .build()));
  return new VisionPageResult(
      resp.getResult().getOutput().getContent(),
      resp.getMetadata().getUsage().getPromptTokens());
}

public List<AudioChunk> transcribe(Path audioFile) throws IOException {
  var request = TranscriptionRequest.builder()
      .model("whisper-1")
      .responseFormat(TranscriptionResponseFormat.VERBOSE_JSON)
      .build();
  TranscriptionResult result = openAiAudio.transcribe(audioFile, request);
  return result.segments().stream()
      .map(s -> new AudioChunk(s.text(), s.startSec(), s.endSec(), null))
      .toList();
}

public List<AudioChunk> mergeDiarization(List<AudioChunk> chunks, List<DiarSegment> diar) {
  for (AudioChunk c : chunks) {
    double mid = (c.startSec() + c.endSec()) / 2.0;
    diar.stream()
        .filter(d -> d.start() <= mid && mid <= d.end())
        .findFirst()
        .ifPresent(d -> c.setSpeaker(d.speaker()));
  }
  return chunks;
}

public SyncResult crawlAndSync(String url, CrawlStateDb state) {
  if (!robotsTxt.allowed(url, "SharpbyteBot/1.0")) {
    return SyncResult.BLOCKED_BY_ROBOTS;
  }
  FirecrawlResponse page = firecrawlClient.scrape(url);
  String hash = contentHash(normalize(page.markdown()));
  Optional<String> prev = state.getHash(url);
  if (prev.isPresent() && prev.get().equals(hash)) {
    return SyncResult.UNCHANGED;
  }
  var chunks = markdownChunker.chunk(page.markdown(), url);
  vectorStore.upsert(chunks);
  state.save(url, hash);
  return SyncResult.UPDATED;
}

String contentHash(String normalized) {
  return HexFormat.of().formatHex(
      MessageDigest.getInstance("SHA-256").digest(normalized.getBytes(StandardCharsets.UTF_8)));
}