端云协同的AI Agent硬件助手架构方案 #

// IntentRouter - 端侧轻量意图分类与路由决策
type IntentRouter struct {
    model    *SLM           // 端侧小模型
    rules    *RuleEngine    // 规则引擎兜底
    config   *RouterConfig  // 路由阈值配置
}

func (r *IntentRouter) Route(text string) *RouteDecision {
    // 1. 规则优先（高确定性意图直接匹配，零推理成本）
    if ruleMatch := r.rules.Match(text); ruleMatch != nil {
        if ruleMatch.Confidence > 0.95 {
            return &RouteDecision{
                Target:  ExecutionTargetLocal,
                Action:  ruleMatch.Action,
                Reason:  "rule_match",
            }
        }
    }
    
    // 2. SLM推理（0.3B参数，INT4量化，内存占用~180MB）
    result := r.model.Predict(text)
    
    // 3. 路由决策
    if result.Confidence >= r.config.LocalThreshold {
        // 高置信度：本地执行
        return &RouteDecision{
            Target:  ExecutionTargetLocal,
            Action:  result.Action,
            Reason:  "slm_high_confidence",
        }
    } else if result.Confidence >= r.config.CloudThreshold {
        // 中等置信度：需要云端增强
        return &RouteDecision{
            Target:  ExecutionTargetCloud,
            Action:  result.Action,
            Reason:  "slm_medium_confidence",
        }
    } else if r.isNetworkAvailable() {
        // 低置信度：云端LLM处理
        return &RouteDecision{
            Target:  ExecutionTargetCloud,
            Action:  "general_query",
            Reason:  "slm_low_confidence",
        }
    } else {
        // 网络不可用：端侧兜底
        return &RouteDecision{
            Target:  ExecutionTargetFallback,
            Action:  "fallback_response",
            Reason:  "no_network",
        }
    }
}

class EdgeCloudPolicy:
    """端云协同策略引擎"""
    
    def decide(self, intent: Intent, context: Context) -> ExecutionTarget:
        # 规则1：隐私敏感操作优先本地
        if intent.category in [PRIVACY_LOCAL_EXECUTION]:
            return LOCAL
        
        # 规则2：网络状态评估
        network_quality = self.measure_network()
        if network_quality < THRESHOLD_POOR:
            return LOCAL_FALLBACK  # 弱网强制本地
        
        # 规则3：延迟预算评估
        estimated_cloud_latency = (
            self.rtt_to_cloud() + 
            self.estimate_cloud_processing(intent)
        )
        if estimated_cloud_latency > intent.max_acceptable_latency:
            return LOCAL  # 云端太慢，降级本地
        
        # 规则4：能力匹配
        if not self.local_model.can_handle(intent):
            return CLOUD  # 端侧模型不支持
        
        # 规则5：成本权衡（云侧LLM调用成本 vs 本地推理电量消耗）
        if self.should_save_cloud_quota(intent):
            return LOCAL  # 节省云端配额
        
        return CLOUD  # 默认走云端

class CloudAgent:
    """云端AI Agent核心"""
    
    async def execute(self, request: AgentRequest) -> AgentResponse:
        # 1. 加载对话上下文
        context = await self.memory.load_context(request.session_id)
        
        # 2. 检索相关记忆
        memories = await self.memory.retrieve(request.query, top_k=5)
        
        # 3. 任务规划（LLM生成行动计划）
        plan = await self.planner.generate(
            query=request.query,
            context=context,
            memories=memories
        )
        
        # 4. 工具编排执行
        results = []
        for step in plan.steps:
            if step.requires_tool:
                tool_result = await self.tool_orchestrator.execute(step)
                results.append(tool_result)
        
        # 5. 响应生成
        response = await self.response_generator.generate(
            plan=plan,
            results=results,
            context=context
        )
        
        # 6. 更新Memory
        await self.memory.store(
            session_id=request.session_id,
            interaction=request.query,
            response=response.text,
            metadata={"tools_used": [t.name for t in results]}
        )
        
        return response

