babylon.ai.llm_provider
LLM Provider strategy pattern for text generation.
This module provides the “Mouth” of the AI Observer - the interface through which the NarrativeDirector speaks. It follows the same Protocol pattern as SimulationObserver for loose coupling.
Components: - LLMProvider: Protocol defining the text generation interface - MockLLM: Deterministic mock for testing - DeepSeekClient: Production client using DeepSeek API
SYNC API: All providers implement synchronous generate() to match the SimulationObserver pattern. Async implementations wrap internally.
Classes
|
DeepSeek LLM client using OpenAI-compatible API. |
|
Protocol for LLM text generation providers. |
|
Deterministic mock LLM for testing. |
- class babylon.ai.llm_provider.LLMProvider(*args, **kwargs)[source]
Bases:
ProtocolProtocol for LLM text generation providers.
Follows the same pattern as SimulationObserver - loose coupling via Protocol enables easy testing and provider swapping.
SYNC API: All implementations use synchronous interfaces to avoid event loop conflicts with other asyncio.run() callers (e.g., RAG).
- generate(prompt, system_prompt=None, temperature=0.7)[source]
Generate text from prompt (synchronous).
- __init__(*args, **kwargs)
- class babylon.ai.llm_provider.MockLLM(responses=None, default_response='Mock LLM response')[source]
Bases:
objectDeterministic mock LLM for testing.
Returns pre-configured responses in queue order, or a fixed default response. Synchronous API.
This is the primary testing tool for NarrativeDirector - it allows tests to verify behavior without network calls.
- property call_history: list[dict[str, Any]]
History of all calls with arguments.
Returns a copy to prevent external modification.
- class babylon.ai.llm_provider.DeepSeekClient(config=None)[source]
Bases:
objectDeepSeek LLM client using OpenAI-compatible API.
Primary LLM provider for Babylon narrative generation. Uses the openai Python package with custom base_url.
SYNC API: Uses synchronous OpenAI client to avoid event loop conflicts with RAG queries that use asyncio.run().