vibe.llm_providers.system_proxy_provider

SystemProxyProvider: Adapter that proxies to the core VIBE system before delegating to LLM.

In assistant mode, the LLM needs a system prompt. But that prompt is a Jinja2 template that may reference interview variables (e.g., {% if user_type == 'business' %}). Before we can render the prompt, we must ask questions to resolve those variables.

This provider runs the VIBE probing mechanism iteratively until the prompt template is renderable, then delegates to the real LLM.

Architecture

Question Sources (VIBE Probing + LLM Endpoints) ↓ SystemProxyProvider (adapter) ↓ StreamChunk objects (ask_question tool calls or LLM responses) ↓ StreamTranslator.translate() ↓ SSE Events

This unifies system questions and LLM questions into a single pipeline, eliminating special-case code paths in AssistantService.

SystemProxyProvider

Proxy that routes to VIBE system for required variables before delegating to real LLM.

Flow on each stream_completion() call: 1. Run probe_callback() to check what variables are needed for prompt template 2. If questions returned: emit them as ask_question tool calls (via StreamChunks) 3. If no questions returned: prompt template is renderable, delegate to real_provider

This makes system questions indistinguishable from LLM questions to StreamTranslator.

Example

Template: "You are helping with {% if user_type %}{{ specific_context }}{% endif %}"

Turn 1: probe returns ['user_type'] → emit ask_question(mode='predefined', question_id='user_type') Turn 2: probe returns ['specific_context'] → emit ask_question( mode='predefined', question_id='specific_context') Turn 3: probe returns [] → delegate to real LLM

config 

config: dict[str, Any]

Proxy config from real provider to ensure correct settings are used.

provider_config 

provider_config: object

Proxy provider_config from real provider.

__init__ 

__init__(real_provider: LLMProvider, get_pending_questions: Callable[[], list[str]], question_definitions: dict[str, dict[str, Any]]) -> None

Initialize the adapter.

Parameters:
  • real_provider (LLMProvider) –

    The actual LLM provider to delegate to

  • get_pending_questions (Callable[[], list[str]]) –

    Function that returns list of pending system question IDs from session

  • question_definitions (dict[str, dict[str, Any]]) –

    Question definitions (for answer_type metadata)

get_capabilities 

get_capabilities() -> ProviderCapabilities

Inherit capabilities from real provider.

get_effective_tools_enabled 

get_effective_tools_enabled() -> bool

Delegate to real provider's resolved tools setting.

convert_messages_to_provider_format 

convert_messages_to_provider_format(messages: list[Message], tools: list[Tool] | None = None) -> object

Delegate message conversion to real provider.

get_usage_stats 

get_usage_stats() -> UsageStats | None

Delegate usage stats retrieval to real provider.

get_last_response_id 

get_last_response_id() -> str | None

Delegate response ID retrieval to real provider.

stream_generate 

stream_generate(messages: list[Message], sequence_number: int, session_id: str, assistant_name: str, endpoint_name: str, turn_id: str, previous_response_id: str | None = None, tool_outputs: list[ToolOutput] | None = None, unanswered_predefined_questions: list[dict[str, Any]] | None = None) -> Generator[StreamChunk, None, None]

Check if there are pending system questions. If yes, emit them. If no, delegate to real LLM.