Cloudflare 如何搭建统一数据平台 Town Lake 及其上的 AI Agent Skipper

import duckdb

con = duckdb.connect()

# 注册一个 Parquet 文件作为表（模拟 S3 上的冷数据）
con.execute("""
    CREATE VIEW request_logs AS
    SELECT * FROM read_parquet('data/request_logs.parquet')
""")

# 注册一个 PostgreSQL 远端表（模拟业务数据库）
# 需先安装 duckdb postgres 扩展
con.execute("INSTALL postgres; LOAD postgres;")
con.execute("""
    CREATE VIEW user_accounts AS
    SELECT * FROM postgres_scan(
        'host=localhost port=5432 dbname=myapp user=postgres password=secret',
        'public', 'accounts'
    )
""")

# 注册一个 CSV（模拟第三方导出数据）
con.execute("""
    CREATE VIEW cdn_metrics AS
    SELECT * FROM read_csv_auto('data/cdn_metrics.csv')
""")

# 现在可以跨源 JOIN
result = con.execute("""
    SELECT r.status_code, COUNT(*) AS cnt
    FROM request_logs r
    JOIN user_accounts u ON r.user_id = u.id
    WHERE u.tier = 'pro'
    GROUP BY r.status_code
    ORDER BY cnt DESC
""").fetchall()
print(result)

import json
import duckdb
from openai import OpenAI

client = OpenAI()  # 需设置 OPENAI_API_KEY 环境变量
con = duckdb.connect()

# 先注册同样的视图（同上，此处省略重复代码）
# ...

# 从 DuckDB 拿 schema 信息，作为 LLM 的上下文
def get_schema_info(conn):
    tables = conn.execute("SHOW ALL TABLES").fetchall()
    schema_text = ""
    for t in tables:
        name = t[0]
        cols = conn.execute(f"DESCRIBE {name}").fetchall()
        schema_text += f"\n表 {name}:\n"
        for c in cols:
            schema_text += f"  - {c[0]} ({c[1]})\n"
    return schema_text

SCHEMA = get_schema_info(con)

SYSTEM_PROMPT = f"""你是数据分析助手 Skipper。用户会用自然语言提问，你需要：
1. 生成一条 DuckDB SQL 查询
2. 只使用以下已知表和字段，不要猜测不存在的字段：
{SCHEMA}
3. 输出格式为 JSON：{{"sql": "...", "explanation": "简要解释查询逻辑"}}
4. 如果问题无法用现有表回答，说明缺少什么数据。
"""

def ask_skipper(question: str) -> dict:
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": question}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    parsed = json.loads(resp.choices[0].message.content)
    sql = parsed["sql"]
    explanation = parsed["explanation"]

    print(f"🔍 Skipper 解释：{explanation}")
    print(f"📝 生成的 SQL：\n{sql}")

    try:
        rows = con.execute(sql).fetchall()
        cols = [desc[0] for desc in con.description]
        print(f"\n📊 查询结果（{len(rows)} 行）：")
        for row in rows[:20]:
            print(dict(zip(cols, row)))
        return {"sql": sql, "rows": rows, "columns": cols}
    except Exception as e:
        print(f"❌ SQL 执行失败：{e}")
        return {"sql": sql, "error": str(e)}

# 使用示例
ask_skipper("过去一周 pro 用户最常见的 5 个 HTTP 状态码是什么？")

import duckdb
con = duckdb.connect()
con.execute("""
    COPY (SELECT
        range AS id,
        CASE WHEN range % 3 = 0 THEN 'pro' WHEN range % 3 = 1 THEN 'free' ELSE 'enterprise' END AS tier,
        'user' || range::VARCHAR AS name
      FROM range(1, 100))
    TO 'data/user_accounts.parquet'
""")
con.execute("""
    COPY (SELECT
        range AS log_id,
        range % 50 AS user_id,
        CASE WHEN range % 5 = 0 THEN 500 WHEN range % 4 = 0 THEN 404 ELSE 200 END AS status_code,
        DATE '2024-01-01' + (range % 7) AS date
      FROM range(1, 1000))
    TO 'data/request_logs.parquet'
""")