用 Amazon Bedrock AgentCore 落地 AI 驱动开发生命周期：从 SQL 建模到多智能体代码安全分析

目标：从 schema.sql 生成可审查的 ER 图
输入：SQL DDL，支持 CREATE TABLE、主键和外键
输出：Mermaid erDiagram 文件
验收标准：
- 每张表都出现在图中
- 主键和外键关系可追踪
- 无法解析的语句被记录，不静默丢弃
- 生成脚本可重复运行

import re
import sys
from pathlib import Path


def build_er_diagram(sql: str) -> str:
    tables = {}
    relationships = []

    table_pattern = re.compile(
        r"CREATE\\s+TABLE\\s+(?:IF\\s+NOT\\s+EXISTS\\s+)?([\\w.]+)\\s*\\((.*?)\\);",
        re.IGNORECASE | re.DOTALL,
    )

    for match in table_pattern.finditer(sql):
        table_name = match.group(1).split(".")[-1]
        columns = []
        body = match.group(2)

        for raw_line in body.splitlines():
            line = raw_line.strip().rstrip(",")
            column_match = re.match(r"([\\w]+)\\s+([A-Za-z0-9_()]+)", line)
            if not column_match:
                continue

            name, data_type = column_match.groups()
            is_pk = bool(re.search(r"\\bPRIMARY\\s+KEY\\b", line, re.IGNORECASE))
            columns.append((name, data_type, "PK" if is_pk else ""))

            fk_match = re.search(
                r"REFERENCES\\s+([\\w.]+)\\s*\\(\\s*([\\w]+)\\s*\\)",
                line,
                re.IGNORECASE,
            )
            if fk_match:
                target_table = fk_match.group(1).split(".")[-1]
                target_column = fk_match.group(2)
                relationships.append((table_name, target_table, name, target_column))

        tables[table_name] = columns

    output = ["erDiagram"]
    for table_name, columns in tables.items():
        output.append(f"    {table_name} {{")
        for name, data_type, marker in columns:
            suffix = f" {marker}" if marker else ""
            output.append(f"        {data_type} {name}{suffix}")
        output.append("    }")

    for source, target, source_col, target_col in relationships:
        output.append(f"    {source} }}o--|| {target} : \"{source_col} references {target_col}\"")

    return "\\n".join(output) + "\\n"


if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("usage: python sql_to_mermaid.py schema.sql")

    sql_file = Path(sys.argv[1])
    print(build_er_diagram(sql_file.read_text(encoding="utf-8")))

CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    email VARCHAR(255) NOT NULL
);

CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    total DECIMAL(10, 2) NOT NULL
);

python sql_to_mermaid.py schema.sql > schema.mmd

workflow: repository-security-review
input:
  repository: ./src
  branch: current
  language: python
agents:
  - name: dependency-reviewer
    responsibility: inspect dependency manifests and report risky versions
    output: findings.json
    access: read-only
  - name: code-auditor
    responsibility: inspect data flows for injection and secret exposure
    output: findings.json
    access: read-only
  - name: verifier
    responsibility: validate high-confidence findings with isolated tests
    output: verification.json
    access: sandbox
review:
  required_fields:
    - rule_id
    - severity
    - file
    - line
    - evidence
    - remediation
  deduplicate_by:
    - rule_id
    - file
    - line
  human_approval_required_for:
    - code_changes
    - dependency_upgrades

# 1. 先锁定分析范围，避免代理扫描不相关目录
find ./src -type f \\
  \( -name '*.py' -o -name 'pyproject.toml' -o -name 'requirements*.txt' \) \\
  -print

# 2. 使用现有工具提供确定性结果，再交给智能体解释上下文
python -m compileall ./src
ruff check ./src
bandit -r ./src -f json -o bandit.json

# 3. 让审查智能体基于扫描结果和只读代码上下文生成 findings.json