AI Agent 平民化：从实验室到桌面的演进之路

// 消费级 AI Agent 的技术栈组装示例
import { createAgent } from '@openclaw/core';
import { MemoryManager } from '@openclaw/memory';
import { ToolRegistry } from '@openclaw/tools';

// 1. 选择模型：本地开源模型（隐私优先）
const model = {
  provider: "local",
  modelId: "qwen-3-14b",
  maxContextTokens: 32768,
  temperature: 0.7
};

// 2. 配置记忆系统
const memory = new MemoryManager({
  vectorStore: { type: "chroma", path: "~/.agent/memory" },
  relationalStore: { type: "sqlite", path: "~/.agent/profile.db" },
  documentStore: { type: "local", path: "~/.agent/docs" }
});

// 3. 注册工具
const tools = new ToolRegistry();
tools.register("webSearch", { description: "搜索互联网", fn: searchWeb });
tools.register("calendar", { description: "管理日历", fn: manageCalendar });
tools.register("email", { description: "收发邮件", fn: handleEmail });
tools.register("fileSystem", { description: "读写文件", fn: fileOps });

// 4. 组装 Agent
const agent = createAgent({
  name: "个人助理",
  model,
  memory,
  tools,
  personality: { style: "专业友好", language: "中文" },
  safety: {
    maxToolCallsPerTurn: 5,
    requireConfirmation: ["email", "payment"],
    dataRetention: "local-only"
  }
});

// 5. 启动
await agent.initialize();
console.log("个人助理已启动，随时待命！");

# 对比：使用 CrewAI 构建团队协作型 Agent
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# 模型选择：API 模型（能力优先）
llm = ChatOpenAI(model="gpt-5.5-instant", temperature=0.7)

# 定义角色化 Agent
researcher = Agent(
    role="高级研究员",
    goal="深度调研指定主题并提供结构化报告",
    backstory="你是一位拥有 10 年经验的行业研究员，擅长从海量信息中提炼核心观点。",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="资深内容创作者",
    goal="基于研究报告撰写高质量内容",
    backstory="你是一位擅长将复杂技术概念转化为通俗易懂内容的写作者。",
    llm=llm,
    verbose=True
)

# 定义任务
research_task = Task(
    description="调研 AI Agent 平民化的最新趋势",
    expected_output="包含关键发现、数据支撑和趋势预判的研究报告",
    agent=researcher
)

writing_task = Task(
    description="基于研究报告撰写博客文章",
    expected_output="5000 字以上的深度技术博客",
    agent=writer
)

# 组装团队
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

# 执行
result = crew.kickoff()
print(result)

// Agent 安全治理框架：权限控制与审计
import { AgentSecurity } from '@openclaw/security';

const security = new AgentSecurity({
  // 最小权限原则：只授予 Agent 必要的工具
  permissionModel: "least-privilege",

  // 权限分级定义
  permissionLevels: {
    read: {    // 读取权限：无需确认
      tools: ["calendar_read", "weather", "search"],
      requireConfirmation: false
    },
    write: {   // 写入权限：操作前确认
      tools: ["calendar_write", "email_send", "file_write"],
      requireConfirmation: true
    },
    critical: { // 关键权限：双重确认
      tools: ["payment", "account_delete", "system_config"],
      requireConfirmation: true,
      doubleConfirmation: true
    }
  },

  // 审计日志：所有操作都记录
  auditConfig: {
    enabled: true,
    storage: "local",  // 审计日志本地存储
    retention: "90d",  // 保留 90 天
    includeInputs: true,
    includeOutputs: true,
    includeDecisions: true
  },

  // 渐进式授权
  progressiveAuth: {
    initialLevel: "read",
    trustThreshold: {
      promotions: [
        { from: "read", to: "write", after: "7d", minActions: 50 },
        { from: "write", to: "critical", after: "30d", minActions: 200 }
      ]
    }
  }
});

// 拦截高风险操作
security.on("action:critical", async (event) => {
  console.log(`⚠️ 关键操作请求: ${event.toolName}`);
  console.log(`操作详情: ${JSON.stringify(event.details)}`);
  
  // 请求用户确认
  const confirmed = await requestUserConfirmation(event);
  if (!confirmed) {
    console.log("❌ 用户拒绝，操作已取消");
    event.deny();
    return;
  }
  
  // 双重确认
  if (security.requiresDoubleConfirmation(event)) {
    const secondConfirm = await requestSecondConfirmation(event);
    if (!secondConfirm) {
      event.deny();
      return;
    }
  }
  
  event.approve();
});

# Agent 行为审计与分析工具
import json
from datetime import datetime, timedelta
from collections import defaultdict

