Stonebraker 批判 AI Agent：数据库泰斗为什么说「Agent 绕不开数据库老问题」

-- Agent 任务状态表
CREATE TABLE agent_tasks (
    task_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    agent_id VARCHAR(64) NOT NULL,
    task_name VARCHAR(256) NOT NULL,
    status VARCHAR(32) NOT NULL DEFAULT 'pending',
    -- status: pending, running, completed, failed, cancelled
    current_phase VARCHAR(128),
    input_data JSONB,
    output_data JSONB,
    error_message TEXT,
    retry_count INTEGER DEFAULT 0,
    max_retries INTEGER DEFAULT 3,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ
);

-- Agent 执行日志表（类似 WAL）
CREATE TABLE agent_execution_log (
    log_id BIGSERIAL PRIMARY KEY,
    task_id UUID REFERENCES agent_tasks(task_id),
    phase VARCHAR(128) NOT NULL,
    action VARCHAR(64) NOT NULL,
    -- action: start, complete, fail, retry, rollback
    input_snapshot JSONB,
    output_snapshot JSONB,
    duration_ms INTEGER,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Agent 状态检查点表
CREATE TABLE agent_checkpoints (
    checkpoint_id BIGSERIAL PRIMARY KEY,
    task_id UUID REFERENCES agent_tasks(task_id),
    checkpoint_data JSONB NOT NULL,
    -- 完整状态快照
    sequence_num INTEGER NOT NULL,
    -- 检查点序号，用于恢复时定位
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 索引优化
CREATE INDEX idx_agent_tasks_status ON agent_tasks(status);
CREATE INDEX idx_agent_tasks_agent ON agent_tasks(agent_id);
CREATE INDEX idx_agent_tasks_created ON agent_tasks(created_at DESC);
CREATE INDEX idx_agent_log_task ON agent_execution_log(task_id);
CREATE INDEX idx_agent_checkpoint_task ON agent_checkpoints(task_id, sequence_num DESC);

import { Pool, PoolClient } from 'pg';

interface AgentTask {
  taskId: string;
  agentId: string;
  taskName: string;
  status: 'pending' | 'running' | 'completed' | 'failed' | 'cancelled';
  currentPhase?: string;
  inputData?: Record<string, any>;
  outputData?: Record<string, any>;
  errorMessage?: string;
  retryCount: number;
  maxRetries: number;
}

class AgentStateManager {
  private pool: Pool;

  constructor(connectionString: string) {
    this.pool = new Pool({ connectionString });
  }

  // 创建任务（带事务）
  async createTask(
    agentId: string,
    taskName: string,
    inputData?: Record<string, any>
  ): Promise<string> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');
      
      const result = await client.query(
        `INSERT INTO agent_tasks (agent_id, task_name, input_data, status)
         VALUES ($1, $2, $3, 'pending')
         RETURNING task_id`,
        [agentId, taskName, inputData ? JSON.stringify(inputData) : null]
      );
      
      await client.query('COMMIT');
      return result.rows[0].task_id;
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }

  // 更新任务状态（带并发控制）
  async updateTaskStatus(
    taskId: string,
    newStatus: AgentTask['status'],
    expectedStatus?: AgentTask['status']
  ): Promise<boolean> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');
      
      let sql = `UPDATE agent_tasks 
                  SET status = $1, updated_at = NOW()`;
      const params: any[] = [newStatus];
      
      if (expectedStatus) {
        sql += ` WHERE task_id = $2 AND status = $3`;
        params.push(taskId, expectedStatus);
      } else {
        sql += ` WHERE task_id = $2`;
        params.push(taskId);
      }
      
      if (newStatus === 'running') {
        sql = sql.replace('updated_at', 'started_at, updated_at');
      } else if (newStatus === 'completed' || newStatus === 'failed') {
        sql = sql.replace('updated_at', 'completed_at, updated_at');
      }
      
      const result = await client.query(sql, params);
      await client.query('COMMIT');
      
      return result.rowCount! > 0;
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }

  // 创建检查点（用于失败恢复）
  async createCheckpoint(
    taskId: string,
    checkpointData: Record<string, any>
  ): Promise<number> {
    const result = await this.pool.query(
      `INSERT INTO agent_checkpoints (task_id, checkpoint_data, sequence_num)
       SELECT $1, $2, COALESCE(MAX(sequence_num), 0) + 1
       FROM agent_checkpoints WHERE task_id = $1
       RETURNING sequence_num`,
      [taskId, JSON.stringify(checkpointData)]
    );
    return result.rows[0].sequence_num;
  }

  // 恢复任务状态
  async restoreTask(taskId: string): Promise<Record<string, any> | null> {
    const result = await this.pool.query(
      `SELECT checkpoint_data FROM agent_checkpoints
       WHERE task_id = $1
       ORDER BY sequence_num DESC LIMIT 1`,
      [taskId]
    );
    return result.rows[0]?.checkpoint_data || null;
  }

  // 查询历史执行
  async queryTasks(filters: {
    agentId?: string;
    status?: string;
    fromDate?: Date;
    limit?: number;
  }): Promise<AgentTask[]> {
    let sql = 'SELECT * FROM agent_tasks WHERE 1=1';
    const params: any[] = [];
    let paramIndex = 1;
    
    if (filters.agentId) {
      sql += ` AND agent_id = $${paramIndex++}`;
      params.push(filters.agentId);
    }
    if (filters.status) {
      sql += ` AND status = $${paramIndex++}`;
      params.push(filters.status);
    }
    if (filters.fromDate) {
      sql += ` AND created_at >= $${paramIndex++}`;
      params.push(filters.fromDate);
    }
    
    sql += ` ORDER BY created_at DESC LIMIT $${paramIndex}`;
    params.push(filters.limit || 100);
    
    const result = await this.pool.query(sql, params);
    return result.rows;
  }
}

