智能体已经就位，但企业数据还没有：从记录系统走向行动系统

import json
import sqlite3
from datetime import datetime, timezone

DB_PATH = "agent_demo.db"

SEMANTIC_CONTEXT = {
    "delayed_order": "status='delayed' and promised_at is earlier than now",
    "high_value_customer": "customer_tier='gold'",
    "compensation_policy": "Gold customers with delayed orders receive 20 credits",
}


def connect():
    db = sqlite3.connect(DB_PATH)
    db.row_factory = sqlite3.Row
    return db


def initialize():
    with connect() as db:
        db.executescript(
            """
            CREATE TABLE IF NOT EXISTS orders (
                order_id TEXT PRIMARY KEY,
                customer_id TEXT NOT NULL,
                customer_tier TEXT NOT NULL,
                status TEXT NOT NULL,
                promised_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS actions (
                idempotency_key TEXT PRIMARY KEY,
                action_type TEXT NOT NULL,
                payload TEXT NOT NULL,
                created_at TEXT NOT NULL
            );
            """
        )
        db.execute(
            """
            INSERT OR REPLACE INTO orders
            (order_id, customer_id, customer_tier, status, promised_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            ("ORD-1001", "CUS-42", "gold", "delayed", "2026-03-01T10:00:00Z"),
        )


def find_compensation_candidates():
    with connect() as db:
        rows = db.execute(
            """
            SELECT order_id, customer_id
            FROM orders
            WHERE customer_tier = ?
              AND status = ?
              AND promised_at < ?
            """,
            ("gold", "delayed", datetime.now(timezone.utc).isoformat()),
        ).fetchall()
    return [dict(row) for row in rows]


def create_compensation(order_id, customer_id, credits=20):
    key = f"delay-credit:{order_id}"
    payload = json.dumps(
        {"order_id": order_id, "customer_id": customer_id, "credits": credits},
        ensure_ascii=False,
    )
    with connect() as db:
        cursor = db.execute(
            """
            INSERT OR IGNORE INTO actions
            (idempotency_key, action_type, payload, created_at)
            VALUES (?, ?, ?, ?)
            """,
            (key, "grant_credit", payload, datetime.now(timezone.utc).isoformat()),
        )
    return {"created": cursor.rowcount == 1, "idempotency_key": key}


if __name__ == "__main__":
    initialize()
    print("Semantic context:", json.dumps(SEMANTIC_CONTEXT, indent=2))
    for order in find_compensation_candidates():
        result = create_compensation(**order)
        print(order, result)

{
  "name": "create_compensation",
  "description": "为符合已批准政策的延迟订单创建补偿。调用前必须查询订单状态。",
  "input_schema": {
    "type": "object",
    "properties": {
      "order_id": {"type": "string"},
      "customer_id": {"type": "string"},
      "credits": {"type": "integer", "minimum": 1, "maximum": 20},
      "idempotency_key": {"type": "string"}
    },
    "required": ["order_id", "customer_id", "credits", "idempotency_key"],
    "additionalProperties": false
  }
}