// MemorySyncResolver - 处理多设备记忆冲突
type MemorySyncResolver struct {
    store *MemoryStore
}

func (r *MemorySyncResolver) Resolve(conflicts []MemoryEntry) MemoryEntry {
    // 策略1：时间戳优先（最新覆盖）
    latest := conflicts[0]
    for _, entry := range conflicts[1:] {
        if entry.UpdatedAt.After(latest.UpdatedAt) {
            latest = entry
        }
    }
    
    // 策略2：设备权重（手机 > 手表 > TV）
    // 同时间戳时，高优先级设备的数据更可信
    if r.hasTie(conflicts) {
        return r.resolveByDevicePriority(conflicts)
    }
    
    return latest
}

# 技能注册表（云端维护）
skills:
  - skill_id: "weather_query"
    name: "天气查询"
    provider: "third_party:openweathermap"
    endpoint: "https://api.openweathermap.org/data/2.5/weather"
    auth_type: "api_key"
    description: "查询全球城市实时天气"
    capabilities: ["current", "forecast"]
    languages: ["en", "fr", "ha", "sw", "am"]
    rate_limit: 100/minute
    latency_p99: "500ms"
    
  - skill_id: "local_taxi"
    name: "本地打车"
    provider: "third_party:bolt_africa"
    endpoint: "https://api.bolt.eu/v1/ride"
    auth_type: "oauth2"
    description: "非洲本地打车服务"
    capabilities: ["estimate", "book", "track"]
    regions: ["ng", "ke", "za", "eg"]
    rate_limit: 50/minute

class ThirdPartyGateway:
    """第三方API网关 - 安全、可控、可观测"""
    
    def __init__(self, config: GatewayConfig):
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60  # 60秒后尝试恢复
        )
        self.timeout = config.timeout  # 默认3秒超时
        self.retry_policy = RetryPolicy(max_retries=1)
    
    async def call(self, skill_id: str, params: dict) -> ToolResult:
        # 1. 查找技能配置
        skill = self.skill_registry.get(skill_id)
        
        # 2. 鉴权注入
        auth_header = await self.auth_manager.get_token(skill)
        
        # 3. 熔断器检查
        if self.circuit_breaker.is_open(skill_id):
            return self.fallback_response(skill_id)
        
        try:
            # 4. 带超时的HTTP调用
            response = await self.http_client.post(
                url=skill.endpoint,
                json=params,
                headers={"Authorization": auth_header},
                timeout=self.timeout
            )
            
            # 5. 结果校验
            if response.status_code != 200:
                raise APIError(f"HTTP {response.status_code}")
            
            # 6. 审计记录
            await self.audit.record(skill_id, response)
            
            return ToolResult(data=response.json(), success=True)
            
        except TimeoutError:
            self.circuit_breaker.record_failure(skill_id)
            return self.fallback_response(skill_id)
        except APIError as e:
            self.circuit_breaker.record_failure(skill_id)
            return ToolResult(error=str(e), success=False)

// DataSanitizer -  outgoing数据脱敏
func SanitizeForThirdParty(data map[string]interface{}, policy *PrivacyPolicy) map[string]interface {
    sanitized := make(map[string]interface{})
    
    for key, value := range data {
        if policy.IsSensitive(key) {
            // 脱敏处理
            switch policy.GetMaskType(key) {
            case MASK_HASH:
                sanitized[key] = sha256Hash(value)
            case MASK_REDACT:
                sanitized[key] = "[REDACTED]"
            case MASK_PARTIAL:
                sanitized[key] = partialMask(value)
            }
        } else {
            sanitized[key] = value
        }
    }
    
    return sanitized
}

