Skip to content

Commit 2e2bc58

Browse files
silversurfer562claude
silversurfer562
and
claude
committed
fix: Resolve remaining lint issues in benchmarks and examples
- Fix C401: Use set comprehension in benchmark_semantic_cache.py - Fix E741: Rename ambiguous variable 'l' to 'line' in manual_test.py - Fix F841: Rename unused executors to '_executor' with usage comments - Add pragma comment for test fixture false positive All ruff checks now passing in benchmarks/ and examples/ directories. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
1 parent 96a8853 commit 2e2bc58

3 files changed

Lines changed: 233 additions & 3 deletions

File tree

‎benchmarks/benchmark_semantic_cache.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ def run_command(user_input):
6969
os.system(f"echo {user_input}")
7070
7171
def get_secret():
72-
password = "admin123" # Hardcoded secret
72+
password = "admin123" # Hardcoded secret # pragma: allowlist secret
7373
return password
7474
""",
7575
"semantically_similar": """
@@ -307,7 +307,7 @@ def generate_report(results: list[BenchmarkResult], output_file: str = "SEMANTIC
307307
## Comparison: Hash vs Hybrid Cache
308308
309309
"""
310-
for workflow_name in set(r.test_name for r in results):
310+
for workflow_name in {r.test_name for r in results}:
311311
hash_result = next((r for r in hash_results if r.test_name == workflow_name), None)
312312
hybrid_result = next((r for r in hybrid_results if r.test_name == workflow_name), None)
313313

‎benchmarks/manual_test.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -330,7 +330,7 @@ def test_unit_tests():
330330
if "passed" in output:
331331
# Extract test count
332332
lines = output.split("\n")
333-
summary_line = [l for l in lines if "passed" in l][-1]
333+
summary_line = [line for line in lines if "passed" in line][-1]
334334
print_success(f"All unit tests: {summary_line.strip()}")
335335
else:
336336
print_error("Some unit tests failed")

‎examples/sonnet_opus_fallback_example.py‎

Lines changed: 230 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,230 @@
1+
"""Example: Sonnet 4.5 → Opus 4.5 Intelligent Fallback
2+
3+
Demonstrates how to use the intelligent fallback system that automatically
4+
tries Sonnet 4.5 first and upgrades to Opus 4.5 when needed.
5+
6+
This can save up to 80% on API costs while maintaining high quality.
7+
8+
Copyright 2025 Smart-AI-Memory
9+
Licensed under Fair Source License 0.9
10+
"""
11+
12+
# Load .env file first
13+
try:
14+
from dotenv import load_dotenv
15+
16+
load_dotenv() # Load environment variables from .env
17+
except ImportError:
18+
pass # dotenv not installed, continue anyway
19+
20+
import asyncio
21+
from datetime import datetime, timedelta
22+
23+
from empathy_os.models.empathy_executor import EmpathyLLMExecutor
24+
from empathy_os.models.fallback import SONNET_TO_OPUS_FALLBACK, ResilientExecutor
25+
from empathy_os.models.telemetry import TelemetryAnalytics, get_telemetry_store
26+
27+
28+
async def example_basic_fallback():
29+
"""Basic example: Automatic Sonnet → Opus fallback."""
30+
print("=" * 60)
31+
print("Example 1: Basic Sonnet → Opus Fallback")
32+
print("=" * 60)
33+
34+
# Get API key from environment (Option 2: explicit passing)
35+
import os
36+
37+
api_key = os.getenv("ANTHROPIC_API_KEY")
38+
39+
if not api_key:
40+
print("⚠️ ANTHROPIC_API_KEY not found in environment")
41+
print(" Please set it in your .env file or export it.")
42+
return
43+
44+
# Create base executor with explicit API key
45+
base_executor = EmpathyLLMExecutor(
46+
provider="anthropic", api_key=api_key # Explicitly pass the API key
47+
)
48+
49+
# Wrap with resilient fallback
50+
executor = ResilientExecutor(
51+
executor=base_executor,
52+
fallback_policy=SONNET_TO_OPUS_FALLBACK,
53+
)
54+
55+
# Make a call - will try Sonnet 4.5 first
56+
print("\nCalling LLM with fallback enabled...")
57+
response = await executor.run(
58+
task_type="code_review",
59+
prompt="Review this Python code for security issues:\n\ndef process_user_input(data):\n return eval(data)",
60+
)
61+
62+
# Check which model was used
63+
if response.metadata.get("fallback_used"):
64+
print("✅ Fallback triggered: Upgraded to Opus 4.5")
65+
print(f" Reason: {response.metadata.get('fallback_chain')}")
66+
else:
67+
print("✅ Sonnet 4.5 succeeded (no fallback needed)")
68+
69+
print(f"\nResponse: {response.content[:200]}...")
70+
71+
72+
async def example_with_analytics():
73+
"""Example: Track cost savings over time."""
74+
print("\n" + "=" * 60)
75+
print("Example 2: Cost Savings Analytics")
76+
print("=" * 60)
77+
78+
# Get telemetry store
79+
store = get_telemetry_store()
80+
analytics = TelemetryAnalytics(store)
81+
82+
# Analyze last 7 days
83+
since = datetime.utcnow() - timedelta(days=7)
84+
stats = analytics.sonnet_opus_fallback_analysis(since=since)
85+
86+
if stats["total_calls"] == 0:
87+
print("\n⚠️ No Sonnet/Opus calls found in the last 7 days.")
88+
print(" Run some workflows first, then check back!")
89+
return
90+
91+
# Display results
92+
print("\n📊 Fallback Performance (last 7 days):")
93+
print(f" Total Calls: {stats['total_calls']}")
94+
print(f" Sonnet Attempts: {stats['sonnet_attempts']}")
95+
print(f" Sonnet Success Rate: {stats['success_rate_sonnet']:.1f}%")
96+
print(f" Opus Fallbacks: {stats['opus_fallbacks']}")
97+
print(f" Fallback Rate: {stats['fallback_rate']:.1f}%")
98+
99+
print("\n💰 Cost Savings:")
100+
print(f" Actual Cost: ${stats['actual_cost']:.2f}")
101+
print(f" Always-Opus Cost: ${stats['always_opus_cost']:.2f}")
102+
print(f" Savings: ${stats['savings']:.2f} ({stats['savings_percent']:.1f}%)")
103+
104+
# Recommendation
105+
if stats["fallback_rate"] < 5:
106+
print(f"\n✅ Excellent! Sonnet handles {100 - stats['fallback_rate']:.1f}% of tasks.")
107+
elif stats["fallback_rate"] < 15:
108+
print(f"\n⚠️ Moderate fallback rate ({stats['fallback_rate']:.1f}%).")
109+
else:
110+
print(f"\n❌ High fallback rate ({stats['fallback_rate']:.1f}%).")
111+
print(" Consider using Opus directly for complex tasks.")
112+
113+
114+
async def example_custom_retry():
115+
"""Example: Custom retry configuration."""
116+
print("\n" + "=" * 60)
117+
print("Example 3: Custom Retry Configuration")
118+
print("=" * 60)
119+
120+
import os
121+
122+
from empathy_os.models.fallback import RetryPolicy
123+
124+
# Get API key
125+
api_key = os.getenv("ANTHROPIC_API_KEY")
126+
127+
# Create custom retry policy
128+
custom_retry = RetryPolicy(
129+
max_retries=2, # Only 2 retries per model
130+
initial_delay_ms=500, # Start with 500ms
131+
exponential_backoff=True, # Double delay each time
132+
)
133+
134+
base_executor = EmpathyLLMExecutor(provider="anthropic", api_key=api_key)
135+
136+
_executor = ResilientExecutor(
137+
executor=base_executor,
138+
fallback_policy=SONNET_TO_OPUS_FALLBACK,
139+
retry_policy=custom_retry,
140+
)
141+
142+
print("\n✅ Created executor with custom retry policy:")
143+
print(f" Max retries: {custom_retry.max_retries}")
144+
print(f" Initial delay: {custom_retry.initial_delay_ms}ms")
145+
print(f" Exponential backoff: {custom_retry.exponential_backoff}")
146+
# In production: response = await _executor.run(task_type="...", prompt="...")
147+
148+
149+
async def example_direct_opus():
150+
"""Example: When to use Opus directly."""
151+
print("\n" + "=" * 60)
152+
print("Example 4: Direct Opus Usage (No Fallback)")
153+
print("=" * 60)
154+
155+
# For tasks you know need Opus, use it directly
156+
_executor = EmpathyLLMExecutor(provider="anthropic", default_tier="premium")
157+
# In production: response = await _executor.run(task_type="complex_task", prompt="...")
158+
159+
print("\n✅ Created executor using Opus 4.5 directly:")
160+
print(" Provider: anthropic")
161+
print(" Tier: premium (Opus 4.5)")
162+
print(" Use when: Task complexity requires Opus-level reasoning")
163+
print(" Benefit: Avoids retry overhead, faster response")
164+
165+
166+
async def example_circuit_breaker():
167+
"""Example: Circuit breaker status."""
168+
print("\n" + "=" * 60)
169+
print("Example 5: Circuit Breaker Status")
170+
print("=" * 60)
171+
172+
import os
173+
174+
api_key = os.getenv("ANTHROPIC_API_KEY")
175+
176+
base_executor = EmpathyLLMExecutor(provider="anthropic", api_key=api_key)
177+
executor = ResilientExecutor(
178+
executor=base_executor,
179+
fallback_policy=SONNET_TO_OPUS_FALLBACK,
180+
)
181+
182+
# Check circuit breaker status
183+
status = executor.circuit_breaker.get_status()
184+
185+
if not status:
186+
print("\n✅ Circuit breaker: All clear (no failures)")
187+
else:
188+
print("\n⚠️ Circuit breaker status:")
189+
for provider_tier, state in status.items():
190+
print(f"\n {provider_tier}:")
191+
print(f" Failures: {state['failure_count']}")
192+
print(f" Open: {state['is_open']}")
193+
if state["last_failure"]:
194+
print(f" Last failure: {state['last_failure']}")
195+
196+
# Tip
197+
print("\n💡 Tip: Circuit breaker protects against cascading failures")
198+
print(" After 5 consecutive failures, routes to fallback for 60s")
199+
200+
201+
async def main():
202+
"""Run all examples."""
203+
print("\n" + "=" * 60)
204+
print("Sonnet 4.5 → Opus 4.5 Intelligent Fallback Examples")
205+
print("=" * 60)
206+
207+
# Run examples
208+
await example_basic_fallback()
209+
await example_with_analytics()
210+
await example_custom_retry()
211+
await example_direct_opus()
212+
await example_circuit_breaker()
213+
214+
# Final tip
215+
print("\n" + "=" * 60)
216+
print("💡 Pro Tips:")
217+
print("=" * 60)
218+
print("1. Check fallback analytics weekly:")
219+
print(" python -m empathy_os.telemetry.cli sonnet-opus-analysis")
220+
print()
221+
print("2. Aim for < 5% fallback rate for optimal savings")
222+
print()
223+
print("3. Use Opus directly for known complex tasks")
224+
print()
225+
print("4. Monitor circuit breaker to detect systemic issues")
226+
print("=" * 60)
227+
228+
229+
if __name__ == "__main__":
230+
asyncio.run(main())

0 commit comments

Comments
 (0)