Fix Runtime

from risicare.runtime import FixRuntime, FixRuntimeConfig
 
runtime = FixRuntime(
    config=FixRuntimeConfig(
        api_endpoint="https://app.risicare.ai",
        api_key="rsk-...",
        project_id="proj-...",
        enabled=True,
    )
)
 
runtime.start()

import risicare
from risicare import init_runtime, get_runtime, shutdown_runtime
from risicare.runtime import FixRuntimeConfig
 
# 1. Initialize SDK (tracing + auto-instrumentation)
risicare.init(api_key="rsk-...", environment="production")
 
# 2. Configure fix runtime
config = FixRuntimeConfig(
    api_endpoint="https://app.risicare.ai",
    api_key="rsk-...",
    cache_ttl_seconds=300,        # Cache fixes locally for 5 minutes
    refresh_interval_seconds=60,  # Poll API for new fixes every minute
    dry_run=False,                # Set True to log fixes without applying
    ab_testing_enabled=True,      # Enable A/B traffic splitting
)
 
# 3. Initialize and auto-start (fetches active fixes from API)
runtime = init_runtime(config=config, auto_start=True)
 
# 4. Use anywhere in your application
runtime = get_runtime()
fix = runtime.get_fix("TOOL.EXECUTION.TIMEOUT", session_id="sess-123")
 
# 5. Shutdown on exit (stops background refresh, clears cache)
shutdown_runtime()

from risicare import FixRuntimeConfig
 
config = FixRuntimeConfig(
    api_endpoint="https://app.risicare.ai",
    api_key="rsk-...",
    cache_ttl_seconds=300,
    dry_run=True,
)

config = FixRuntimeConfig.from_env()

runtime = FixRuntime(config=config)
 
# Start: loads fixes from API and begins background refresh
runtime.start()
 
# ... your application runs ...
 
# Stop: halts background refresh and clears cache
runtime.stop()

fix = runtime.get_fix(
    error_code="TOOL.EXECUTION.TIMEOUT",
    session_id="session-123",  # optional, used for A/B bucketing
)
 
if fix:
    print(f"Fix {fix.fix_id} (type: {fix.fix_type})")

def call_llm(messages: list) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    return response.choices[0].message.content
 
wrapped_fn = runtime.wrap_call(
    call_llm,
    operation_type="llm_call",
    operation_name="gpt-4o",
    session_id="session-123",
    max_retries=3,
)
 
# Call the wrapped function -- once stage-4 promotion is connected, any
# active fix for this operation would be applied here (none today: the
# active-fixes list is empty, so this call runs unchanged)
result = wrapped_fn(messages=[{"role": "user", "content": "Hello"}])

async def call_llm_async(messages: list) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    return response.choices[0].message.content
 
wrapped_fn = runtime.wrap_async_call(  # sync — returns an async callable
    call_llm_async,
    operation_type="llm_call",
    operation_name="gpt-4o",
    session_id="session-123",
    max_retries=3,
)
 
result = await wrapped_fn(messages=[{"role": "user", "content": "Hello"}])

# Pre-call: apply prompt and parameter fixes
messages, params, context = runtime.intercept_call(
    operation_type="llm_call",
    operation_name="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    params={"temperature": 0.7},
    session_id="session-123",
    error_code="TOOL.EXECUTION.TIMEOUT",  # if retrying after error
)
 
# Make the call with modified messages/params
try:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        **params,
    )
 
    # Post-call: apply output guards
    response, should_continue = runtime.intercept_response(context, response)
 
except Exception as e:
    # Error: decide whether to retry
    should_retry, modified_params = runtime.intercept_error(context, e)
    if should_retry:
        # Retry with modified_params
        pass

# Synchronous
fixes = runtime.refresh_fixes()
 
# Asynchronous
fixes = await runtime.refresh_fixes_async()

stats = runtime.get_effectiveness_stats()
for fix_id, data in stats.items():
    print(f"{fix_id}: {data['success_rate']:.1%} "
          f"({data['successes']}/{data['applications']})")

runtime.add_interceptor(MyCustomInterceptor())

from risicare.runtime.interceptors import FixInterceptor, InterceptContext
 
class MyInterceptor(FixInterceptor):
    def pre_call(self, context: InterceptContext, messages=None, params=None):
        """Called before an LLM/tool call. Return (messages, params)."""
        if context.operation_name == "gpt-4o":
            params = params or {}
            params["temperature"] = 0.3
        return messages, params
 
    def post_call(self, context: InterceptContext, response):
        """Called after a call. Return (response, should_continue)."""
        # Validate output, transform response, etc.
        return response, True
 
    def on_error(self, context: InterceptContext, error: Exception):
        """Called on error. Return (should_retry, modified_params)."""
        if "timeout" in str(error).lower():
            return True, {"timeout": 60000}
        return False, None
 
runtime.add_interceptor(MyInterceptor())

config = FixRuntimeConfig(
    ab_testing_enabled=True,  # Enable A/B testing
)

# Rollback a deployment via API
curl -X DELETE "https://app.risicare.ai/api/v1/deployments/{id}" \
  -H "Authorization: Bearer rsk-..."

from risicare import init_runtime, get_runtime, shutdown_runtime
 
# Initialize and start
runtime = init_runtime(config=config, auto_start=True)
 
# Get the global instance anywhere
runtime = get_runtime()
 
# Shutdown
shutdown_runtime()