generated from john/python-template
Step 3 partially implemented.
This commit is contained in:
Jim Lancaster
@@ -0,0 +1,10 @@
|
||||
You are an assistant that may call tools.
|
||||
|
||||
Tool safety rules:
|
||||
1) Tool arguments MUST be strict JSON matching the schema exactly.
|
||||
2) Never place disallowed, sensitive, explicit, or policy-violating text directly into tool arguments.
|
||||
3) If user content may be unsafe, first produce a brief neutral summary and pass only that summary.
|
||||
4) Prefer IDs, enums, booleans, and short fields over raw free-form text.
|
||||
5) Keep all string arguments <= 300 chars unless schema says otherwise.
|
||||
6) If you cannot safely provide valid tool args, do not call the tool; respond with "NO_TOOL_CALL" and explain briefly.
|
||||
7) Never include markdown/code fences in tool arguments.
|
||||
@@ -0,0 +1,31 @@
|
||||
"""Provider exports and factory for transcription adapters."""
|
||||
|
||||
from transcription.config import Provider, Settings, get_settings
|
||||
from transcription.providers.base import (
|
||||
ProviderAuthError,
|
||||
ProviderError,
|
||||
ProviderResponseError,
|
||||
TranscriptionProvider,
|
||||
TranscriptionResult,
|
||||
)
|
||||
from transcription.providers.openrouter import OpenRouterTranscriptionProvider
|
||||
|
||||
|
||||
def get_transcription_provider(*, settings: Settings | None = None) -> TranscriptionProvider:
|
||||
"""Return the configured transcription provider adapter."""
|
||||
runtime_settings = settings or get_settings()
|
||||
if runtime_settings.provider == Provider.OPENROUTER:
|
||||
return OpenRouterTranscriptionProvider(settings=runtime_settings)
|
||||
|
||||
raise ProviderError(f"Unsupported transcription provider: {runtime_settings.provider}")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ProviderAuthError",
|
||||
"ProviderError",
|
||||
"ProviderResponseError",
|
||||
"TranscriptionProvider",
|
||||
"TranscriptionResult",
|
||||
"OpenRouterTranscriptionProvider",
|
||||
"get_transcription_provider",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Provider interfaces and shared types for transcription adapters."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Protocol
|
||||
|
||||
|
||||
class ProviderError(RuntimeError):
|
||||
"""Base error for provider failures."""
|
||||
|
||||
|
||||
class ProviderAuthError(ProviderError):
|
||||
"""Raised when provider authentication fails."""
|
||||
|
||||
|
||||
class ProviderResponseError(ProviderError):
|
||||
"""Raised when provider responses are malformed or unusable."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TranscriptionResult:
|
||||
"""Normalized output returned by any transcription provider."""
|
||||
|
||||
text: str
|
||||
provider: str
|
||||
model: str
|
||||
|
||||
|
||||
class TranscriptionProvider(Protocol):
|
||||
"""Contract every transcription provider adapter must satisfy."""
|
||||
|
||||
def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
|
||||
"""Transcribe the provided image according to the prompt text."""
|
||||
@@ -0,0 +1,131 @@
|
||||
"""OpenRouter transcription provider adapter."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from openrouter import OpenRouter
|
||||
|
||||
from transcription.config import Settings, get_settings
|
||||
from transcription.providers.base import (
|
||||
ProviderAuthError,
|
||||
ProviderError,
|
||||
ProviderResponseError,
|
||||
TranscriptionResult,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_OPENROUTER_MODEL = "google/gemini-2.5-flash"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OpenRouterRequest:
|
||||
"""Normalized request payload fields for OpenRouter calls."""
|
||||
|
||||
model: str
|
||||
messages: list[dict[str, Any]]
|
||||
http_referer: str | None
|
||||
x_open_router_title: str | None
|
||||
|
||||
|
||||
class OpenRouterTranscriptionProvider:
|
||||
"""Adapter that performs image transcription through OpenRouter."""
|
||||
|
||||
def __init__(self, *, settings: Settings | None = None, client: OpenRouter | None = None):
|
||||
self._settings = settings or get_settings()
|
||||
self._model = self._settings.provider_model or DEFAULT_OPENROUTER_MODEL
|
||||
self._client = client or OpenRouter(api_key=self._settings.openrouter_api_key)
|
||||
|
||||
@property
|
||||
def model(self) -> str:
|
||||
"""Return the resolved OpenRouter model slug."""
|
||||
return self._model
|
||||
|
||||
def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
|
||||
"""Send prompt + image to OpenRouter and return normalized text output."""
|
||||
request = self._build_request(prompt_text=prompt_text, image_bytes=image_bytes, mime_type=mime_type)
|
||||
try:
|
||||
response = self._client.chat.send(
|
||||
messages=request.messages,
|
||||
model=request.model,
|
||||
http_referer=request.http_referer,
|
||||
x_open_router_title=request.x_open_router_title,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
message = str(exc).lower()
|
||||
if "401" in message or "auth" in message or "api key" in message:
|
||||
raise ProviderAuthError("OpenRouter authentication failed") from exc
|
||||
raise ProviderError("OpenRouter request failed") from exc
|
||||
|
||||
text = self._extract_text(response)
|
||||
model = self._get_optional_attr(response, "model") or self.model
|
||||
logger.info("OpenRouter transcription completed using model=%s", model)
|
||||
return TranscriptionResult(text=text, provider="openrouter", model=model)
|
||||
|
||||
def _build_request(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> OpenRouterRequest:
|
||||
image_b64 = base64.b64encode(image_bytes).decode("ascii")
|
||||
data_url = f"data:{mime_type};base64,{image_b64}"
|
||||
|
||||
messages: list[dict[str, Any]] = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": prompt_text},
|
||||
{"type": "image_url", "image_url": {"url": data_url}},
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
return OpenRouterRequest(
|
||||
model=self.model,
|
||||
messages=messages,
|
||||
http_referer=self._settings.openrouter_http_referer,
|
||||
x_open_router_title=self._settings.openrouter_app_title,
|
||||
)
|
||||
|
||||
def _extract_text(self, response: Any) -> str:
|
||||
choices = self._get_optional_attr(response, "choices")
|
||||
if not choices:
|
||||
raise ProviderResponseError("OpenRouter response missing choices")
|
||||
|
||||
first_choice = choices[0]
|
||||
message = self._get_optional_attr(first_choice, "message")
|
||||
if message is None:
|
||||
raise ProviderResponseError("OpenRouter response missing assistant message")
|
||||
|
||||
content = self._get_optional_attr(message, "content")
|
||||
text = self._normalize_content(content)
|
||||
if not text:
|
||||
raise ProviderResponseError("OpenRouter response contained no transcription text")
|
||||
return text
|
||||
|
||||
def _normalize_content(self, content: Any) -> str:
|
||||
if isinstance(content, str):
|
||||
return content.strip()
|
||||
|
||||
if isinstance(content, list):
|
||||
parts: list[str] = []
|
||||
for item in content:
|
||||
text_part = None
|
||||
if isinstance(item, dict):
|
||||
text_part = item.get("text")
|
||||
else:
|
||||
text_part = self._get_optional_attr(item, "text")
|
||||
|
||||
if isinstance(text_part, str) and text_part.strip():
|
||||
parts.append(text_part.strip())
|
||||
return "\n".join(parts).strip()
|
||||
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def _get_optional_attr(obj: Any, key: str) -> Any:
|
||||
if obj is None:
|
||||
return None
|
||||
if isinstance(obj, dict):
|
||||
return obj.get(key)
|
||||
return getattr(obj, key, None)
|
||||
@@ -0,0 +1,19 @@
|
||||
"""Service layer exports."""
|
||||
|
||||
from transcription.services.transcription import (
|
||||
DEFAULT_PROMPT_FILE,
|
||||
PromptLoadError,
|
||||
TranscriptionError,
|
||||
load_image_payload,
|
||||
load_prompt_text,
|
||||
transcribe_document_image,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"DEFAULT_PROMPT_FILE",
|
||||
"PromptLoadError",
|
||||
"TranscriptionError",
|
||||
"load_image_payload",
|
||||
"load_prompt_text",
|
||||
"transcribe_document_image",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Prompt loading and provider-backed transcription service."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import mimetypes
|
||||
from pathlib import Path
|
||||
|
||||
from transcription.config import Settings, get_settings
|
||||
from transcription.providers import ProviderError, TranscriptionProvider, TranscriptionResult, get_transcription_provider
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_PROMPT_FILE = "transcribe_document.md"
|
||||
SUPPORTED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
|
||||
|
||||
|
||||
class PromptLoadError(RuntimeError):
|
||||
"""Raised when prompt artifacts cannot be loaded safely."""
|
||||
|
||||
|
||||
class TranscriptionError(RuntimeError):
|
||||
"""Raised when transcription execution fails."""
|
||||
|
||||
|
||||
def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str:
|
||||
"""Load and validate prompt text from PROMPT_DIR."""
|
||||
runtime_settings = settings or get_settings()
|
||||
prompt_path = runtime_settings.prompt_dir / prompt_name
|
||||
|
||||
if not prompt_path.exists() or not prompt_path.is_file():
|
||||
raise PromptLoadError(f"Prompt file not found: {prompt_path}")
|
||||
|
||||
prompt_text = prompt_path.read_text(encoding="utf-8").strip()
|
||||
if not prompt_text:
|
||||
raise PromptLoadError(f"Prompt file is empty: {prompt_path}")
|
||||
|
||||
logger.info("Loaded prompt artifact: %s", prompt_path)
|
||||
return prompt_text
|
||||
|
||||
|
||||
def load_image_payload(image_path: str | Path) -> tuple[bytes, str]:
|
||||
"""Read image bytes and detect mime type for supported uploads."""
|
||||
path = Path(image_path)
|
||||
|
||||
if not path.exists() or not path.is_file():
|
||||
raise TranscriptionError(f"Image file not found: {path}")
|
||||
|
||||
suffix = path.suffix.lower()
|
||||
if suffix not in SUPPORTED_EXTENSIONS:
|
||||
raise TranscriptionError(f"Unsupported file type: {suffix}")
|
||||
|
||||
mime_type, _ = mimetypes.guess_type(path.name)
|
||||
if suffix in {".tif", ".tiff"}:
|
||||
mime_type = "image/tiff"
|
||||
if not mime_type:
|
||||
raise TranscriptionError(f"Unable to determine MIME type for: {path}")
|
||||
|
||||
return path.read_bytes(), mime_type
|
||||
|
||||
|
||||
def transcribe_document_image(
|
||||
image_path: str | Path,
|
||||
*,
|
||||
prompt_name: str = DEFAULT_PROMPT_FILE,
|
||||
settings: Settings | None = None,
|
||||
provider: TranscriptionProvider | None = None,
|
||||
) -> TranscriptionResult:
|
||||
"""Transcribe a local image using the configured prompt and provider."""
|
||||
runtime_settings = settings or get_settings()
|
||||
prompt_text = load_prompt_text(prompt_name=prompt_name, settings=runtime_settings)
|
||||
image_bytes, mime_type = load_image_payload(image_path)
|
||||
|
||||
adapter = provider or get_transcription_provider(settings=runtime_settings)
|
||||
logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
|
||||
|
||||
try:
|
||||
result = adapter.transcribe(
|
||||
prompt_text=prompt_text,
|
||||
image_bytes=image_bytes,
|
||||
mime_type=mime_type,
|
||||
)
|
||||
except ProviderError as exc:
|
||||
raise TranscriptionError("Provider transcription failed") from exc
|
||||
|
||||
logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
|
||||
return result
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Tests for transcription.providers.openrouter."""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from transcription.config import Settings
|
||||
from transcription.providers.base import ProviderError, ProviderResponseError
|
||||
from transcription.providers.openrouter import DEFAULT_OPENROUTER_MODEL, OpenRouterTranscriptionProvider
|
||||
|
||||
|
||||
class _FakeChat:
|
||||
def __init__(self, response=None, error: Exception | None = None):
|
||||
self._response = response
|
||||
self._error = error
|
||||
self.calls = []
|
||||
|
||||
def send(self, **kwargs):
|
||||
self.calls.append(kwargs)
|
||||
if self._error:
|
||||
raise self._error
|
||||
return self._response
|
||||
|
||||
|
||||
class _FakeClient:
|
||||
def __init__(self, response=None, error: Exception | None = None):
|
||||
self.chat = _FakeChat(response=response, error=error)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOpenRouterProviderInit:
|
||||
"""Verify OpenRouter provider initialization behavior."""
|
||||
|
||||
def test_model_falls_back_to_default_when_unset(self):
|
||||
"""Provider uses adapter default model when provider_model is None."""
|
||||
settings = Settings(openrouter_api_key="test-key", provider_model=None)
|
||||
provider = OpenRouterTranscriptionProvider(settings=settings, client=_FakeClient())
|
||||
assert provider.model == DEFAULT_OPENROUTER_MODEL
|
||||
|
||||
def test_model_uses_configured_value(self):
|
||||
"""Provider uses configured provider_model when present."""
|
||||
settings = Settings(openrouter_api_key="test-key", provider_model="vendor/custom-model")
|
||||
provider = OpenRouterTranscriptionProvider(settings=settings, client=_FakeClient())
|
||||
assert provider.model == "vendor/custom-model"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOpenRouterProviderTranscribe:
|
||||
"""Verify OpenRouter request construction and response parsing."""
|
||||
|
||||
def test_includes_optional_referer_and_title_when_set(self):
|
||||
"""Transcribe sends app attribution fields when configured."""
|
||||
response = {"model": "vendor/model-a", "choices": [{"message": {"content": "Transcript text"}}]}
|
||||
client = _FakeClient(response=response)
|
||||
settings = Settings(
|
||||
openrouter_api_key="test-key",
|
||||
openrouter_http_referer="https://example.test",
|
||||
openrouter_app_title="Transcription App",
|
||||
)
|
||||
provider = OpenRouterTranscriptionProvider(settings=settings, client=client)
|
||||
|
||||
result = provider.transcribe(
|
||||
prompt_text="Prompt body",
|
||||
image_bytes=b"img-bytes",
|
||||
mime_type="image/png",
|
||||
)
|
||||
|
||||
send_call = client.chat.calls[0]
|
||||
assert send_call["http_referer"] == "https://example.test"
|
||||
assert send_call["x_open_router_title"] == "Transcription App"
|
||||
assert result.text == "Transcript text"
|
||||
|
||||
def test_parses_successful_response_text(self):
|
||||
"""Transcribe returns normalized text from a valid response payload."""
|
||||
response = {
|
||||
"model": "vendor/model-b",
|
||||
"choices": [{"message": {"content": [{"text": "Line 1"}, {"text": "Line 2"}]}}],
|
||||
}
|
||||
provider = OpenRouterTranscriptionProvider(
|
||||
settings=Settings(openrouter_api_key="test-key"),
|
||||
client=_FakeClient(response=response),
|
||||
)
|
||||
|
||||
result = provider.transcribe(
|
||||
prompt_text="Prompt body",
|
||||
image_bytes=b"img-bytes",
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
|
||||
assert result.text == "Line 1\nLine 2"
|
||||
assert result.provider == "openrouter"
|
||||
assert result.model == "vendor/model-b"
|
||||
|
||||
def test_maps_sdk_exception_to_provider_error(self):
|
||||
"""Transcribe converts SDK failures to ProviderError."""
|
||||
provider = OpenRouterTranscriptionProvider(
|
||||
settings=Settings(openrouter_api_key="test-key"),
|
||||
client=_FakeClient(error=RuntimeError("network down")),
|
||||
)
|
||||
|
||||
with pytest.raises(ProviderError):
|
||||
provider.transcribe(
|
||||
prompt_text="Prompt body",
|
||||
image_bytes=b"img-bytes",
|
||||
mime_type="image/png",
|
||||
)
|
||||
|
||||
def test_raises_on_empty_or_invalid_response(self):
|
||||
"""Transcribe raises ProviderResponseError for missing completion text."""
|
||||
provider = OpenRouterTranscriptionProvider(
|
||||
settings=Settings(openrouter_api_key="test-key"),
|
||||
client=_FakeClient(response=SimpleNamespace(choices=[])),
|
||||
)
|
||||
|
||||
with pytest.raises(ProviderResponseError):
|
||||
provider.transcribe(
|
||||
prompt_text="Prompt body",
|
||||
image_bytes=b"img-bytes",
|
||||
mime_type="image/png",
|
||||
)
|
||||
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"name": "search_docs",
|
||||
"description": "Search internal docs with a short safe query.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"properties": {
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "Short neutral search phrase",
|
||||
"minLength": 1,
|
||||
"maxLength": 180
|
||||
},
|
||||
"scope": {
|
||||
"type": "string",
|
||||
"enum": ["public", "internal"]
|
||||
},
|
||||
"top_k": {
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 10
|
||||
}
|
||||
},
|
||||
"required": ["query", "scope"]
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user