import { createClient, RedisClientType } from 'redis';

interface AgentState {
  taskId: string;
  agentId: string;
  status: string;
  phase: string;
  result?: string;
  error?: string;
  retryCount: number;
  updatedAt: number;
}

class AgentRedisCache {
  private client: RedisClientType;
  private readonly KEY_PREFIX = 'agent:task:';
  private readonly STREAM_KEY = 'agent:events';
  private readonly TTL_SECONDS = 3600; // 1 小时过期

  constructor(redisUrl: string) {
    this.client = createClient({ url: redisUrl });
    this.client.on('error', err => console.error('Redis Error:', err));
  }

  async connect(): Promise<void> {
    await this.client.connect();
  }

  // 保存 Agent 状态（热数据）
  async saveState(state: AgentState): Promise<void> {
    const key = `${this.KEY_PREFIX}${state.taskId}`;
    
    // 使用 Pipeline 批量写入，减少网络往返
    const pipeline = this.client.multi();
    pipeline.hSet(key, {
      agentId: state.agentId,
      status: state.status,
      phase: state.phase,
      result: state.result || '',
      error: state.error || '',
      retryCount: String(state.retryCount),
      updatedAt: String(state.updatedAt)
    });
    pipeline.expire(key, this.TTL_SECONDS);
    await pipeline.exec();

    // 异步写入事件流（用于持久化到冷数据层）
    await this.client.xAdd(this.STREAM_KEY, '*', {
      taskId: state.taskId,
      action: 'state_update',
      status: state.status,
      phase: state.phase,
      timestamp: String(state.updatedAt)
    });
  }

