Coverage for agentos/hitl/gradio_ui.py: 34%
265 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2AgentOS v1.14.2 — Dynamic HITL UI (Gradio-based Approval Dashboard).
4受 LangGraph Studio / AutoGen UI 启发，在已有 HITL 审批引擎之上
5增加 Gradio 驱动的响应式审批面板。Agent 遇到高风险操作时，
6自动弹出 Web UI 而非阻塞终端。
8Core features:
9- GradioApp: 一键启动的审批 Dashboard
10- ApprovalQueue: 实时审批队列，WebSocket 推送
11- AgentStatusPanel: Agent 状态监控面板
12- ApprovalCard: 可定制的审批卡片组件
13- HistoryView: 审批历史追溯
14- PolicyEditor: 可视化策略编辑器
16与 hitl/approver.py 的关系:
17- approver.py: 审批引擎（决策逻辑、风险评级、策略执行）
18- gradio_ui.py: 交互层（Web UI、实时推送、可视化配置）
19"""
21from __future__ import annotations
23import asyncio
24import time
25import uuid
26import threading
27import queue
28from dataclasses import dataclass, field
29from datetime import datetime
30from enum import Enum
31from typing import (
32 Any, Callable, Dict, List, Optional, Tuple,
33)
36# ── UI Data Models ──────────────────────────
39class ApprovalStatus(str, Enum):
40 PENDING = "pending"
41 APPROVED = "approved"
42 DENIED = "denied"
43 EXPIRED = "expired"
44 CANCELLED = "cancelled"
47class RiskLevelUI(str, Enum):
48 SAFE = "safe"
49 LOW = "low"
50 MEDIUM = "medium"
51 HIGH = "high"
52 CRITICAL = "critical"
55@dataclass
56class ApprovalRequestUI:
57 """UI 层的审批请求，与底层 HITL 解耦。"""
59 request_id: str = field(default_factory=lambda: f"apr-{uuid.uuid4().hex[:8]}")
60 agent_name: str = ""
61 action: str = "" # 人类可读的操作描述
62 details: str = "" # 详细说明
63 risk_level: RiskLevelUI = RiskLevelUI.MEDIUM
64 status: ApprovalStatus = ApprovalStatus.PENDING
65 created_at: float = field(default_factory=time.time)
66 expires_at: float = 0.0 # 超时自动拒绝
67 source_file: str = "" # 触发操作的文件/代码位置
68 metadata: Dict[str, Any] = field(default_factory=dict)
69 # Callback when approved/denied
70 on_decision: Optional[Callable] = None
72 @property
73 def elapsed_seconds(self) -> float:
74 return time.time() - self.created_at
76 @property
77 def is_expired(self) -> bool:
78 if self.expires_at <= 0:
79 return False
80 return time.time() > self.expires_at
83@dataclass
84class ApprovalHistory:
85 """审批历史记录。"""
87 request: ApprovalRequestUI
88 decision: ApprovalStatus
89 decided_by: str = "user" # "user" | "auto" | "timeout"
90 reason: str = ""
91 decided_at: float = field(default_factory=time.time)
93 def to_dict(self) -> dict:
94 return {
95 "request_id": self.request.request_id,
96 "action": self.request.action,
97 "risk_level": self.request.risk_level.value,
98 "decision": self.decision.value,
99 "decided_by": self.decided_by,
100 "reason": self.reason,
101 "created_at": self.request.created_at,
102 "decided_at": self.decided_at,
103 "elapsed_ms": int((self.decided_at - self.request.created_at) * 1000),
104 }
107@dataclass
108class AgentStatusSnapshot:
109 """Agent 运行状态快照。"""
111 agent_id: str = ""
112 agent_name: str = ""
113 status: str = "idle" # idle | running | waiting_approval | paused | error
114 current_task: str = ""
115 elapsed_seconds: float = 0.0
116 pending_approvals: int = 0
117 memory_fragments: int = 0
118 last_error: str = ""
121# ── Approval Queue ─────────────────────────
124class ApprovalQueue:
125 """线程安全的审批队列，支持 WebSocket 推送通知。
127 在 Gradio UI 与 Agent HITL 引擎之间架设实时通信桥梁。
128 """
130 def __init__(self, max_size: int = 100, default_timeout: float = 300.0):
131 self._queue: queue.Queue = queue.Queue(maxsize=max_size)
132 self._pending: Dict[str, ApprovalRequestUI] = {}
133 self._history: List[ApprovalHistory] = []
134 self._subscribers: List[Callable] = [] # WebSocket callbacks
135 self._default_timeout = default_timeout
136 self._lock = threading.Lock()
138 def submit(self, request: ApprovalRequestUI) -> str:
139 """提交审批请求。注册到队列并通知订阅者。"""
140 if request.expires_at <= 0:
141 request.expires_at = time.time() + self._default_timeout
143 with self._lock:
144 self._pending[request.request_id] = request
146 self._notify_subscribers({
147 "event": "new_request",
148 "request_id": request.request_id,
149 "action": request.action,
150 "risk_level": request.risk_level.value,
151 "pending_count": len(self._pending),
152 })
154 return request.request_id
156 def decide(self, request_id: str, approved: bool, reason: str = "") -> Optional[ApprovalHistory]:
157 """处理审批决定。"""
158 with self._lock:
159 request = self._pending.pop(request_id, None)
161 if request is None:
162 return None
164 decision = ApprovalStatus.APPROVED if approved else ApprovalStatus.DENIED
165 history = ApprovalHistory(
166 request=request,
167 decision=decision,
168 reason=reason,
169 )
171 # Trigger callback
172 if request.on_decision:
173 try:
174 request.on_decision(approved, reason)
175 except Exception:
176 pass
178 with self._lock:
179 self._history.append(history)
181 self._notify_subscribers({
182 "event": "decision",
183 "request_id": request_id,
184 "approved": approved,
185 "reason": reason,
186 "pending_count": len(self._pending),
187 })
189 return history
191 def approve_all(self) -> int:
192 """批量批准所有待处理请求。"""
193 ids = list(self._pending.keys())
194 for rid in ids:
195 self.decide(rid, True, "batch_approve")
196 return len(ids)
198 def deny_all(self) -> int:
199 """批量拒绝所有待处理请求。"""
200 ids = list(self._pending.keys())
201 for rid in ids:
202 self.decide(rid, False, "batch_deny")
203 return len(ids)
205 def check_timeouts(self) -> int:
206 """检查并自动拒绝超时请求。"""
207 now = time.time()
208 expired_ids = []
209 with self._lock:
210 for rid, req in self._pending.items():
211 if req.is_expired:
212 expired_ids.append(rid)
214 for rid in expired_ids:
215 history = self.decide(rid, False, "timeout")
216 if history:
217 history.decided_by = "timeout"
219 return len(expired_ids)
221 @property
222 def pending_requests(self) -> List[ApprovalRequestUI]:
223 with self._lock:
224 return list(self._pending.values())
226 @property
227 def pending_count(self) -> int:
228 with self._lock:
229 return len(self._pending)
231 @property
232 def recent_history(self, limit: int = 50) -> List[ApprovalHistory]:
233 with self._lock:
234 return self._history[-limit:]
236 def subscribe(self, callback: Callable) -> None:
237 """注册 WebSocket 通知回调。"""
238 self._subscribers.append(callback)
240 def _notify_subscribers(self, data: dict) -> None:
241 for cb in self._subscribers:
242 try:
243 cb(data)
244 except Exception:
245 pass
248# ── Gradio Approval Dashboard ──────────────
251class ApprovalDashboard:
252 """Gradio 驱动的审批面板。
254 核心布局:
255 - 顶部: Agent 状态栏
256 - 左侧: 待审批队列
257 - 右侧: 审批详情 + 历史
258 - 底部: 批量操作按钮
260 Usage:
261 dashboard = ApprovalDashboard(queue, port=7860)
262 dashboard.launch()
263 """
265 def __init__(
266 self,
267 approval_queue: ApprovalQueue,
268 port: int = 7860,
269 title: str = "AgentOS — HITL Approval Dashboard",
270 theme: str = "soft",
271 auto_launch: bool = True,
272 ):
273 self._queue = approval_queue
274 self._port = port
275 self._title = title
276 self._theme = theme
277 self._auto_launch = auto_launch
278 self._app = None
279 self._agent_statuses: Dict[str, AgentStatusSnapshot] = {}
280 self._selected_request_id: str = ""
282 def launch(self, share: bool = False) -> Any:
283 """启动 Gradio 面板。
285 返回 Gradio Blocks 实例，可在 Jupyter 中内嵌或独立运行。
286 """
287 try:
288 import gradio as gr
289 except ImportError:
290 raise ImportError(
291 "Gradio is required for ApprovalDashboard. "
292 "Install with: pip install gradio>=4.0"
293 )
295 with gr.Blocks(title=self._title, theme=self._theme) as app:
296 self._app = app
297 self._build_ui(app)
299 if self._auto_launch:
300 app.launch(
301 server_port=self._port,
302 share=share,
303 prevent_thread_lock=False,
304 )
306 return app
308 def _build_ui(self, app: Any) -> None:
309 """构建完整 UI 布局。"""
310 import gradio as gr
312 # ── Header ──
313 gr.Markdown(
314 f"# {self._title}\n"
315 f"### Real-time Human-in-the-Loop Approval Dashboard"
316 )
318 # ── Agent Status Bar ──
319 with gr.Row():
320 self._agent_status_display = gr.HTML(
321 value=self._render_agent_status_bar(),
322 every=3.0,
323 )
325 # ── Main Layout ──
326 with gr.Row():
327 # Left: Pending Queue
328 with gr.Column(scale=1):
329 gr.Markdown("### Pending Approvals")
330 self._queue_list = gr.HTML(
331 value=self._render_pending_queue(),
332 every=2.0,
333 )
334 with gr.Row():
335 self._btn_approve_all = gr.Button(
336 "Approve All", variant="primary", size="sm"
337 )
338 self._btn_deny_all = gr.Button(
339 "Deny All", variant="stop", size="sm"
340 )
342 # Right: Detail + History
343 with gr.Column(scale=2):
344 with gr.Tabs():
345 with gr.TabItem("Approval Detail"):
346 self._detail_view = gr.HTML(
347 value="<p>Select a request from the queue...</p>"
348 )
349 with gr.Row():
350 self._btn_approve = gr.Button(
351 "Approve", variant="primary"
352 )
353 self._btn_deny = gr.Button(
354 "Deny", variant="stop"
355 )
356 self._reason_input = gr.Textbox(
357 label="Reason (optional)",
358 placeholder="Why this decision...",
359 )
361 with gr.TabItem("History"):
362 self._history_view = gr.HTML(
363 value=self._render_history(),
364 every=3.0,
365 )
367 with gr.TabItem("Policy"):
368 self._policy_view = gr.HTML(
369 value=self._render_policy_editor()
370 )
372 # ── Event Handlers ──
373 self._btn_approve.click(
374 fn=self._handle_approve,
375 inputs=[self._reason_input],
376 outputs=[self._detail_view, self._queue_list, self._history_view],
377 )
378 self._btn_deny.click(
379 fn=self._handle_deny,
380 inputs=[self._reason_input],
381 outputs=[self._detail_view, self._queue_list, self._history_view],
382 )
383 self._btn_approve_all.click(
384 fn=lambda: self._queue.approve_all(),
385 outputs=[],
386 )
387 self._btn_deny_all.click(
388 fn=lambda: self._queue.deny_all(),
389 outputs=[],
390 )
392 def _handle_approve(self, reason: str) -> Tuple[str, str, str]:
393 if not self._selected_request_id:
394 return self._detail_view, self._queue_list, self._history_view
395 self._queue.decide(self._selected_request_id, True, reason)
396 self._selected_request_id = ""
397 return (
398 "<p>Select a request from the queue...</p>",
399 self._render_pending_queue(),
400 self._render_history(),
401 )
403 def _handle_deny(self, reason: str) -> Tuple[str, str, str]:
404 if not self._selected_request_id:
405 return self._detail_view, self._queue_list, self._history_view
406 self._queue.decide(self._selected_request_id, False, reason)
407 self._selected_request_id = ""
408 return (
409 "<p>Select a request from the queue...</p>",
410 self._render_pending_queue(),
411 self._render_history(),
412 )
414 def update_agent_status(self, snapshot: AgentStatusSnapshot) -> None:
415 """更新 Agent 状态（由外部 Agent loop 调用）。"""
416 self._agent_statuses[snapshot.agent_id] = snapshot
418 def _render_agent_status_bar(self) -> str:
419 """渲染 Agent 状态栏 HTML。"""
420 if not self._agent_statuses:
421 return (
422 '<div style="padding:12px;background:#f0f0f0;border-radius:8px;">'
423 '<span style="color:#888;">No agents connected</span></div>'
424 )
426 rows = []
427 for agent_id, snap in self._agent_statuses.items():
428 status_color = {
429 "idle": "#4CAF50",
430 "running": "#2196F3",
431 "waiting_approval": "#FF9800",
432 "paused": "#9E9E9E",
433 "error": "#F44336",
434 }.get(snap.status, "#9E9E9E")
436 rows.append(
437 f'<div style="display:inline-block;margin:4px 8px;padding:8px 12px;'
438 f'background:#fff;border-radius:6px;border-left:4px solid {status_color};">'
439 f'<b>{snap.agent_name}</b> '
440 f'<span style="color:{status_color};">● {snap.status}</span> '
441 f'| {snap.current_task[:30]} '
442 f'| {snap.pending_approvals} pending'
443 f'</div>'
444 )
446 return (
447 '<div style="padding:12px;background:#f0f0f0;border-radius:8px;">'
448 + "".join(rows)
449 + "</div>"
450 )
452 def _render_pending_queue(self) -> str:
453 """渲染待处理队列 HTML。"""
454 pending = self._queue.pending_requests
455 if not pending:
456 return '<p style="color:#888;">No pending approvals</p>'
458 risk_colors = {
459 "safe": "#4CAF50",
460 "low": "#8BC34A",
461 "medium": "#FF9800",
462 "high": "#F44336",
463 "critical": "#B71C1C",
464 }
466 cards = []
467 for req in pending:
468 color = risk_colors.get(req.risk_level.value, "#999")
469 elapsed = int(req.elapsed_seconds)
470 cards.append(
471 f'<div onclick="selectRequest(\'{req.request_id}\')" '
472 f'style="cursor:pointer;margin:6px 0;padding:10px;'
473 f'background:#fff;border-radius:6px;'
474 f'border-left:4px solid {color};">'
475 f'<div style="font-weight:bold;">{req.action[:60]}</div>'
476 f'<div style="color:{color};font-size:0.85em;">'
477 f'Risk: {req.risk_level.value} | Agent: {req.agent_name} | '
478 f'{elapsed}s ago</div>'
479 f'</div>'
480 )
482 return "".join(cards)
484 def _render_history(self) -> str:
485 """渲染审批历史。"""
486 history = self._queue.recent_history(limit=30)
487 if not history:
488 return "<p>No history yet</p>"
490 rows = ["<table style='width:100%;border-collapse:collapse;'>"]
491 rows.append(
492 "<tr style='background:#eee;'><th>Time</th><th>Action</th>"
493 "<th>Decision</th><th>By</th><th>Latency</th></tr>"
494 )
495 for h in reversed(history):
496 dt = datetime.fromtimestamp(h.decided_at).strftime("%H:%M:%S")
497 decision_color = "#4CAF50" if h.decision == ApprovalStatus.APPROVED else "#F44336"
498 rows.append(
499 f"<tr><td>{dt}</td>"
500 f"<td>{h.request.action[:40]}</td>"
501 f"<td style='color:{decision_color};font-weight:bold;'>"
502 f"{h.decision.value}</td>"
503 f"<td>{h.decided_by}</td>"
504 f"<td>{h.decided_at - h.request.created_at:.1f}s</td></tr>"
505 )
506 rows.append("</table>")
507 return "".join(rows)
509 def _render_policy_editor(self) -> str:
510 """渲染策略编辑器（占位，可扩展为交互式表单）。"""
511 return """
512 <div style="padding:16px;">
513 <h3>Approval Policy</h3>
514 <p>Configure auto-approval thresholds by risk level:</p>
515 <ul>
516 <li><b>Safe/Low:</b> Auto-approve</li>
517 <li><b>Medium:</b> Ask if confidence < 90%</li>
518 <li><b>High:</b> Always ask</li>
519 <li><b>Critical:</b> Always ask + require 2FA</li>
520 </ul>
521 <p><i>Interactive policy editor coming in v1.14.3</i></p>
522 </div>
523 """
525 @property
526 def queue(self) -> ApprovalQueue:
527 return self._queue
530# ── Agent Integration Bridge ────────────────
533class HITLUIBridge:
534 """连接 Agent HITL 引擎与 Gradio UI 的桥梁。
536 在 Agent loop 中使用:
537 bridge = HITLUIBridge(queue, agent_id)
538 bridge.send_approval_request(action="Delete file X", risk_level="high")
539 # UI 弹出审批卡片，Agent 在此阻塞等待结果
540 approved = await bridge.wait_for_decision(timeout=60)
541 """
543 def __init__(
544 self,
545 approval_queue: ApprovalQueue,
546 agent_id: str = "",
547 agent_name: str = "",
548 ):
549 self._queue = approval_queue
550 self.agent_id = agent_id
551 self.agent_name = agent_name
552 self._decision_events: Dict[str, asyncio.Event] = {}
553 self._decision_results: Dict[str, Tuple[bool, str]] = {}
555 async def send_approval_request(
556 self,
557 action: str,
558 details: str = "",
559 risk_level: str = "medium",
560 source_file: str = "",
561 timeout: float = 300.0,
562 metadata: Optional[Dict[str, Any]] = None,
563 ) -> Tuple[bool, str]:
564 """发送审批请求到 UI，阻塞等待用户决定。
566 Returns:
567 (approved: bool, reason: str)
568 """
569 event = asyncio.Event()
570 request_id = f"apr-{uuid.uuid4().hex[:8]}"
571 self._decision_events[request_id] = event
573 def on_decision(approved: bool, reason: str) -> None:
574 self._decision_results[request_id] = (approved, reason)
575 event.set()
577 request = ApprovalRequestUI(
578 request_id=request_id,
579 agent_name=self.agent_name,
580 action=action,
581 details=details,
582 risk_level=RiskLevelUI(risk_level),
583 expires_at=time.time() + timeout,
584 source_file=source_file,
585 metadata=metadata or {},
586 on_decision=on_decision,
587 )
589 self._queue.submit(request)
591 # 等待用户决定或超时
592 try:
593 await asyncio.wait_for(event.wait(), timeout=timeout)
594 result = self._decision_results.pop(request_id, (False, "timeout"))
595 self._decision_events.pop(request_id, None)
596 return result
597 except asyncio.TimeoutError:
598 self._queue.decide(request_id, False, "timeout")
599 self._decision_events.pop(request_id, None)
600 return (False, "timeout")
603# ── Quick Launch ────────────────────────────
606def create_hitl_dashboard(
607 port: int = 7860,
608 theme: str = "soft",
609 share: bool = False,
610) -> Tuple[ApprovalDashboard, ApprovalQueue]:
611 """一键创建并启动 HITL 审批面板。
613 Usage:
614 dashboard, queue = create_hitl_dashboard(port=7860)
615 # Agent 代码中:
616 bridge = HITLUIBridge(queue, agent_name="FileAgent")
617 approved, reason = await bridge.send_approval_request(
618 action="Delete 50 files in /tmp/",
619 risk_level="high",
620 )
621 """
622 queue = ApprovalQueue()
623 dashboard = ApprovalDashboard(
624 approval_queue=queue,
625 port=port,
626 theme=theme,
627 auto_launch=True,
628 )
630 # 在后台线程启动 Gradio
631 thread = threading.Thread(
632 target=dashboard.launch,
633 kwargs={"share": share},
634 daemon=True,
635 )
636 thread.start()
638 return dashboard, queue