Coverage for agentos/swarm/human_loop.py: 48%
119 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 19:15 +0800
1"""
2v1.9.5: Human-in-the-Loop (HITL) breakpoint system.
4Enables task execution to pause at configurable checkpoints for human
5review, approval, or intervention before continuing.
6"""
8from __future__ import annotations
10import asyncio
11import time
12import uuid
13from collections.abc import Callable
14from dataclasses import dataclass, field
15from enum import StrEnum
16from typing import Any
19class BreakpointType(StrEnum):
20 """Types of human-in-the-loop breakpoints."""
22 BEFORE_TASK = "before_task" # Before a sub-task starts
23 AFTER_RESULT = "after_result" # After a sub-task produces output
24 ON_FAILURE = "on_failure" # When a sub-task fails
25 ON_LOW_CONFIDENCE = "on_low_confidence" # When fusion confidence is low
26 MANUAL = "manual" # Explicitly placed by developer
29class HumanDecision(StrEnum):
30 """Human responses at a breakpoint."""
32 APPROVE = "approve" # Approve and continue
33 REJECT = "reject" # Reject and skip/retry
34 RETRY = "retry" # Reject and retry with feedback
35 MODIFY = "modify" # Accept with modifications
36 ABORT = "abort" # Abort entire task
39@dataclass
40class Breakpoint:
41 """A checkpoint where execution pauses for human input."""
43 id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
44 type: BreakpointType = BreakpointType.MANUAL
45 task_id: str = ""
46 context: dict[str, Any] = field(default_factory=dict)
47 message: str = ""
48 options: list[str] = field(default_factory=lambda: ["approve", "reject", "retry", "abort"])
49 timeout: float = 0.0 # 0 = no timeout
50 created_at: float = field(default_factory=time.time)
51 resolved_at: float = 0.0
52 decision: HumanDecision | None = None
53 feedback: str = ""
54 resolved: bool = False
56 def to_dict(self) -> dict:
57 return {
58 "id": self.id,
59 "type": self.type.value,
60 "task_id": self.task_id,
61 "message": self.message,
62 "options": self.options,
63 "resolved": self.resolved,
64 "decision": self.decision.value if self.decision else None,
65 }
68@dataclass
69class HITLConfig:
70 """Configuration for human-in-the-loop behavior."""
72 enabled: bool = True
73 break_on_failure: bool = True
74 break_on_low_confidence: float = 0.3 # confidence below this triggers break
75 break_on_first_task: bool = False # break before first sub-task
76 break_on_every_task: bool = False
77 break_on_final_result: bool = False # break before returning final result
78 max_pending_breakpoints: int = 5 # queue limit
79 default_timeout: float = 300.0 # 5 min default
82class HITLManager:
83 """Manages human-in-the-loop breakpoints during task execution.
85 Usage:
86 hitl = HITLManager(config=HITLConfig(break_on_failure=True))
88 # Register a callback for human input
89 hitl.register_handler(my_human_input_function)
91 # During execution:
92 decision = await hitl.request_decision(
93 bp_type=BreakpointType.ON_FAILURE,
94 task_id="task_1",
95 message="Task failed. Retry?",
96 context={"error": "...", "attempts": 2}
97 )
98 if decision == HumanDecision.RETRY:
99 ...
100 """
102 def __init__(
103 self,
104 config: HITLConfig | None = None,
105 handler: Callable | None = None,
106 ):
107 self.config = config or HITLConfig()
108 self._handler = handler
109 self._breakpoints: dict[str, Breakpoint] = {}
110 self._pending: list[Breakpoint] = []
111 self._decision_queue: asyncio.Queue = asyncio.Queue()
113 def register_handler(self, handler: Callable[[Breakpoint], HumanDecision]) -> None:
114 """
115 Register a human input handler.
117 Args:
118 handler: Callable that receives a Breakpoint and returns a HumanDecision.
119 Can be sync or async.
120 """
121 self._handler = handler
123 async def request_decision(
124 self,
125 bp_type: BreakpointType,
126 task_id: str,
127 message: str,
128 context: dict | None = None,
129 timeout: float | None = None,
130 options: list[str] | None = None,
131 ) -> tuple[HumanDecision, str]:
132 """
133 Pause execution and request human decision.
135 Args:
136 bp_type: Type of breakpoint
137 task_id: Current task identifier
138 message: Human-readable message explaining what's needed
139 context: Additional context for the decision
140 timeout: Max wait time (None = use config default)
141 options: Available decision options
143 Returns:
144 Tuple of (decision, feedback text)
145 """
146 if not self.config.enabled:
147 return HumanDecision.APPROVE, ""
149 bp = Breakpoint(
150 type=bp_type,
151 task_id=task_id,
152 context=context or {},
153 message=message,
154 options=options or ["approve", "reject", "retry", "abort"],
155 timeout=timeout or self.config.default_timeout,
156 )
158 self._breakpoints[bp.id] = bp
159 self._pending.append(bp)
161 # If pending exceeds limit, auto-approve oldest
162 if len(self._pending) > self.config.max_pending_breakpoints:
163 oldest = self._pending.pop(0)
164 oldest.decision = HumanDecision.APPROVE
165 oldest.resolved = True
166 oldest.resolved_at = time.time()
168 # Call handler
169 if self._handler:
170 try:
171 result = self._handler(bp)
172 if asyncio.iscoroutine(result):
173 result = await result
174 if isinstance(result, HumanDecision):
175 bp.decision = result
176 elif isinstance(result, tuple) and len(result) == 2:
177 bp.decision, bp.feedback = result
178 else:
179 bp.decision = HumanDecision.APPROVE
180 except Exception:
181 bp.decision = HumanDecision.APPROVE
182 else:
183 # No handler — wait on queue
184 try:
185 decision, feedback = await asyncio.wait_for(
186 self._decision_queue.get(),
187 timeout=bp.timeout,
188 )
189 bp.decision = decision
190 bp.feedback = feedback
191 except TimeoutError:
192 bp.decision = HumanDecision.APPROVE
194 bp.resolved = True
195 bp.resolved_at = time.time()
197 # Remove from pending
198 if bp in self._pending:
199 self._pending.remove(bp)
201 return bp.decision, bp.feedback
203 def provide_decision(
204 self,
205 breakpoint_id: str,
206 decision: HumanDecision,
207 feedback: str = "",
208 ) -> None:
209 """Provide a decision for a pending breakpoint (alternative to handler)."""
210 if breakpoint_id in self._breakpoints:
211 self._breakpoints[breakpoint_id]
212 self._decision_queue.put_nowait((decision, feedback))
214 async def should_break_before_task(self, task_id: str, task_name: str) -> bool:
215 """Check if we should break before a sub-task."""
216 if not self.config.enabled:
217 return False
218 if self.config.break_on_first_task or self.config.break_on_every_task:
219 decision, _ = await self.request_decision(
220 bp_type=BreakpointType.BEFORE_TASK,
221 task_id=task_id,
222 message=f"About to execute: {task_name}\nProceed?",
223 options=["approve", "abort", "modify"],
224 )
225 if decision == HumanDecision.ABORT:
226 return False
227 return True
229 async def should_break_on_result(
230 self, task_id: str, output: Any, confidence: float
231 ) -> tuple[HumanDecision, str]:
232 """Check if we should break after a result."""
233 if not self.config.enabled:
234 return HumanDecision.APPROVE, ""
236 # Low confidence trigger
237 if confidence < self.config.break_on_low_confidence:
238 return await self.request_decision(
239 bp_type=BreakpointType.ON_LOW_CONFIDENCE,
240 task_id=task_id,
241 message=(
242 f"Low confidence result (confidence: {confidence:.2f})\n"
243 f"Output: {str(output)[:300]}\n"
244 f"What would you like to do?"
245 ),
246 context={"confidence": confidence, "output": str(output)[:500]},
247 options=["approve", "retry", "modify", "abort"],
248 )
250 # Final result break
251 if self.config.break_on_final_result:
252 return await self.request_decision(
253 bp_type=BreakpointType.AFTER_RESULT,
254 task_id=task_id,
255 message=f"Result: {str(output)[:300]}\nApprove?",
256 context={"output": str(output)[:500]},
257 options=["approve", "retry", "modify"],
258 )
260 return HumanDecision.APPROVE, ""
262 async def should_break_on_failure(
263 self, task_id: str, error: str, attempt: int
264 ) -> tuple[HumanDecision, str]:
265 """Check if we should break on failure."""
266 if not self.config.enabled or not self.config.break_on_failure:
267 return HumanDecision.RETRY, ""
269 return await self.request_decision(
270 bp_type=BreakpointType.ON_FAILURE,
271 task_id=task_id,
272 message=(
273 f"Task failed (attempt {attempt})\n"
274 f"Error: {error[:300]}\n"
275 f"Retry, skip, or abort?"
276 ),
277 context={"error": error, "attempt": attempt},
278 options=["retry", "abort", "modify"],
279 )
281 @property
282 def pending_count(self) -> int:
283 return len(self._pending)
285 @property
286 def total_breakpoints(self) -> int:
287 return len(self._breakpoints)