把 AI 从“会回答”变成“可控协作”：更安全、更高效的人机交互方法

需求 -> Prompt -> 模型输出 -> 直接使用

定义任务 -> 限制上下文 -> 生成候选方案 -> 自动检查 -> 人工批准 -> 执行与复盘

角色：你是负责审查后端改动的高级工程师。

目标：定位订单接口 P95 延迟上升的可能原因，并提出最小改动方案。

允许使用的上下文：
- src/orders/ 下的代码
- 最近 30 分钟的脱敏追踪数据
- 当前数据库索引定义

禁止事项：
- 不得要求或输出密码、令牌和客户个人信息
- 不得直接执行数据库变更
- 不得修改订单接口的公开响应结构

输出要求：
1. 按可信度排序的原因列表
2. 每项原因对应的证据
3. 建议修改的文件和配置
4. 验证命令与回滚步骤
5. 尚未确认的假设

如果证据不足，请明确回答“证据不足”，不要补造日志或指标。

#!/usr/bin/env python3
import argparse
import json
import os
import re
import sys
import urllib.request
from datetime import datetime, timezone

PATTERNS = {
    "private_key": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
    "bearer_token": re.compile(r"Bearer\\s+[A-Za-z0-9._~+/-]+=*", re.IGNORECASE),
    "aws_access_key": re.compile(r"AKIA[0-9A-Z]{16}"),
}

SYSTEM_RULES = """You are an engineering assistant.
Use only the context supplied by the user.
Do not invent logs, metrics, files, or test results.
Do not request secrets or personal data.
Separate evidence, assumptions, and recommendations.
For destructive operations, provide commands but require human approval.
"""


def find_sensitive_data(text: str) -> list[str]:
    return [name for name, pattern in PATTERNS.items() if pattern.search(text)]


def build_payload(task: str) -> dict:
    return {
        "messages": [
            {"role": "system", "content": SYSTEM_RULES},
            {"role": "user", "content": task},
        ]
    }


def write_audit(event: dict) -> None:
    event["timestamp"] = datetime.now(timezone.utc).isoformat()
    with open("ai-audit.jsonl", "a", encoding="utf-8") as handle:
        handle.write(json.dumps(event, ensure_ascii=False) + "\n")


def call_model(payload: dict) -> str:
    endpoint = os.environ["AI_ENDPOINT"]
    api_key = os.environ["AI_API_KEY"]
    request = urllib.request.Request(
        endpoint,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=60) as response:
        body = json.load(response)
    return body["output"]


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("task", help="Task and sanitized context sent to the model")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    findings = find_sensitive_data(args.task)
    if findings:
        write_audit({"status": "blocked", "findings": findings})
        print(f"Blocked sensitive input: {', '.join(findings)}", file=sys.stderr)
        raise SystemExit(2)

    payload = build_payload(args.task)
    write_audit({"status": "accepted", "task_length": len(args.task)})

    if args.dry_run:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
        return

    print(call_model(payload))


if __name__ == "__main__":
    main()

python3 safe_ai.py --dry-run \
  "审查订单查询的性能问题。只给出建议、验证命令和回滚步骤，不执行修改。"

export AI_ENDPOINT="https://ai-gateway.example.internal/v1/chat"
export AI_API_KEY="replace-with-a-short-lived-token"
python3 safe_ai.py \
  "根据以下脱敏指标分析延迟：P50=80ms, P95=920ms, error_rate=0.3%。"