class AgentAuditor:
    """Agent 行为审计系统"""
    
    def __init__(self, log_path: str):
        self.logs = self._load_logs(log_path)
    
    def _load_logs(self, path: str) -> list:
        with open(path, 'r') as f:
            return [json.loads(line) for line in f]
    
    def analyze_behavior(self, days: int = 7) -> dict:
        """分析 Agent 最近 N 天的行为模式"""
        cutoff = datetime.now() - timedelta(days=days)
        recent_logs = [
            log for log in self.logs
            if datetime.fromisoformat(log['timestamp']) > cutoff
        ]
        
        # 统计工具调用频率
        tool_usage = defaultdict(int)
        for log in recent_logs:
            if 'tool_call' in log:
                tool_usage[log['tool_call']['name']] += 1
        
        # 检测异常行为
        anomalies = self._detect_anomalies(recent_logs)
        
        # 分析决策质量
        decision_quality = self._analyze_decisions(recent_logs)
        
        return {
            "period": f"最近 {days} 天",
            "total_actions": len(recent_logs),
            "tool_usage": dict(tool_usage),
            "anomalies": anomalies,
            "decision_quality": decision_quality
        }
    
    def _detect_anomalies(self, logs: list) -> list:
        """检测异常行为模式"""
        anomalies = []
        
        # 检测异常高频调用
        hourly_counts = defaultdict(int)
        for log in logs:
            hour = datetime.fromisoformat(log['timestamp']).hour
            hourly_counts[hour] += 1
        
        avg_per_hour = sum(hourly_counts.values()) / max(len(hourly_counts), 1)
        for hour, count in hourly_counts.items():
            if count > avg_per_hour * 3:
                anomalies.append({
                    "type": "高频调用",
                    "hour": hour,
                    "count": count,
                    "avg": avg_per_hour,
                    "severity": "warning"
                })
        
        return anomalies
    
    def _analyze_decisions(self, logs: list) -> dict:
        """分析决策质量"""
        total = len(logs)
        confirmed = sum(1 for log in logs if log.get('confirmation') == True)
        denied = sum(1 for log in logs if log.get('confirmation') == False)
        
        return {
            "confirmation_rate": confirmed / max(total, 1),
            "denial_rate": denied / max(total, 1),
            "success_rate": sum(1 for log in logs if log.get('success')) / max(total, 1)
        }

# 使用示例
auditor = AgentAuditor("~/.agent/audit.log")
report = auditor.analyze_behavior(days=7)
print(json.dumps(report, indent=2, ensure_ascii=False))

# agent-config.yaml — 消费级 AI Agent 配置文件
# 这个文件定义了一个个人助理 Agent 的完整配置

agent:
  name: "我的个人助理"
  version: "1.0.0"
  language: "zh-CN"

model:
  # 本地模型（推荐，数据隐私最佳）
  provider: "local"
  model: "qwen-3-14b"
  # 或云端模型（更强能力，但有隐私考虑）
  # provider: "openai"
  # model: "gpt-5.5-instant"
  temperature: 0.7
  max_tokens: 4096

memory:
  vector_store:
    type: "chroma"
    path: "~/.openclaw/workspace/memory/vectors"
    embedding_model: "text-embedding-3-small"
  relational_store:
    type: "sqlite"
    path: "~/.openclaw/workspace/memory/profile.db"
  document_store:
    type: "local"
    path: "~/.openclaw/workspace/memory/docs"

capabilities:
  schedule:
    enabled: true
    calendar_source: "google"  # 或 apple, outlook
    auto_create_events: false  # 需要确认后创建
    natural_language: true
  
  email:
    enabled: true
    accounts: ["personal@example.com"]
    auto_reply: false          # 不自动发送回复
    summarization: true        # 自动生成摘要
    max_emails_per_batch: 20
  
  search:
    enabled: true
    sources: ["web", "wiki", "arxiv"]
    summarize_results: true
    max_results: 10
  
  knowledge:
    enabled: true
    auto_categorize: true
    indexing_enabled: true
    formats: ["md", "txt", "pdf"]

security:
  permission_model: "least-privilege"
  confirmation_required:
    - "email_send"
    - "calendar_delete"
    - "payment"
  audit_log:
    enabled: true
    retention_days: 90

// 消费级 Agent 的启动与交互示例
import { OpenClawAgent } from '@openclaw/core';
import { loadConfig } from '@openclaw/config';

async function main() {
  // 加载配置
  const config = loadConfig('~/.openclaw/workspace/agent-config.yaml');
  
  // 初始化 Agent
  const agent = new OpenClawAgent(config);
  
  console.log('🍪 奥利奥个人助理已启动！');
  console.log('可用能力：日程管理、邮件处理、信息搜索、知识管理');
  console.log('输入 "help" 查看使用说明');
  
  // 主循环：接收用户输入，调用 Agent 处理
  while (true) {
    const userInput = await promptUser('> ');
    
    if (userInput === 'exit') break;
    if (userInput === 'status') {
      console.log(agent.getStatus());
      continue;
    }
    if (userInput === 'audit') {
      const report = agent.getAuditReport({ days: 7 });
      console.log(JSON.stringify(report, null, 2));
      continue;
    }
    
    try {
      // Agent 处理用户请求
      const result = await agent.process(userInput);
      
      // 如果有需要确认的操作
      if (result.requiresConfirmation) {
        console.log(`⚠️ Agent 建议执行：${result.action}`);
        const confirm = await promptUser('确认？(y/n): ');
        if (confirm === 'y') {
          await result.execute();
          console.log('✅ 操作已完成');
        } else {
          console.log('❌ 操作已取消');
        }
      } else {
        // 直接输出结果
        console.log(result.output);
      }
    } catch (error) {
      console.error(`❌ 处理失败：${error.message}`);
    }
  }
  
  await agent.shutdown();
  console.log('👋 个人助理已关闭');
}

main();