MCPcopy Create free account
hub / github.com/Emmimal/control-layer / demo_4_fallback_router

Function demo_4_fallback_router

demo.py:272–309  ·  view source on GitHub ↗
()

Source from the content-addressed store, hash-verified

270# =============================================================================
271
272def demo_4_fallback_router():
273 print("\n" + "=" * 60)
274 print("Demo 4: Fallback Router - Exhausted Retries")
275 print("=" * 60)
276
277 schema = ResponseSchema(must_be_json=True, required_keys=["result"])
278 config = ControlLayerConfig(total_tokens=800, max_attempts=3, base_delay_ms=50)
279
280 llm = make_deterministic_llm([
281 "This is not JSON at all.",
282 "Still not JSON: just plain text.",
283 "Nope, still plain text on attempt 3.",
284 ])
285 layer = ControlLayer(
286 llm_fn=llm,
287 system_prompt="You are a structured output assistant.",
288 schema=schema,
289 config=config,
290 )
291 layer.register_fallback(
292 "cached_response",
293 lambda q: json.dumps({"result": "fallback", "source": "cache", "query": q[:50]}),
294 )
295 layer.register_fallback(
296 "escalate",
297 lambda q: json.dumps({"result": "escalated", "reason": "max retries exceeded"}),
298 )
299
300 packet = layer.run(
301 user_input="What is the result of running the context engine on query X?",
302 constraints=["Respond only with a JSON object containing a 'result' key."],
303 )
304
305 print(f" Final outcome: {'PASSED' if packet.validation.passed else 'FAILED'}")
306 print(f" Strategy used: {packet.strategy_used.value}")
307 print(f" Attempts used: {packet.attempts}")
308 print(f" Response: {packet.response[:120]}")
309 return packet, layer.audit
310
311
312# =============================================================================

Callers 1

demo.pyFile · 0.85

Calls 6

register_fallbackMethod · 0.95
runMethod · 0.95
ResponseSchemaClass · 0.90
ControlLayerConfigClass · 0.90
ControlLayerClass · 0.90
make_deterministic_llmFunction · 0.85

Tested by

no test coverage detected