1  #!/usr/bin/env python3
  2  """
  3  PraisonAI Run History Example
  4  
  5  This example demonstrates how to:
  6  1. Store recipe run results in history
  7  2. List and query run history
  8  3. Export runs for replay/debugging
  9  
 10  Prerequisites:
 11  - pip install praisonai
 12  
 13  Usage:
 14      python run_history_example.py
 15  """
 16  
 17  import tempfile
 18  from pathlib import Path
 19  
 20  
 21  def main():
 22      """Main example function."""
 23      print("=" * 60)
 24      print("PraisonAI Run History Example")
 25      print("=" * 60)
 26      
 27      # Use a temporary directory for the example
 28      with tempfile.TemporaryDirectory() as tmp_dir:
 29          tmp_path = Path(tmp_dir)
 30          
 31          # Create history storage in temp directory
 32          history_path = tmp_path / "runs"
 33          
 34          print("\n1. Creating run history storage...")
 35          from praisonai.recipe.history import RunHistory
 36          from praisonai.recipe.models import RecipeResult, RecipeStatus
 37          
 38          history = RunHistory(history_path)
 39          print(f"   Storage created at: {history_path}")
 40          
 41          # Create sample run results
 42          print("\n2. Storing sample run results...")
 43          
 44          # Run 1: Successful run
 45          result1 = RecipeResult(
 46              run_id="run-abc123",
 47              recipe="support-reply",
 48              version="1.0.0",
 49              status=RecipeStatus.SUCCESS,
 50              output={"reply": "Thank you for contacting us..."},
 51              metrics={"duration_sec": 2.5, "tokens": 150},
 52              trace={"session_id": "session-001", "trace_id": "trace-xyz"},
 53          )
 54          
 55          history.store(
 56              result=result1,
 57              input_data={"ticket_id": "T-123", "message": "I need help"},
 58          )
 59          print(f"   Stored: {result1.run_id} (success)")
 60          
 61          # Run 2: Failed run
 62          result2 = RecipeResult(
 63              run_id="run-def456",
 64              recipe="code-review",
 65              version="2.0.0",
 66              status=RecipeStatus.FAILED,
 67              error="Timeout exceeded",
 68              metrics={"duration_sec": 30.0},
 69              trace={"session_id": "session-002"},
 70          )
 71          
 72          history.store(
 73              result=result2,
 74              input_data={"code": "def hello(): pass"},
 75          )
 76          print(f"   Stored: {result2.run_id} (failed)")
 77          
 78          # Run 3: Another successful run
 79          result3 = RecipeResult(
 80              run_id="run-ghi789",
 81              recipe="support-reply",
 82              version="1.0.0",
 83              status=RecipeStatus.SUCCESS,
 84              output={"reply": "Your issue has been resolved..."},
 85              metrics={"duration_sec": 1.8},
 86              trace={"session_id": "session-001"},
 87          )
 88          
 89          history.store(
 90              result=result3,
 91              input_data={"ticket_id": "T-456"},
 92          )
 93          print(f"   Stored: {result3.run_id} (success)")
 94          
 95          # List all runs
 96          print("\n3. Listing all runs...")
 97          runs = history.list_runs()
 98          for run in runs:
 99              status_icon = "✓" if run["status"] == "success" else "✗"
100              print(f"   {status_icon} {run['run_id']} - {run['recipe']} ({run['status']})")
101          
102          # Filter by recipe
103          print("\n4. Filtering runs by recipe 'support-reply'...")
104          filtered = history.list_runs(recipe="support-reply")
105          print(f"   Found {len(filtered)} run(s)")
106          
107          # Filter by session
108          print("\n5. Filtering runs by session 'session-001'...")
109          filtered = history.list_runs(session_id="session-001")
110          print(f"   Found {len(filtered)} run(s)")
111          
112          # Get specific run details
113          print("\n6. Getting details for run-abc123...")
114          run_data = history.get("run-abc123")
115          print(f"   Recipe: {run_data['recipe']}")
116          print(f"   Status: {run_data['status']}")
117          print(f"   Input: {run_data.get('input', {})}")
118          print(f"   Output: {run_data.get('output', {})}")
119          
120          # Export a run
121          print("\n7. Exporting run-abc123...")
122          export_path = history.export("run-abc123", tmp_path / "export.json")
123          print(f"   Exported to: {export_path}")
124          
125          # Show export contents
126          import json
127          with open(export_path) as f:
128              export_data = json.load(f)
129          print(f"   Format: {export_data['format']}")
130          print(f"   Exported at: {export_data['exported_at']}")
131          
132          # Get storage stats
133          print("\n8. Getting storage statistics...")
134          stats = history.get_stats()
135          print(f"   Total runs: {stats['total_runs']}")
136          print(f"   Storage size: {stats['total_size_bytes']} bytes")
137          
138          # Cleanup old runs
139          print("\n9. Running cleanup...")
140          deleted = history.cleanup(retention_days=0)  # Delete all for demo
141          print(f"   Deleted {deleted} run(s)")
142          
143          print("\n" + "=" * 60)
144          print("Example completed successfully!")
145          print("=" * 60)
146  
147  
148  if __name__ == "__main__":
149      main()