  // 获取 Agent 状态
  async getState(taskId: string): Promise<AgentState | null> {
    const key = `${this.KEY_PREFIX}${taskId}`;
    const data = await this.client.hGetAll(key);
    
    if (!data || !data.agentId) return null;
    
    return {
      taskId,
      agentId: data.agentId,
      status: data.status,
      phase: data.phase,
      result: data.result || undefined,
      error: data.error || undefined,
      retryCount: parseInt(data.retryCount) || 0,
      updatedAt: parseInt(data.updatedAt) || 0
    };
  }

  // 乐观锁：原子更新状态
  async updateStateIfMatch(
    taskId: string,
    expectedStatus: string,
    newState: Partial<AgentState>
  ): Promise<boolean> {
    const key = `${this.KEY_PREFIX}${taskId}`;
    
    // Lua 脚本实现原子条件更新
    const luaScript = `
      local current = redis.call('HGET', KEYS[1], 'status')
      if current == ARGV[1] then
        for i = 2, #ARGV, 2 do
          redis.call('HSET', KEYS[1], ARGV[i], ARGV[i+1])
        end
        return 1
      end
      return 0
    `;
    
    const fields: string[] = [];
    for (const [key, value] of Object.entries(newState)) {
      fields.push(key, String(value));
    }
    
    const result = await this.client.eval(luaScript, {
      keys: [key],
      arguments: [expectedStatus, ...fields]
    });
    
    return result === 1;
  }

  // 清理过期状态
  async cleanupExpired(): Promise<number> {
    // 通过 SCAN 查找过期 key（非阻塞）
    let count = 0;
    let cursor = 0;
    
    do {
      const result = await this.client.scan(cursor, {
        MATCH: `${this.KEY_PREFIX}*`,
        COUNT: 100
      });
      cursor = result.cursor;
      count += result.keys.length;
    } while (cursor !== 0);
    
    return count;
  }
}

import asyncio
import asyncpg
import json
from datetime import datetime

class ColdDataSyncer:
    """将 Redis Stream 中的事件异步同步到 PostgreSQL"""
    
    def __init__(self, redis_client, pg_dsn: str, batch_size: int = 100):
        self.redis = redis_client
        self.pg_dsn = pg_dsn
        self.batch_size = batch_size
        self.consumer_group = "cold_sync_group"
        self.consumer_name = "syncer_1"
        self.stream_key = "agent:events"
        self.running = False
    
    async def start(self):
        """启动消费者"""
        # 创建消费者组
        try:
            await self.redis.xgroup_create(
                self.stream_key,
                self.consumer_group,
                id="0",
                mkstream=True
            )
        except Exception:
            pass  # 组已存在
        
        self.running = True
        pg_pool = await asyncpg.create_pool(dsn=self.pg_dsn)
        
        while self.running:
            # 读取新消息
            messages = await self.redis.xreadgroup(
                self.consumer_group,
                self.consumer_name,
                {self.stream_key: ">"},
                count=self.batch_size,
                block=1000  # 阻塞 1 秒
            )
            
            if not messages:
                continue
            
            async with pg_pool.acquire() as conn:
                async with conn.transaction():
                    for stream, msg_list in messages:
                        for msg_id, fields in msg_list:
                            await self._process_event(conn, fields)
                            
                            # 确认消息
                            await self.redis.xack(
                                self.stream_key,
                                self.consumer_group,
                                msg_id
                            )
    
    async def _process_event(self, conn, fields: dict):
        """处理单个事件"""
        action = fields.get("action", "")
        task_id = fields.get("taskId", "")
        
        if action == "state_update":
            await conn.execute(
                """UPDATE agent_tasks 
                   SET status = $1, current_phase = $2, updated_at = NOW()
                   WHERE task_id = $3""",
                fields.get("status"),
                fields.get("phase"),
                task_id
            )
            
            # 写入执行日志
            await conn.execute(
                """INSERT INTO agent_execution_log 
                   (task_id, phase, action, created_at)
                   VALUES ($1, $2, $3, NOW())""",
                task_id,
                fields.get("phase", ""),
                "state_sync"
            )
    
    def stop(self):
        self.running = False