class CloudHealthMonitor:
    """云端服务健康度评估"""
    
    def __init__(self):
        self.active_probe_interval = 30  # 主动探测间隔（秒）
        self.failure_window = 10         # 滑动窗口大小
        self.failure_threshold = 0.4     # 失败率阈值
        
    async def evaluate(self, session_id: str) -> ServiceHealth:
        # 1. 主动探测（轻量心跳请求）
        active = await self.probe_health_check()
        
        # 2. 被动监控（近期请求失败率）
        recent_failures = self.get_failure_rate(window=self.failure_window)
        
        # 3. 端侧探针（多设备RTT统计）
        edge_rtt = await self.aggregate_edge_probes()
        
        # 4. 综合判定
        if active.status != 200 or recent_failures > self.failure_threshold:
            return ServiceHealth(
                status=UNHEALTHY,
                reason=f"active={active.status}, failure_rate={recent_failures:.2f}",
                suggested_level=self.compute_degradation_level()
            )
        
        return ServiceHealth(status=HEALTHY, rtt=edge_rtt.median)

class CloudRouter:
    """云端服务路由与故障切换"""
    
    def __init__(self):
        self.providers = [
            Provider(name="primary",   endpoint="api.transsion.com/v1",  priority=1),
            Provider(name="fallback-1", endpoint="api-backup.transsion.com/v1", priority=2),
            Provider(name="fallback-2", endpoint="api-edge.transsion.com/v1", priority=3),
        ]
        self.circuit_breakers = {p.name: CircuitBreaker() for p in self.providers}
    
    async def route(self, request: AgentRequest) -> AgentResponse:
        # 按优先级尝试可用provider
        for provider in sorted(self.providers, key=lambda p: p.priority):
            cb = self.circuit_breakers[provider.name]
            
            if cb.state == OPEN:
                continue  # 熔断中，跳过
            
            try:
                response = await self.call_provider(provider, request)
                cb.record_success()
                return response
            except Exception as e:
                cb.record_failure()
                logger.warning(f"Provider {provider.name} failed: {e}")
                continue
        
        # 所有云端provider均不可用 → 降级到端侧
        return self.trigger_edge_fallback(request)

# 第三方服务多供应商配置
services:
  weather:
    primary:
      provider: "openweathermap"
      endpoint: "https://api.openweathermap.org/data/2.5/weather"
      rate_limit: 100/min
      cost_tier: "free"
    backup:
      - provider: "weatherapi"
        endpoint: "https://api.weatherapi.com/v1/current.json"
        rate_limit: 1000000/month
        cost_tier: "paid"
      - provider: "accuweather"
        endpoint: "https://dataservice.accuweather.com/currentconditions"
        cost_tier: "paid"
      
  translation:
    primary:
      provider: "google_translate"
    backup:
      - provider: "deepl"
      - provider: "local_model"  # 端侧轻量翻译模型
      
  maps:
    primary:
      provider: "google_maps"
    backup:
      - provider: "mapbox"
      - provider: "osm_local"  # 离线OpenStreetMap数据

class ThirdPartyMultiProvider:
    """第三方服务多供应商路由"""
    
    async def call(self, service: str, **params) -> ServiceResult:
        config = self.service_config[service]
        providers = [config.primary] + config.backup
        
        for provider in providers:
            health = self.health_checker.check(provider)
            if health != HEALTHY:
                continue
                
            try:
                result = await self.invoke(provider, params)
                # 记录成功，更新provider优先级（成功的排前面）
                self.promote_provider(service, provider)
                return result
            except (TimeoutError, APIError) as e:
                self.demote_provider(service, provider)
                logger.warning(f"Provider {provider.name} failed: {e}")
                continue
        
        # 所有供应商均不可用 → 使用缓存或兜底回复
        return self.fallback_for_service(service, params)

