企业官网建设流程全解析

KiwiQ AI人机交互(HITL)功能详解：实时WebSocket流式审批与干预

【免费下载链接】kiwiqProduction-grade multi-agent orchestration platform - JSON-defined agents, multi-tier memory, and built-in observability. Battle-tested on 200+ enterprise AI agents. Now fully open-sourced (prod at https://kiwiq.ai).项目地址: https://gitcode.com/gh_mirrors/ki/kiwiq

KiwiQ是一款生产级的多智能体编排平台，为企业和开发者提供强大的人工智能工作流管理能力。在当今AI自动化流程中，人机交互(HITL)功能已成为确保关键决策质量的核心需求。KiwiQ通过创新的实时WebSocket流式审批与干预机制，让人类专家能够在AI工作流的关键节点进行精准干预，实现智能与人工的完美协作。

🔍 为什么需要HITL人机交互？

在复杂的AI工作流中，完全自动化有时会带来风险或不确定性。KiwiQ的HITL功能解决了以下核心痛点：

• 关键决策需要人工审核：内容发布、财务审批、敏感操作等场景
• AI生成内容的质量把关：确保输出符合企业标准和品牌调性
• 实时异常处理：在AI工作流遇到不确定情况时及时干预
• 合规性要求：满足监管和合规的人工审核要求

🚀 KiwiQ HITL核心架构解析

KiwiQ的人机交互系统基于现代化的微服务架构设计，主要包含以下组件：

1.工作流引擎与HITL节点集成

在workflow_runner_node_guide.md中，KiwiQ定义了完整的HITL工作流节点支持：

# 当子工作流进入HITL状态时，节点会输出以下信息 { "status": "waiting_hitl", "hitl_job_id": "123e4567-e89b-12d3-a456-426614174001", "hitl_request_schema": { "type": "object", "properties": { "user_decision": {"type": "string", "enum": ["approve", "reject", "modify"]} }, "required": ["user_decision"] }, "hitl_request_details": { "prompt": "请审核生成的内容并提供审批决定", "context": "内容生成工作流已完成，等待编辑审核" } }

2.实时WebSocket通知系统

KiwiQ通过websockets.py实现了高效的实时通信：

• 运行特定通知：/ws/runs/{run_id}- 订阅特定工作流运行的事件
• 通用用户通知：/ws/notifications- 接收所有HITL任务通知
• 双向实时通信：支持前端实时推送审批请求，后端接收审批结果

3.HITL任务管理API

通过notification_hitl_client.py，开发者可以轻松管理HITL任务：

# 获取待处理的HITL任务 hitl_jobs = await hitl_tester.list_hitl_jobs( limit=5, status=HITLJobStatus.PENDING, pending_only=True ) # 获取特定HITL任务详情 job_details = await hitl_tester.get_hitl_job_details(job_id) # 提交人工审批结果 response = await hitl_tester.submit_hitl_decision( job_id=job_id, decision={"user_decision": "approve", "feedback": "内容质量优秀"} )

🎯 HITL工作流实战指南

第1步：定义需要人工审批的工作流节点

在KiwiQ中，您可以在工作流中标记需要人工干预的节点：

{ "workflow_id": "content-review-workflow", "nodes": [ { "id": "content-generation", "type": "ai_content_generator" }, { "id": "human-approval", "type": "hitl_node", "config": { "require_approval": true, "approval_schema": { "user_decision": {"type": "string", "enum": ["approve", "reject"]}, "feedback": {"type": "string"} } } }, { "id": "publishing", "type": "content_publisher", "condition": "{{human_approval.decision == 'approve'}}" } ] }

第2步：配置WebSocket客户端连接

使用websocket_client.py建立实时连接：

# 建立WebSocket连接接收实时通知 ws_client = WebSocketTestClient(auth_client) # 连接特定工作流运行 run_ws = ws_client.connect_to_run_notifications( run_id=run_id, on_message=handle_hitl_notification, on_error=handle_websocket_error ) # 启动WebSocket监听 ws_client.run_websocket(run_ws, reconnect_interval=5)

第3步：处理实时HITL通知

当工作流到达HITL节点时，前端会收到实时通知：

def handle_hitl_notification(ws, message): data = json.loads(message) if data.get("event_type") == "hitl_request": # 显示审批界面 show_approval_ui( job_id=data["hitl_job_id"], request_details=data["hitl_request_details"], schema=data["hitl_request_schema"] )

第4步：提交人工决策

用户在前端完成审批后，通过API提交决策：

# 提交审批结果 response = await hitl_client.submit_hitl_response( job_id=hitl_job_id, user_input={ "user_decision": "approve", "feedback": "内容符合品牌标准，建议增加SEO关键词", "modifications": { "add_keywords": ["AI自动化", "工作流管理"] } } )

第5步：工作流自动恢复

收到人工决策后，工作流引擎会自动恢复执行：

# 工作流根据审批结果继续执行 if decision["user_decision"] == "approve": # 执行批准后的流程 await workflow.resume_with_inputs(decision) elif decision["user_decision"] == "reject": # 执行拒绝后的流程 await workflow.cancel_with_reason(decision["feedback"])

📊 HITL状态管理与监控

KiwiQ提供了完整的HITL状态管理机制：

状态流转图

工作流执行 → 遇到HITL节点 → 状态变为WAITING_HITL ↓ 创建HITL任务记录 ↓ WebSocket推送通知到前端 ↓ 用户在前端查看并提交决策 ↓ 验证决策符合schema ↓ 工作流恢复执行 → 继续后续节点

HITL任务状态

• PENDING：等待人工处理
• COMPLETED：已完成审批
• CANCELLED：已取消
• EXPIRED：已超时

🔧 高级配置选项

1.审批超时设置

{ "hitl_config": { "timeout_seconds": 3600, // 1小时超时 "timeout_action": "auto_reject", // 超时自动拒绝 "reminder_intervals": [300, 900] // 5分钟和15分钟提醒 } }

2.多级审批流程

{ "approval_chain": [ { "role": "editor", "required": true, "parallel": false }, { "role": "legal", "required": true, "parallel": false }, { "role": "executive", "required": false, "parallel": true } ] }

3.条件性HITL触发

{ "hitl_conditions": [ { "condition": "{{content_length > 1000}}", "required_approvers": ["senior_editor"] }, { "condition": "{{contains_sensitive_topics}}", "required_approvers": ["legal", "compliance"] } ] }

🛠️ 故障排除与最佳实践

常见问题解决方案

1. WebSocket连接断开

   • 启用自动重连机制
   • 配置心跳包保持连接
   • 实现连接状态监控

2. HITL任务超时处理

   • 设置合理的超时时间
   • 配置超时后的默认行为
   • 实现超时通知机制

3. 审批流程卡住

   • 添加审批提醒功能
   • 支持审批委派
   • 实现强制跳过机制

性能优化建议

• 批量处理：对于多个相似的HITL任务，支持批量审批
• 缓存策略：缓存常用的审批模板和schema
• 异步处理：将非关键审批操作异步化
• 监控告警：设置HITL任务积压告警

🚀 实际应用场景

场景1：内容营销自动化

内容生成 → AI质量检查 → HITL人工审核 → 多平台发布 ↑ 编辑审核内容质量 确保品牌一致性

场景2：金融风控审批

交易分析 → 风险评分 → HITL人工复核 → 执行交易 ↑ 风控专家审核高风险交易 确保合规性和安全性

场景3：客户服务自动化

客户查询 → AI回复生成 → HITL服务质量检查 → 发送给客户 ↑ 客服主管审核回复质量 确保客户满意度

📈 监控与数据分析

KiwiQ的HITL系统提供完整的监控指标：

• 平均审批时间：监控审批效率
• 审批通过率：评估AI输出质量
• 人工干预频率：优化自动化程度
• 超时任务比例：改进流程设计

🎉 总结

KiwiQ的HITL人机交互功能通过实时WebSocket流式审批机制，实现了AI自动化与人工监督的完美平衡。无论是内容审核、金融风控还是客户服务，KiwiQ都能提供灵活、可靠的人机协作解决方案。

通过简单的配置和强大的API，您可以轻松将人工审批集成到任何AI工作流中，确保关键决策的质量和合规性。立即体验KiwiQ，开启智能与人工协作的新篇章！ 🚀

了解更多技术细节，请参考官方文档：docs/official.md 和AI功能源码：plugins/ai/

【免费下载链接】kiwiqProduction-grade multi-agent orchestration platform - JSON-defined agents, multi-tier memory, and built-in observability. Battle-tested on 200+ enterprise AI agents. Now fully open-sourced (prod at https://kiwiq.ai).项目地址: https://gitcode.com/gh_mirrors/ki/kiwiq