| from __future__ import annotations |
|
|
| from abc import ABC, abstractmethod |
| from typing import TYPE_CHECKING |
|
|
| from densefeed.llm_types import ( |
| EpisodeOutline, |
| MovementPlan, |
| RankedItem, |
| ScoredItem, |
| Verdict, |
| ) |
|
|
| if TYPE_CHECKING: |
| from densefeed.config import DenseFeedConfig |
| from densefeed.models import ScriptLine |
|
|
|
|
| class InferenceBackend(ABC): |
| @abstractmethod |
| async def filter_items(self, items: list[dict]) -> list[Verdict]: |
| """Binary KEEP/DROP filter.""" |
| pass |
|
|
| @abstractmethod |
| async def score_items(self, items: list[dict]) -> list[ScoredItem]: |
| """Graded 0-10 scoring.""" |
| pass |
|
|
| @abstractmethod |
| async def generate_outline(self, date: str, stories: list[dict]) -> EpisodeOutline: |
| """Generate a structured episode outline.""" |
| pass |
|
|
| @abstractmethod |
| async def generate_script_segment( |
| self, |
| date: str, |
| story_title: str, |
| content_chunk: str, |
| previous_context: str | None, |
| segment_type: str, |
| outline_context: str | None, |
| ) -> list[ScriptLine]: |
| """Generate a script segment.""" |
| pass |
|
|
| async def filter_score_items(self, items: list[dict]) -> list[dict]: |
| """Collapsed filter+score in a single call. |
| |
| Default implementation delegates to filter_items then score_items. |
| Backends with a native collapsed call (e.g. ZeroGPU) should override. |
| """ |
| verdicts = await self.filter_items(items) |
| keep_ids = {v.i for v in verdicts if v.v == "KEEP"} |
| kept = [item for item in items if item.get("i") in keep_ids] |
| scored = await self.score_items(kept) |
| return [{"i": s.i, "score": s.s, "reason": s.r} for s in scored] |
|
|
|
|
| def create_backend(config: DenseFeedConfig) -> InferenceBackend: |
| backend_type = config.llm.backend.lower() |
| if backend_type == "job": |
| from .job_backend import JobBackend |
|
|
| return JobBackend(config) |
| if backend_type == "baml": |
| from .baml_backend import BAMLBackend |
|
|
| return BAMLBackend(config) |
| if backend_type == "zerogpu": |
| from .zerogpu_backend import ZeroGPUBackend |
|
|
| return ZeroGPUBackend(config) |
| raise ValueError(f"Unknown LLM backend: {config.llm.backend}") |
|
|
