hokusai/bulkgen/providers/text.py
Konstantin Fickel 15fdff7763
feat: add Mistral text generation provider
- Uses native async API (chat.complete_async)
- Appends text input file contents to prompt with headers
- Notes image inputs as attached references
- Writes raw LLM response directly to output file
2026-02-13 20:12:45 +01:00

63 lines
1.9 KiB
Python

"""Mistral text generation provider."""
from __future__ import annotations
from pathlib import Path
from typing import override
from mistralai import Mistral, models
from bulkgen.config import IMAGE_EXTENSIONS, TargetConfig
from bulkgen.providers import Provider
class TextProvider(Provider):
"""Generates text via the Mistral API."""
def __init__(self, api_key: str) -> None:
self._api_key = api_key
@override
async def generate(
self,
target_name: str,
target_config: TargetConfig,
resolved_prompt: str,
resolved_model: str,
project_dir: Path,
) -> None:
output_path = project_dir / target_name
content_parts: list[str] = [resolved_prompt]
for input_name in target_config.inputs:
input_path = project_dir / input_name
suffix = input_path.suffix.lower()
if suffix in IMAGE_EXTENSIONS:
content_parts.append(f"\n[Attached image: {input_name}]")
else:
file_content = input_path.read_text()
content_parts.append(
f"\n--- Contents of {input_name} ---\n{file_content}"
)
full_prompt = "\n".join(content_parts)
async with Mistral(api_key=self._api_key) as client:
response = await client.chat.complete_async(
model=resolved_model,
messages=[models.UserMessage(content=full_prompt)],
)
if response is None or not response.choices:
msg = f"Mistral API returned no response for target '{target_name}'"
raise RuntimeError(msg)
content = response.choices[0].message.content
if content is None:
msg = f"Mistral API returned empty content for target '{target_name}'"
raise RuntimeError(msg)
text = content if isinstance(content, str) else str(content)
output_path.write_text(text)