File size: 1,832 Bytes
70b8adf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | #!/usr/bin/env python3
"""Minimal example of placing Dot Reflex around an existing agent harness."""
from __future__ import annotations
from typing import Any
from inference import DotReflex
class Harness:
"""Replace these methods with the primitives from your agent runtime."""
def snapshot_for_supervisor(self) -> dict[str, Any]:
return {
"task_summary": "Fix the failing authentication test",
"execution_history": [
{
"step": 1,
"actor": "agent",
"action": "Edited auth.py and reported completion.",
"result": "Patch applied; tests were not run.",
}
],
"tool_results": [
{"tool": "pytest", "status": "not_run", "summary": "No test receipt."}
],
"current_state": "A code change exists without verification evidence.",
"detected_failure_signals": ["false_completion_risk"],
}
def dispatch(self, decision: dict[str, Any]) -> None:
action = decision["action"]
if action == "continue":
print("Resume the current plan")
elif action == "verify":
print("Run a verification gate:", decision["recovery_instructions"])
elif action == "ask_human":
print("Pause and ask the user:", decision["rationale"])
elif action in {"stop_successfully", "stop_failure"}:
print("Stop the run:", action)
else:
print("Apply recovery control:", action, decision["recovery_instructions"])
def main() -> None:
harness = Harness()
reflex = DotReflex(adapter="usedot/Dot-Reflex-14B")
harness.dispatch(reflex.predict(harness.snapshot_for_supervisor()))
if __name__ == "__main__":
main()
|