class SmartCache:
    """智能缓存 - 支持预取、过期降级、staleness容忍"""
    
    async def get(self, key: str, service: str) -> CacheResult:
        cached = self.store.get(key)
        
        if cached and not cached.expired:
            return CacheResult(data=cached.value, source="cache_fresh")
        
        if cached and cached.stale_while_revalidate:
            # 返回过期数据，后台异步刷新
            asyncio.create_task(self.refresh(key, service))
            return CacheResult(data=cached.value, source="cache_stale")
        
        # 缓存未命中，尝试同步获取
        try:
            fresh = await self.fetch_and_cache(key, service)
            return CacheResult(data=fresh, source="live")
        except Exception:
            # 即使过期缓存也优于无响应
            if cached:
                return CacheResult(data=cached.value, source="cache_expired_fallback")
            raise ServiceUnavailable(f"No cached data for {key}")
    
    async def prefetch(self, predictions: List[PrefetchHint]):
        """WiFi连接时预取可能需要的数据"""
        if self.is_wifi_connected() and self.is_charging():
            for hint in predictions:
                await self.fetch_and_cache(hint.key, hint.service)

class GracefulFallback:
    """优雅降级回复生成"""
    
    FALLBACK_TEMPLATES = {
        "weather": {
            "offline": "目前无法获取实时天气数据。建议您查看手机天气组件，或稍后网络恢复时再问我。",
            "partial": "我能查到{city}的历史平均气温约{temp}°C，但实时数据暂不可用。",
        },
        "taxi": {
            "offline": "打车服务暂时不可用。您可以直接打开Bolt/Uber应用叫车。",
        },
        "translation": {
            "offline": "离线翻译引擎已启动（准确率约85%）。如需更精准的翻译，请连接网络。",
        },
        "general": {
            "offline": "这个功能需要联网才能使用。我现在可以帮您做这些本地操作：{local_capabilities}",
        },
    }
    
    def generate(self, service: str, context: dict, network_state: str) -> str:
        templates = self.FALLBACK_TEMPLATES.get(service, self.FALLBACK_TEMPLATES["general"])
        template = templates.get(network_state, templates.get("offline"))
        
        # 填充模板变量
        return template.format(**context)

class GradualRecovery:
    """服务恢复不是瞬间完成的，需要渐进式验证"""
    
    async def on_service_restored(self, service: str):
        # Phase 1: 小流量验证（1%请求走新恢复的服务）
        await self.canary_test(service, traffic_pct=0.01)
        
        # Phase 2: 逐步放量（10% → 50% → 100%）
        for pct in [0.1, 0.5, 1.0]:
            success_rate = await self.ramp_traffic(service, pct)
            if success_rate < 0.95:
                # 恢复不稳定，回退
                self.rollback(service)
                return
        
        # Phase 3: 重放断网期间的积压请求
        pending = self.dequeue_pending_requests(service)
        for req in pending:
            await self.replay_request(req)
            
        # Phase 4: 通知端侧恢复全功能
        await self.broadcast_service_restored(service)

class EdgeSummary:
    """端侧对话摘要 - 最小化云端同步数据量"""
    
    def summarize(self, conversation: List[Turn]) -> SyncPayload:
        return SyncPayload(
            session_id=self.session_id,
            device_id=self.device_id,
            timestamp=time.now(),
            # 不上传原始对话，只上传结构化摘要
            intents=[turn.intent for turn in conversation],
            entities_extracted=self.extract_entities(conversation),
            user_preferences=self.detect_preferences(conversation),
            # 关键记忆点（而非完整对话）
            key_memories=[
                MemoryPoint(
                    type="preference",
                    content=f"用户喜欢{topic}",
                    confidence=0.85
                )
                for topic in self.detect_topics(conversation)
            ],
            # 统计信息
            turn_count=len(conversation),
            total_duration=conversation[-1].timestamp - conversation[0].timestamp,
        )
    
    def payload_size(self) -> int:
        """摘要大小通常 < 2KB，原始对话可能 > 50KB"""
        return len(json.dumps(self.summarize([])))

class DegradationManager:
    def evaluate(self, metrics: SystemMetrics) -> DegradationLevel:
        if metrics.cloud_available and metrics.network_quality > 0.7:
            return L0_FULL
        elif metrics.cloud_available and metrics.network_quality > 0.3:
            return L1_LITE
        elif not metrics.cloud_available and metrics.slm_available:
            return L2_EDGE
        elif not metrics.slm_available and metrics.rules_available:
            return L3_RULES
        else:
            return L4_BASIC