Coverage for agentos/hitl/gradio_ui.py: 0%
266 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-07 00:45 +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 queue
25import threading
26import time
27import uuid
28from collections.abc import Callable
29from dataclasses import dataclass, field
30from datetime import datetime
31from enum import StrEnum
32from typing import (
33 Any,
34)
36# ── UI Data Models ──────────────────────────
39class ApprovalStatus(StrEnum):
40 PENDING = "pending"
41 APPROVED = "approved"
42 DENIED = "denied"
43 EXPIRED = "expired"
44 CANCELLED = "cancelled"
47class RiskLevelUI(StrEnum):
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: Callable | None = 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 {
148 "event": "new_request",
149 "request_id": request.request_id,
150 "action": request.action,
151 "risk_level": request.risk_level.value,
152 "pending_count": len(self._pending),
153 }
154 )
156 return request.request_id
158 def decide(self, request_id: str, approved: bool, reason: str = "") -> ApprovalHistory | None:
159 """处理审批决定。"""
160 with self._lock:
161 request = self._pending.pop(request_id, None)
163 if request is None:
164 return None
166 decision = ApprovalStatus.APPROVED if approved else ApprovalStatus.DENIED
167 history = ApprovalHistory(
168 request=request,
169 decision=decision,
170 reason=reason,
171 )
173 # Trigger callback
174 if request.on_decision:
175 try:
176 request.on_decision(approved, reason)
177 except Exception:
178 pass
180 with self._lock:
181 self._history.append(history)
183 self._notify_subscribers(
184 {
185 "event": "decision",
186 "request_id": request_id,
187 "approved": approved,
188 "reason": reason,
189 "pending_count": len(self._pending),
190 }
191 )
193 return history
195 def approve_all(self) -> int:
196 """批量批准所有待处理请求。"""
197 ids = list(self._pending.keys())
198 for rid in ids:
199 self.decide(rid, True, "batch_approve")
200 return len(ids)
202 def deny_all(self) -> int:
203 """批量拒绝所有待处理请求。"""
204 ids = list(self._pending.keys())
205 for rid in ids:
206 self.decide(rid, False, "batch_deny")
207 return len(ids)
209 def check_timeouts(self) -> int:
210 """检查并自动拒绝超时请求。"""
211 time.time()
212 expired_ids = []
213 with self._lock:
214 for rid, req in self._pending.items():
215 if req.is_expired:
216 expired_ids.append(rid)
218 for rid in expired_ids:
219 history = self.decide(rid, False, "timeout")
220 if history:
221 history.decided_by = "timeout"
223 return len(expired_ids)
225 @property
226 def pending_requests(self) -> list[ApprovalRequestUI]:
227 with self._lock:
228 return list(self._pending.values())
230 @property
231 def pending_count(self) -> int:
232 with self._lock:
233 return len(self._pending)
235 @property
236 def recent_history(self, limit: int = 50) -> list[ApprovalHistory]:
237 with self._lock:
238 return self._history[-limit:]
240 def subscribe(self, callback: Callable) -> None:
241 """注册 WebSocket 通知回调。"""
242 self._subscribers.append(callback)
244 def _notify_subscribers(self, data: dict) -> None:
245 for cb in self._subscribers:
246 try:
247 cb(data)
248 except Exception:
249 pass
252# ── Gradio Approval Dashboard ──────────────
255class ApprovalDashboard:
256 """Gradio 驱动的审批面板。
258 核心布局:
259 - 顶部: Agent 状态栏
260 - 左侧: 待审批队列
261 - 右侧: 审批详情 + 历史
262 - 底部: 批量操作按钮
264 Usage:
265 dashboard = ApprovalDashboard(queue, port=7860)
266 dashboard.launch()
267 """
269 def __init__(
270 self,
271 approval_queue: ApprovalQueue,
272 port: int = 7860,
273 title: str = "AgentOS — HITL Approval Dashboard",
274 theme: str = "soft",
275 auto_launch: bool = True,
276 ):
277 self._queue = approval_queue
278 self._port = port
279 self._title = title
280 self._theme = theme
281 self._auto_launch = auto_launch
282 self._app = None
283 self._agent_statuses: dict[str, AgentStatusSnapshot] = {}
284 self._selected_request_id: str = ""
286 def launch(self, share: bool = False) -> Any:
287 """启动 Gradio 面板。
289 返回 Gradio Blocks 实例，可在 Jupyter 中内嵌或独立运行。
290 """
291 try:
292 import gradio as gr
293 except ImportError:
294 raise ImportError(
295 "Gradio is required for ApprovalDashboard. " "Install with: pip install gradio>=4.0"
296 )
298 with gr.Blocks(title=self._title, theme=self._theme) as app:
299 self._app = app
300 self._build_ui(app)
302 if self._auto_launch:
303 app.launch(
304 server_port=self._port,
305 share=share,
306 prevent_thread_lock=False,
307 )
309 return app
311 def _build_ui(self, app: Any) -> None:
312 """构建完整 UI 布局。"""
313 import gradio as gr
315 # ── Header ──
316 gr.Markdown(f"# {self._title}\n" f"### Real-time Human-in-the-Loop Approval Dashboard")
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("Approve All", variant="primary", size="sm")
336 self._btn_deny_all = gr.Button("Deny All", variant="stop", size="sm")
338 # Right: Detail + History
339 with gr.Column(scale=2):
340 with gr.Tabs():
341 with gr.TabItem("Approval Detail"):
342 self._detail_view = gr.HTML(
343 value="<p>Select a request from the queue...</p>"
344 )
345 with gr.Row():
346 self._btn_approve = gr.Button("Approve", variant="primary")
347 self._btn_deny = gr.Button("Deny", variant="stop")
348 self._reason_input = gr.Textbox(
349 label="Reason (optional)",
350 placeholder="Why this decision...",
351 )
353 with gr.TabItem("History"):
354 self._history_view = gr.HTML(
355 value=self._render_history(),
356 every=3.0,
357 )
359 with gr.TabItem("Policy"):
360 self._policy_view = gr.HTML(value=self._render_policy_editor())
362 # ── Event Handlers ──
363 self._btn_approve.click(
364 fn=self._handle_approve,
365 inputs=[self._reason_input],
366 outputs=[self._detail_view, self._queue_list, self._history_view],
367 )
368 self._btn_deny.click(
369 fn=self._handle_deny,
370 inputs=[self._reason_input],
371 outputs=[self._detail_view, self._queue_list, self._history_view],
372 )
373 self._btn_approve_all.click(
374 fn=lambda: self._queue.approve_all(),
375 outputs=[],
376 )
377 self._btn_deny_all.click(
378 fn=lambda: self._queue.deny_all(),
379 outputs=[],
380 )
382 def _handle_approve(self, reason: str) -> tuple[str, str, str]:
383 if not self._selected_request_id:
384 return self._detail_view, self._queue_list, self._history_view
385 self._queue.decide(self._selected_request_id, True, reason)
386 self._selected_request_id = ""
387 return (
388 "<p>Select a request from the queue...</p>",
389 self._render_pending_queue(),
390 self._render_history(),
391 )
393 def _handle_deny(self, reason: str) -> tuple[str, str, str]:
394 if not self._selected_request_id:
395 return self._detail_view, self._queue_list, self._history_view
396 self._queue.decide(self._selected_request_id, False, reason)
397 self._selected_request_id = ""
398 return (
399 "<p>Select a request from the queue...</p>",
400 self._render_pending_queue(),
401 self._render_history(),
402 )
404 def update_agent_status(self, snapshot: AgentStatusSnapshot) -> None:
405 """更新 Agent 状态（由外部 Agent loop 调用）。"""
406 self._agent_statuses[snapshot.agent_id] = snapshot
408 def _render_agent_status_bar(self) -> str:
409 """渲染 Agent 状态栏 HTML。"""
410 if not self._agent_statuses:
411 return (
412 '<div style="padding:12px;background:#f0f0f0;border-radius:8px;">'
413 '<span style="color:#888;">No agents connected</span></div>'
414 )
416 rows = []
417 for agent_id, snap in self._agent_statuses.items():
418 status_color = {
419 "idle": "#4CAF50",
420 "running": "#2196F3",
421 "waiting_approval": "#FF9800",
422 "paused": "#9E9E9E",
423 "error": "#F44336",
424 }.get(snap.status, "#9E9E9E")
426 rows.append(
427 f'<div style="display:inline-block;margin:4px 8px;padding:8px 12px;'
428 f'background:#fff;border-radius:6px;border-left:4px solid {status_color};">'
429 f"<b>{snap.agent_name}</b> "
430 f'<span style="color:{status_color};">● {snap.status}</span> '
431 f"| {snap.current_task[:30]} "
432 f"| {snap.pending_approvals} pending"
433 f"</div>"
434 )
436 return (
437 '<div style="padding:12px;background:#f0f0f0;border-radius:8px;">'
438 + "".join(rows)
439 + "</div>"
440 )
442 def _render_pending_queue(self) -> str:
443 """渲染待处理队列 HTML。"""
444 pending = self._queue.pending_requests
445 if not pending:
446 return '<p style="color:#888;">No pending approvals</p>'
448 risk_colors = {
449 "safe": "#4CAF50",
450 "low": "#8BC34A",
451 "medium": "#FF9800",
452 "high": "#F44336",
453 "critical": "#B71C1C",
454 }
456 cards = []
457 for req in pending:
458 color = risk_colors.get(req.risk_level.value, "#999")
459 elapsed = int(req.elapsed_seconds)
460 cards.append(
461 f"<div onclick=\"selectRequest('{req.request_id}')\" "
462 f'style="cursor:pointer;margin:6px 0;padding:10px;'
463 f"background:#fff;border-radius:6px;"
464 f'border-left:4px solid {color};">'
465 f'<div style="font-weight:bold;">{req.action[:60]}</div>'
466 f'<div style="color:{color};font-size:0.85em;">'
467 f"Risk: {req.risk_level.value} | Agent: {req.agent_name} | "
468 f"{elapsed}s ago</div>"
469 f"</div>"
470 )
472 return "".join(cards)
474 def _render_history(self) -> str:
475 """渲染审批历史。"""
476 history = self._queue.recent_history(limit=30)
477 if not history:
478 return "<p>No history yet</p>"
480 rows = ["<table style='width:100%;border-collapse:collapse;'>"]
481 rows.append(
482 "<tr style='background:#eee;'><th>Time</th><th>Action</th>"
483 "<th>Decision</th><th>By</th><th>Latency</th></tr>"
484 )
485 for h in reversed(history):
486 dt = datetime.fromtimestamp(h.decided_at).strftime("%H:%M:%S")
487 decision_color = "#4CAF50" if h.decision == ApprovalStatus.APPROVED else "#F44336"
488 rows.append(
489 f"<tr><td>{dt}</td>"
490 f"<td>{h.request.action[:40]}</td>"
491 f"<td style='color:{decision_color};font-weight:bold;'>"
492 f"{h.decision.value}</td>"
493 f"<td>{h.decided_by}</td>"
494 f"<td>{h.decided_at - h.request.created_at:.1f}s</td></tr>"
495 )
496 rows.append("</table>")
497 return "".join(rows)
499 def _render_policy_editor(self) -> str:
500 """渲染策略编辑器（占位，可扩展为交互式表单）。"""
501 return """
502 <div style="padding:16px;">
503 <h3>Approval Policy</h3>
504 <p>Configure auto-approval thresholds by risk level:</p>
505 <ul>
506 <li><b>Safe/Low:</b> Auto-approve</li>
507 <li><b>Medium:</b> Ask if confidence < 90%</li>
508 <li><b>High:</b> Always ask</li>
509 <li><b>Critical:</b> Always ask + require 2FA</li>
510 </ul>
511 <p><i>Interactive policy editor coming in v1.14.3</i></p>
512 </div>
513 """
515 @property
516 def queue(self) -> ApprovalQueue:
517 return self._queue
520# ── Agent Integration Bridge ────────────────
523class HITLUIBridge:
524 """连接 Agent HITL 引擎与 Gradio UI 的桥梁。
526 在 Agent loop 中使用:
527 bridge = HITLUIBridge(queue, agent_id)
528 bridge.send_approval_request(action="Delete file X", risk_level="high")
529 # UI 弹出审批卡片，Agent 在此阻塞等待结果
530 approved = await bridge.wait_for_decision(timeout=60)
531 """
533 def __init__(
534 self,
535 approval_queue: ApprovalQueue,
536 agent_id: str = "",
537 agent_name: str = "",
538 ):
539 self._queue = approval_queue
540 self.agent_id = agent_id
541 self.agent_name = agent_name
542 self._decision_events: dict[str, asyncio.Event] = {}
543 self._decision_results: dict[str, tuple[bool, str]] = {}
545 async def send_approval_request(
546 self,
547 action: str,
548 details: str = "",
549 risk_level: str = "medium",
550 source_file: str = "",
551 timeout: float = 300.0,
552 metadata: dict[str, Any] | None = None,
553 ) -> tuple[bool, str]:
554 """发送审批请求到 UI，阻塞等待用户决定。
556 Returns:
557 (approved: bool, reason: str)
558 """
559 event = asyncio.Event()
560 request_id = f"apr-{uuid.uuid4().hex[:8]}"
561 self._decision_events[request_id] = event
563 def on_decision(approved: bool, reason: str) -> None:
564 self._decision_results[request_id] = (approved, reason)
565 event.set()
567 request = ApprovalRequestUI(
568 request_id=request_id,
569 agent_name=self.agent_name,
570 action=action,
571 details=details,
572 risk_level=RiskLevelUI(risk_level),
573 expires_at=time.time() + timeout,
574 source_file=source_file,
575 metadata=metadata or {},
576 on_decision=on_decision,
577 )
579 self._queue.submit(request)
581 # 等待用户决定或超时
582 try:
583 await asyncio.wait_for(event.wait(), timeout=timeout)
584 result = self._decision_results.pop(request_id, (False, "timeout"))
585 self._decision_events.pop(request_id, None)
586 return result
587 except TimeoutError:
588 self._queue.decide(request_id, False, "timeout")
589 self._decision_events.pop(request_id, None)
590 return (False, "timeout")
593# ── Quick Launch ────────────────────────────
596def create_hitl_dashboard(
597 port: int = 7860,
598 theme: str = "soft",
599 share: bool = False,
600) -> tuple[ApprovalDashboard, ApprovalQueue]:
601 """一键创建并启动 HITL 审批面板。
603 Usage:
604 dashboard, queue = create_hitl_dashboard(port=7860)
605 # Agent 代码中:
606 bridge = HITLUIBridge(queue, agent_name="FileAgent")
607 approved, reason = await bridge.send_approval_request(
608 action="Delete 50 files in /tmp/",
609 risk_level="high",
610 )
611 """
612 queue = ApprovalQueue()
613 dashboard = ApprovalDashboard(
614 approval_queue=queue,
615 port=port,
616 theme=theme,
617 auto_launch=True,
618 )
620 # 在后台线程启动 Gradio
621 thread = threading.Thread(
622 target=dashboard.launch,
623 kwargs={"share": share},
624 daemon=True,
625 )
626 thread.start()
628 return dashboard, queue