pengxiao
219232de74
以讲义内容为骨架迁移到标准目录格式： - officefile/ 主内容（12章 + 附录 + CC4SI补充） - dofile/ 代码示例（11个Python脚本） - data/ 图片资源 - output/ 生成输出（忽略） - Archive/ 归档旧目录（忽略） - .claude/skills/ 保留markdown-to-docx工具链 - .pandoc/ 保留CSL和本地化配置 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
825 lines
28 KiB
Python
825 lines
28 KiB
Python
"""
|
|
人机协同示例 (Human-in-the-Loop Example)
|
|
========================================
|
|
|
|
本示例展示如何在空间智能系统中实现人机协同工作模式。
|
|
人机协同 (HITL) 结合人类专家的领域知识和AI的计算能力，
|
|
实现更可靠的决策。
|
|
|
|
核心概念:
|
|
1. 主动学习 - AI主动请求人类帮助
|
|
2. 交互式决策 - 人机共同完成决策
|
|
3. 反馈收集 - 收集并整合人类反馈
|
|
4. 置信度估计 - AI评估自身确定性
|
|
5. 专业知识注入 - 将专家知识整合到系统中
|
|
|
|
应用场景:
|
|
- 空间数据标注与验证
|
|
- 复杂选址决策
|
|
- 应急响应规划
|
|
- 土地利用评估
|
|
|
|
作者: CC4SI 项目组
|
|
"""
|
|
|
|
import math
|
|
import json
|
|
from typing import List, Dict, Tuple, Optional, Any, Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from datetime import datetime
|
|
import random
|
|
|
|
|
|
# ============================================================================
|
|
# 协同模式枚举
|
|
# ============================================================================
|
|
|
|
class HITLMode(Enum):
|
|
"""人机协同模式"""
|
|
AUTOMATIC = "automatic" # 全自动模式
|
|
ADVISORY = "advisory" # 建议模式 (AI提供建议，人类决策)
|
|
INTERACTIVE = "interactive" # 交互模式 (人机共同决策)
|
|
SUPERVISED = "supervised" # 监督模式 (人类监督AI)
|
|
MANUAL = "manual" # 手动模式 (人类完全控制)
|
|
|
|
|
|
class ConfidenceLevel(Enum):
|
|
"""置信度级别"""
|
|
VERY_LOW = "very_low" # 0.0 - 0.3
|
|
LOW = "low" # 0.3 - 0.5
|
|
MEDIUM = "medium" # 0.5 - 0.7
|
|
HIGH = "high" # 0.7 - 0.9
|
|
VERY_HIGH = "very_high" # 0.9 - 1.0
|
|
|
|
|
|
class InteractionType(Enum):
|
|
"""交互类型"""
|
|
CONFIRMATION = "confirmation" # 确认请求
|
|
CLARIFICATION = "clarification" # 澄清请求
|
|
VALIDATION = "validation" # 验证请求
|
|
CORRECTION = "correction" # 纠正请求
|
|
RANKING = "ranking" # 排序请求
|
|
ANNOTATION = "annotation" # 标注请求
|
|
|
|
|
|
# ============================================================================
|
|
# 交互数据结构
|
|
# ============================================================================
|
|
|
|
@dataclass
|
|
class AIConfidence:
|
|
"""AI置信度"""
|
|
value: float # 0-1之间的值
|
|
reason: str = ""
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
@property
|
|
def level(self) -> ConfidenceLevel:
|
|
"""获取置信度级别"""
|
|
if self.value < 0.3:
|
|
return ConfidenceLevel.VERY_LOW
|
|
elif self.value < 0.5:
|
|
return ConfidenceLevel.LOW
|
|
elif self.value < 0.7:
|
|
return ConfidenceLevel.MEDIUM
|
|
elif self.value < 0.9:
|
|
return ConfidenceLevel.HIGH
|
|
else:
|
|
return ConfidenceLevel.VERY_HIGH
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Confidence({self.value:.2f}, {self.level.value})"
|
|
|
|
|
|
@dataclass
|
|
class HumanInput:
|
|
"""人类输入"""
|
|
interaction_type: InteractionType
|
|
response: Any
|
|
confidence: float = 1.0 # 人类对自己回答的置信度
|
|
timestamp: datetime = field(default_factory=datetime.now)
|
|
expert_id: str = "default_expert"
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class InteractionRequest:
|
|
"""交互请求"""
|
|
request_id: str
|
|
interaction_type: InteractionType
|
|
question: str
|
|
context: Dict[str, Any]
|
|
options: Optional[List[Any]] = None
|
|
ai_suggestion: Optional[Any] = None
|
|
ai_confidence: Optional[AIConfidence] = None
|
|
priority: int = 0 # 优先级 (0=普通, 1=重要, 2=紧急)
|
|
deadline: Optional[datetime] = None
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
# ============================================================================
|
|
# 决策建议
|
|
# ============================================================================
|
|
|
|
@dataclass
|
|
class DecisionProposal:
|
|
"""决策建议"""
|
|
proposal_id: str
|
|
decision: Any
|
|
reasoning: str
|
|
confidence: AIConfidence
|
|
alternatives: List[Any] = field(default_factory=list)
|
|
supporting_evidence: List[str] = field(default_factory=list)
|
|
caveats: List[str] = field(default_factory=list) # 警告/注意事项
|
|
requires_human_review: bool = False
|
|
timestamp: datetime = field(default_factory=datetime.now)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""转换为字典"""
|
|
return {
|
|
"proposal_id": self.proposal_id,
|
|
"decision": self.decision,
|
|
"reasoning": self.reasoning,
|
|
"confidence": self.confidence.value,
|
|
"alternatives": self.alternatives,
|
|
"supporting_evidence": self.supporting_evidence,
|
|
"caveats": self.caveats,
|
|
"requires_human_review": self.requires_human_review
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# 人类专家接口
|
|
# ============================================================================
|
|
|
|
class HumanExpert(ABC):
|
|
"""人类专家抽象接口"""
|
|
|
|
def __init__(self, expert_id: str, name: str = "", expertise: List[str] = None):
|
|
self.expert_id = expert_id
|
|
self.name = name or expert_id
|
|
self.expertise = expertise or []
|
|
|
|
@abstractmethod
|
|
def respond_to_request(self, request: InteractionRequest) -> HumanInput:
|
|
"""响应交互请求"""
|
|
pass
|
|
|
|
def can_handle(self, request: InteractionRequest) -> bool:
|
|
"""检查是否能处理请求"""
|
|
return True
|
|
|
|
def get_expertise_summary(self) -> str:
|
|
"""获取专长摘要"""
|
|
return f"{self.name}: {', '.join(self.expertise) if self.expertise else '通用'}"
|
|
|
|
|
|
class MockHumanExpert(HumanExpert):
|
|
"""
|
|
模拟人类专家 (用于演示)
|
|
|
|
在实际应用中，这会连接到真实的用户界面。
|
|
"""
|
|
|
|
def __init__(self, expert_id: str, name: str = "",
|
|
expertise: List[str] = None,
|
|
response_style: str = "balanced"):
|
|
super().__init__(expert_id, name, expertise)
|
|
self.response_style = response_style
|
|
self.response_log: List[Dict[str, Any]] = []
|
|
|
|
def respond_to_request(self, request: InteractionRequest) -> HumanInput:
|
|
"""模拟响应请求"""
|
|
# 记录请求
|
|
self.response_log.append({
|
|
"request_id": request.request_id,
|
|
"type": request.interaction_type.value,
|
|
"question": request.question,
|
|
"timestamp": datetime.now()
|
|
})
|
|
|
|
# 根据不同类型生成响应
|
|
if request.interaction_type == InteractionType.CONFIRMATION:
|
|
# 确认请求 - 模拟基于置信度的决策
|
|
if request.ai_confidence and request.ai_confidence.value > 0.7:
|
|
# 高置信度时倾向于接受AI建议
|
|
response = "accept" if random.random() > 0.2 else "reject"
|
|
else:
|
|
# 低置信度时更谨慎
|
|
response = "accept" if random.random() > 0.5 else "reject"
|
|
|
|
return HumanInput(
|
|
interaction_type=request.interaction_type,
|
|
response=response,
|
|
confidence=0.8
|
|
)
|
|
|
|
elif request.interaction_type == InteractionType.VALIDATION:
|
|
# 验证请求
|
|
is_valid = random.random() > 0.3 # 70%概率验证通过
|
|
return HumanInput(
|
|
interaction_type=request.interaction_type,
|
|
response=is_valid,
|
|
confidence=0.9,
|
|
metadata={"comment": "看起来正确" if is_valid else "需要修正"}
|
|
)
|
|
|
|
elif request.interaction_type == InteractionType.RANKING:
|
|
# 排序请求
|
|
if request.options:
|
|
# 随机打乱选项作为人类排序
|
|
shuffled = request.options.copy()
|
|
random.shuffle(shuffled)
|
|
return HumanInput(
|
|
interaction_type=request.interaction_type,
|
|
response=shuffled,
|
|
confidence=0.7
|
|
)
|
|
|
|
elif request.interaction_type == InteractionType.ANNOTATION:
|
|
# 标注请求
|
|
return HumanInput(
|
|
interaction_type=request.interaction_type,
|
|
response={
|
|
"label": random.choice(["高价值", "中价值", "低价值"]),
|
|
"notes": "基于现场评估"
|
|
},
|
|
confidence=0.75
|
|
)
|
|
|
|
# 默认响应
|
|
return HumanInput(
|
|
interaction_type=request.interaction_type,
|
|
response="acknowledged",
|
|
confidence=0.5
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# 人机协同系统
|
|
# ============================================================================
|
|
|
|
class HITLSystem:
|
|
"""
|
|
人机协同系统
|
|
|
|
管理AI与人类专家之间的交互。
|
|
"""
|
|
|
|
def __init__(self, name: str = "HITL系统",
|
|
default_mode: HITLMode = HITLMode.INTERACTIVE,
|
|
confidence_threshold: float = 0.7):
|
|
"""
|
|
初始化HITL系统
|
|
|
|
Args:
|
|
name: 系统名称
|
|
default_mode: 默认协同模式
|
|
confidence_threshold: 请求人类帮助的置信度阈值
|
|
"""
|
|
self.name = name
|
|
self.current_mode = default_mode
|
|
self.confidence_threshold = confidence_threshold
|
|
|
|
# 注册的专家
|
|
self.experts: Dict[str, HumanExpert] = {}
|
|
|
|
# 待处理的请求队列
|
|
self.pending_requests: List[InteractionRequest] = []
|
|
|
|
# 交互历史
|
|
self.interaction_history: List[Dict[str, Any]] = []
|
|
|
|
# 统计信息
|
|
self.stats = {
|
|
"total_requests": 0,
|
|
"auto_resolved": 0,
|
|
"human_resolved": 0,
|
|
"human_acceptance_rate": 0.0
|
|
}
|
|
|
|
print(f"[{self.name}] 初始化完成")
|
|
print(f" 模式: {default_mode.value}")
|
|
print(f" 置信度阈值: {confidence_threshold}")
|
|
|
|
def register_expert(self, expert: HumanExpert) -> None:
|
|
"""注册人类专家"""
|
|
self.experts[expert.expert_id] = expert
|
|
print(f"[专家注册] {expert.get_expertise_summary()}")
|
|
|
|
def set_mode(self, mode: HITLMode) -> None:
|
|
"""设置协同模式"""
|
|
self.current_mode = mode
|
|
print(f"[模式切换] {mode.value}")
|
|
|
|
def make_decision(self, proposal: DecisionProposal,
|
|
auto_threshold: float = None) -> Any:
|
|
"""
|
|
做出决策 (带人机协同)
|
|
|
|
Args:
|
|
proposal: AI的决策建议
|
|
auto_threshold: 自动决策的置信度阈值
|
|
|
|
Returns:
|
|
最终决策
|
|
"""
|
|
threshold = auto_threshold or self.confidence_threshold
|
|
|
|
# 根据模式和置信度决定是否需要人类介入
|
|
needs_human = self._needs_human_intervention(proposal, threshold)
|
|
|
|
if not needs_human:
|
|
# 自动决策
|
|
self.stats["auto_resolved"] += 1
|
|
self._record_interaction(proposal, None, "automatic")
|
|
return proposal.decision
|
|
|
|
# 请求人类帮助
|
|
return self._request_human_input(proposal)
|
|
|
|
def _needs_human_intervention(self, proposal: DecisionProposal,
|
|
threshold: float) -> bool:
|
|
"""判断是否需要人类介入"""
|
|
# 检查强制人工审查标记
|
|
if proposal.requires_human_review:
|
|
return True
|
|
|
|
# 检查置信度
|
|
if proposal.confidence.value < threshold:
|
|
return True
|
|
|
|
# 根据模式判断
|
|
if self.current_mode == HITLMode.MANUAL:
|
|
return True
|
|
elif self.current_mode == HITLMode.SUPERVISED:
|
|
return True
|
|
elif self.current_mode == HITLMode.AUTOMATIC:
|
|
return False
|
|
elif self.current_mode == HITLMode.INTERACTIVE:
|
|
# 交互模式下，低置信度需要人类
|
|
return proposal.confidence.value < 0.8
|
|
elif self.current_mode == HITLMode.ADVISORY:
|
|
# 建议模式下，总是需要人类确认
|
|
return True
|
|
|
|
return False
|
|
|
|
def _request_human_input(self, proposal: DecisionProposal) -> Any:
|
|
"""请求人类输入"""
|
|
# 创建交互请求
|
|
request = InteractionRequest(
|
|
request_id=f"req_{len(self.interaction_history)}",
|
|
interaction_type=InteractionType.CONFIRMATION,
|
|
question=f"请确认AI建议: {proposal.reasoning}",
|
|
context={"proposal_id": proposal.proposal_id},
|
|
options=["accept", "reject", "modify"],
|
|
ai_suggestion=proposal.decision,
|
|
ai_confidence=proposal.confidence
|
|
)
|
|
|
|
self.pending_requests.append(request)
|
|
self.stats["total_requests"] += 1
|
|
|
|
# 选择专家
|
|
expert = self._select_expert(request)
|
|
if not expert:
|
|
print("警告: 没有可用的专家，使用AI建议")
|
|
return proposal.decision
|
|
|
|
# 获取专家响应
|
|
print(f"\n[人类交互] 向专家 {expert.name} 请求确认...")
|
|
print(f" AI建议: {proposal.decision}")
|
|
print(f" 置信度: {proposal.confidence}")
|
|
print(f" 理由: {proposal.reasoning}")
|
|
|
|
human_input = expert.respond_to_request(request)
|
|
|
|
print(f" 专家响应: {human_input.response}")
|
|
|
|
# 处理响应
|
|
result = self._process_human_response(proposal, human_input)
|
|
|
|
# 记录交互
|
|
self._record_interaction(proposal, human_input, "human_assisted")
|
|
|
|
# 清理请求
|
|
if request in self.pending_requests:
|
|
self.pending_requests.remove(request)
|
|
|
|
self.stats["human_resolved"] += 1
|
|
|
|
return result
|
|
|
|
def _select_expert(self, request: InteractionRequest) -> Optional[HumanExpert]:
|
|
"""选择合适的专家"""
|
|
if not self.experts:
|
|
return None
|
|
|
|
# 简单实现: 返回第一个可用的专家
|
|
for expert in self.experts.values():
|
|
if expert.can_handle(request):
|
|
return expert
|
|
|
|
return None
|
|
|
|
def _process_human_response(self, proposal: DecisionProposal,
|
|
human_input: HumanInput) -> Any:
|
|
"""处理人类响应"""
|
|
if human_input.interaction_type == InteractionType.CONFIRMATION:
|
|
if human_input.response == "accept":
|
|
# 接受AI建议
|
|
return proposal.decision
|
|
elif human_input.response == "reject":
|
|
# 拒绝AI建议，返回次优选项
|
|
if proposal.alternatives:
|
|
return proposal.alternatives[0]
|
|
return None
|
|
elif human_input.response == "modify":
|
|
# 需要修改 (简化: 返回原建议)
|
|
return proposal.decision
|
|
|
|
return human_input.response
|
|
|
|
def _record_interaction(self, proposal: DecisionProposal,
|
|
human_input: Optional[HumanInput],
|
|
resolution_type: str) -> None:
|
|
"""记录交互"""
|
|
record = {
|
|
"proposal_id": proposal.proposal_id,
|
|
"timestamp": datetime.now(),
|
|
"ai_confidence": proposal.confidence.value,
|
|
"human_input": human_input.response if human_input else None,
|
|
"resolution_type": resolution_type
|
|
}
|
|
self.interaction_history.append(record)
|
|
|
|
def request_annotation(self, item: Any, context: Dict[str, Any] = None) -> Any:
|
|
"""请求人类标注"""
|
|
request = InteractionRequest(
|
|
request_id=f"annotate_{len(self.interaction_history)}",
|
|
interaction_type=InteractionType.ANNOTATION,
|
|
question=f"请对以下项目进行标注: {item}",
|
|
context=context or {},
|
|
ai_suggestion=item
|
|
)
|
|
|
|
expert = self._select_expert(request)
|
|
if not expert:
|
|
return None
|
|
|
|
return expert.respond_to_request(request)
|
|
|
|
def request_validation(self, item: Any, context: Dict[str, Any] = None) -> bool:
|
|
"""请求人类验证"""
|
|
request = InteractionRequest(
|
|
request_id=f"validate_{len(self.interaction_history)}",
|
|
interaction_type=InteractionType.VALIDATION,
|
|
question=f"以下内容是否正确: {item}",
|
|
context=context or {},
|
|
ai_suggestion=item
|
|
)
|
|
|
|
expert = self._select_expert(request)
|
|
if not expert:
|
|
return True # 默认有效
|
|
|
|
response = expert.respond_to_request(request)
|
|
return response.response if isinstance(response.response, bool) else True
|
|
|
|
def get_statistics(self) -> Dict[str, Any]:
|
|
"""获取统计信息"""
|
|
stats = self.stats.copy()
|
|
stats["pending_requests"] = len(self.pending_requests)
|
|
stats["total_interactions"] = len(self.interaction_history)
|
|
|
|
# 计算接受率
|
|
human_interactions = [i for i in self.interaction_history
|
|
if i["resolution_type"] == "human_assisted"]
|
|
if human_interactions:
|
|
accepted = sum(1 for i in human_interactions
|
|
if i["human_input"] == "accept")
|
|
stats["human_acceptance_rate"] = accepted / len(human_interactions)
|
|
|
|
return stats
|
|
|
|
def print_statistics(self) -> None:
|
|
"""打印统计信息"""
|
|
stats = self.get_statistics()
|
|
|
|
print(f"\n{self.name} 统计信息:")
|
|
print("-" * 50)
|
|
print(f"总请求数: {stats['total_requests']}")
|
|
print(f"自动解决: {stats['auto_resolved']}")
|
|
print(f"人类协助: {stats['human_resolved']}")
|
|
print(f"待处理请求: {stats['pending_requests']}")
|
|
print(f"人类接受率: {stats['human_acceptance_rate']:.2%}")
|
|
print("-" * 50)
|
|
|
|
|
|
# ============================================================================
|
|
# 空间决策HITL系统
|
|
# ============================================================================
|
|
|
|
class SpatialDecisionHITL(HITLSystem):
|
|
"""
|
|
空间决策人机协同系统
|
|
|
|
专门用于空间决策场景的HITL实现。
|
|
"""
|
|
|
|
def __init__(self, confidence_threshold: float = 0.7):
|
|
super().__init__(
|
|
name="空间决策HITL系统",
|
|
default_mode=HITLMode.INTERACTIVE,
|
|
confidence_threshold=confidence_threshold
|
|
)
|
|
|
|
def analyze_site_suitability(self, site_data: Dict[str, Any]) -> DecisionProposal:
|
|
"""
|
|
分析场地适宜性
|
|
|
|
Args:
|
|
site_data: 场地数据
|
|
|
|
Returns:
|
|
决策建议
|
|
"""
|
|
# 简化的适宜性评分
|
|
score = self._calculate_suitability_score(site_data)
|
|
|
|
# 确定置信度
|
|
confidence = self._assess_confidence(site_data, score)
|
|
|
|
# 生成建议
|
|
if score > 0.7:
|
|
decision = "highly_suitable"
|
|
reasoning = f"综合评分 {score:.2f} 较高，适宜开发"
|
|
elif score > 0.5:
|
|
decision = "moderately_suitable"
|
|
reasoning = f"综合评分 {score:.2f} 中等，需谨慎评估"
|
|
else:
|
|
decision = "not_suitable"
|
|
reasoning = f"综合评分 {score:.2f} 较低，不建议开发"
|
|
|
|
# 检查注意事项
|
|
caveats = []
|
|
if site_data.get("environmental_risk", 0) > 0.6:
|
|
caveats.append("存在环境风险")
|
|
if site_data.get("infrastructure_score", 1) < 0.4:
|
|
caveats.append("基础设施不足")
|
|
|
|
# 低置信度时标记需要人工审查
|
|
requires_review = confidence.value < 0.6 or len(caveats) > 0
|
|
|
|
return DecisionProposal(
|
|
proposal_id=f"suitability_{random.randint(1000, 9999)}",
|
|
decision=decision,
|
|
reasoning=reasoning,
|
|
confidence=confidence,
|
|
alternatives=["moderately_suitable", "not_suitable"]
|
|
if decision != "not_suitable" else ["moderately_suitable", "highly_suitable"],
|
|
supporting_evidence=[
|
|
f"评分: {score:.2f}",
|
|
f"环境因子: {site_data.get('environmental_score', 0):.2f}",
|
|
f"经济因子: {site_data.get('economic_score', 0):.2f}"
|
|
],
|
|
caveats=caveats,
|
|
requires_human_review=requires_review
|
|
)
|
|
|
|
def _calculate_suitability_score(self, site_data: Dict[str, Any]) -> float:
|
|
"""计算适宜性评分"""
|
|
env = site_data.get("environmental_score", 0.5)
|
|
econ = site_data.get("economic_score", 0.5)
|
|
social = site_data.get("social_score", 0.5)
|
|
infra = site_data.get("infrastructure_score", 0.5)
|
|
|
|
# 加权平均
|
|
return 0.3 * env + 0.3 * econ + 0.2 * social + 0.2 * infra
|
|
|
|
def _assess_confidence(self, site_data: Dict[str, Any],
|
|
score: float) -> AIConfidence:
|
|
"""评估置信度"""
|
|
# 检查数据完整性
|
|
has_all_data = all(k in site_data for k in [
|
|
"environmental_score", "economic_score",
|
|
"social_score", "infrastructure_score"
|
|
])
|
|
|
|
if not has_all_data:
|
|
return AIConfidence(
|
|
value=0.4,
|
|
reason="数据不完整"
|
|
)
|
|
|
|
# 检查数据质量
|
|
data_quality = site_data.get("data_quality", 0.8)
|
|
confidence = data_quality * 0.9
|
|
|
|
# 检查是否有冲突因素
|
|
if site_data.get("environmental_risk", 0) > 0.7:
|
|
confidence *= 0.7 # 降低置信度
|
|
|
|
return AIConfidence(
|
|
value=min(confidence, 0.95),
|
|
reason="基于数据质量和完整性评估"
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# 主程序
|
|
# ============================================================================
|
|
|
|
def main():
|
|
"""主程序 - 演示人机协同的使用"""
|
|
|
|
print("="*70)
|
|
print("人机协同示例演示")
|
|
print("="*70)
|
|
|
|
random.seed(42)
|
|
|
|
# ========================================================================
|
|
# 1. 创建HITL系统
|
|
# ========================================================================
|
|
print("\n[步骤 1] 创建人机协同系统")
|
|
print("-" * 50)
|
|
|
|
hitl_system = SpatialDecisionHITL(confidence_threshold=0.7)
|
|
|
|
# 注册专家
|
|
expert1 = MockHumanExpert(
|
|
expert_id="expert_001",
|
|
name="张工程师",
|
|
expertise=["环境影响评估", "基础设施规划"],
|
|
response_style="conservative"
|
|
)
|
|
expert2 = MockHumanExpert(
|
|
expert_id="expert_002",
|
|
name="李规划师",
|
|
expertise=["经济效益分析", "社会影响评估"],
|
|
response_style="balanced"
|
|
)
|
|
|
|
hitl_system.register_expert(expert1)
|
|
hitl_system.register_expert(expert2)
|
|
|
|
# ========================================================================
|
|
# 2. 场景1: 高置信度自动决策
|
|
# ========================================================================
|
|
print("\n[场景 1] 高置信度 - 自动决策")
|
|
print("-" * 50)
|
|
|
|
site1 = {
|
|
"environmental_score": 0.85,
|
|
"economic_score": 0.90,
|
|
"social_score": 0.88,
|
|
"infrastructure_score": 0.92,
|
|
"data_quality": 0.95,
|
|
"environmental_risk": 0.1
|
|
}
|
|
|
|
proposal1 = hitl_system.analyze_site_suitability(site1)
|
|
print(f"\nAI分析结果:")
|
|
print(f" 建议: {proposal1.decision}")
|
|
print(f" 理由: {proposal1.reasoning}")
|
|
print(f" 置信度: {proposal1.confidence}")
|
|
|
|
decision1 = hitl_system.make_decision(proposal1)
|
|
print(f"\n最终决策: {decision1} (自动)")
|
|
print(f" → 置信度高，无需人工介入")
|
|
|
|
# ========================================================================
|
|
# 3. 场景2: 低置信度请求人类帮助
|
|
# ========================================================================
|
|
print("\n\n[场景 2] 低置信度 - 请求人类确认")
|
|
print("-" * 50)
|
|
|
|
site2 = {
|
|
"environmental_score": 0.45, # 环境评分低
|
|
"economic_score": 0.85, # 但经济评分高
|
|
"social_score": 0.60,
|
|
"infrastructure_score": 0.50,
|
|
"data_quality": 0.70,
|
|
"environmental_risk": 0.65 # 存在环境风险
|
|
}
|
|
|
|
proposal2 = hitl_system.analyze_site_suitability(site2)
|
|
print(f"\nAI分析结果:")
|
|
print(f" 建议: {proposal2.decision}")
|
|
print(f" 理由: {proposal2.reasoning}")
|
|
print(f" 置信度: {proposal2.confidence}")
|
|
print(f" 注意事项: {', '.join(proposal2.caveats)}")
|
|
|
|
decision2 = hitl_system.make_decision(proposal2)
|
|
print(f"\n最终决策: {decision2} (人工协助)")
|
|
print(f" → 置信度低且存在注意事项，请求专家确认")
|
|
|
|
# ========================================================================
|
|
# 4. 场景3: 批量决策
|
|
# ========================================================================
|
|
print("\n\n[场景 3] 批量场地评估")
|
|
print("-" * 50)
|
|
|
|
sites = []
|
|
for i in range(5):
|
|
site = {
|
|
"environmental_score": random.uniform(0.3, 0.95),
|
|
"economic_score": random.uniform(0.3, 0.95),
|
|
"social_score": random.uniform(0.3, 0.95),
|
|
"infrastructure_score": random.uniform(0.3, 0.95),
|
|
"data_quality": random.uniform(0.5, 0.95),
|
|
"environmental_risk": random.uniform(0.0, 0.8)
|
|
}
|
|
sites.append(site)
|
|
|
|
results = []
|
|
for i, site in enumerate(sites, 1):
|
|
proposal = hitl_system.analyze_site_suitability(site)
|
|
decision = hitl_system.make_decision(proposal)
|
|
|
|
results.append({
|
|
"site": i,
|
|
"decision": decision,
|
|
"confidence": proposal.confidence.value,
|
|
"auto": proposal.confidence.value >= hitl_system.confidence_threshold
|
|
})
|
|
|
|
print("\n批量评估结果:")
|
|
print(f"{'场地':<6} {'决策':<20} {'置信度':<10} {'模式':<10}")
|
|
print("-" * 50)
|
|
for r in results:
|
|
mode = "自动" if r["auto"] else "人工"
|
|
print(f"{r['site']:<6} {r['decision']:<20} {r['confidence']:<10.2f} {mode:<10}")
|
|
|
|
# ========================================================================
|
|
# 5. 场景4: 数据标注
|
|
# ========================================================================
|
|
print("\n\n[场景 4] 数据标注")
|
|
print("-" * 50)
|
|
|
|
unlabeled_items = [
|
|
{"coordinates": (120.5, 30.2), "features": "residential"},
|
|
{"coordinates": (121.0, 30.5), "features": "commercial"},
|
|
{"coordinates": (120.8, 30.0), "features": "industrial"}
|
|
]
|
|
|
|
for item in unlabeled_items:
|
|
annotation = hitl_system.request_annotation(
|
|
item,
|
|
context={"task": "land_use_classification"}
|
|
)
|
|
if annotation:
|
|
print(f"\n标注 {item['features']}:")
|
|
print(f" 标签: {annotation.response.get('label')}")
|
|
print(f" 备注: {annotation.response.get('notes')}")
|
|
|
|
# ========================================================================
|
|
# 6. 场景5: 模式切换
|
|
# ========================================================================
|
|
print("\n\n[场景 5] 模式切换对比")
|
|
print("-" * 50)
|
|
|
|
test_site = {
|
|
"environmental_score": 0.70,
|
|
"economic_score": 0.75,
|
|
"social_score": 0.68,
|
|
"infrastructure_score": 0.72,
|
|
"data_quality": 0.85,
|
|
"environmental_risk": 0.3
|
|
}
|
|
|
|
proposal = hitl_system.analyze_site_suitability(test_site)
|
|
print(f"\nAI分析: 置信度 = {proposal.confidence.value:.2f}")
|
|
|
|
# 尝试不同模式
|
|
for mode in [HITLMode.AUTOMATIC, HITLMode.INTERACTIVE, HITLMode.MANUAL]:
|
|
hitl_system.set_mode(mode)
|
|
decision = hitl_system.make_decision(proposal)
|
|
mode_name = {
|
|
HITLMode.AUTOMATIC: "全自动",
|
|
HITLMode.INTERACTIVE: "交互式",
|
|
HITLMode.MANUAL: "手动"
|
|
}[mode]
|
|
print(f" {mode_name}: {decision}")
|
|
|
|
# 恢复默认模式
|
|
hitl_system.set_mode(HITLMode.INTERACTIVE)
|
|
|
|
# ========================================================================
|
|
# 7. 统计信息
|
|
# ========================================================================
|
|
print("\n\n[步骤 7] 系统统计")
|
|
print("-" * 50)
|
|
hitl_system.print_statistics()
|
|
|
|
print("\n" + "="*70)
|
|
print("演示完成!")
|
|
print("="*70)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|