erjonb19/Security-Constrained-Agent-Runtime

# 受治理的临床 Agent 一个面向医疗保健数据的受治理自然语言分析 Agent。你用纯英语提问；语言模型编写 SQL；基于能力（capability）的安全运行时在 SQL 接触数据库之前决定是否允许其运行。模型负责提议，护栏负责裁决。它目前运行在真实的 CMS Medicare 数据上，位于 HTTP API 之后，并在每次响应中返回完整的授权决策。 ## 为什么这很重要 Healthtech 和医疗导航公司希望在他们数据的前端部署一个 LLM，以便员工和成员无需编写 SQL 即可提问。阻碍在于信任：模型可能会产生幻觉查询、试图访问不该访问的表，或者被 prompt injection 诱导去窃取数据。在医疗保健领域，这不是一个 bug，而是一次合规事件。大多数“面向医疗数据的 AI Agent”项目都是先构建聊天机器人，然后再（如果有的话）强行补上治理。这个项目的构建方式恰恰相反。治理本身就是产品，而分析 Agent 则是证明其在真实数据上有效的证据。它提供的保证很简单：无论模型被说服编写什么样的查询，该 Agent 永远只能运行只读的、经允许列表批准的、有行数上限的查询。 ## 架构 系统分为两层：一个可重用的安全运行时，以及构建在其上的参考应用程序。 **安全运行时**通过默认拒绝（default-deny）策略来调解 Agent 发起的每一次工具调用。其组件包括： - **Policy engine** — 风险分级，默认拒绝。能力（Capabilities）被划分为自动、受控（需要批准）或直接拒绝。在 `medicare_policy.yaml` 中定义。 - **SQL guard** (`sql_guard.py`) — 一个 AST 验证器（sqlglot），是 Agent 编写 SQL 的执行边界。仅限 SELECT、Gold 表允许列表、自动行数上限、无堆叠语句、无 catalog/system schema、无文件读取函数。拒绝时安全失败（fail closed）。 - **Groundedness check** (`groundedness.py`) — 验证生成的简报中的每一项陈述是否都能追溯到一个真实的返回行，从而确保 Agent 绝不能陈述它未检索到的数字。 - **Taint tracking** (`src/security/`) — 阻止来自受污染源的数据流入被拒绝的接收端，这是抵御由 prompt injection 驱动的数据窃取的防线。 - **Audit logger** — 每一项决策（允许 / 拒绝 / 需要批准）都会连同能力、原因和延迟写入 JSONL，作为合规审计记录。 **参考应用程序**是一个受治理的 CMS Medicare 分析 Agent： - **NL-to-SQL planner** (`nl_to_sql_planner.py`) — 将纯英语问题转化为针对 Gold schema 的一条 DuckDB SELECT 查询。通过兼容 OpenAI 的客户端实现供应商无关；默认使用 Cerebras (`gpt-oss-120b`)，只需修改一行配置即可切换至 Groq。 - **Analytics tool** (`analytics_query_tool.py`) — 通过 guard 运行规划器的 SQL，然后针对 Gold lakehouse 执行验证过的查询。 - **HTTP service** (`app.py`) — FastAPI。提供自然语言和原始 SQL 端点，数据路由需进行 API-key 认证，每次响应都会返回治理决策。 - **AIOps panel** (`aiops_panel.py`) — 基于 Streamlit 构建的审计日志仪表板：按控制维度划分的结果、拒绝原因、延迟以及决策时间线。 ## 数据 使用双目录抓取器 (`fetch_cms.py`) 和 DuckDB medallion pipeline (`build_hospital_gold.py`) 基于公开的 CMS 数据构建： - **Hospital-profile Gold** (`gold_hospital_profile`) — 涵盖东北部和Mid-Atlantic十二个州的 750 家医院，每家医院一行，通过 CMS facility ID 关联：总体星级评分、人均 Medicare 支出、五项疾病层面的 30 天再入院率，以及包括精神科急诊中位等待时间在内的四项 ED 流量指标。 - **Geographic Gold** — 来自 CMS Geographic Variation 文件的区域级利用率、成本和异常表。 两者都通过相同的 guardrails 进行查询。 ## 运行效果 向 `/query` 发送的自然语言请求： ``` { "question": "Which hospitals give the best value, high quality and low cost?" } ``` 返回模型编写的 SQL、guard 实际运行的 SQL、决策以及数据行： ``` { "allowed": true, "decided_by": "executed", "safe_sql": "SELECT facility_name, state, star_rating, mspb_score ... LIMIT 15", "row_count": 15, "rows": [ { "facility_name": "NEWTON-WELLESLEY HOSPITAL", "state": "MA", "star_rating": 5, "mspb_score": 0.88 } ] } ``` 不允许的查询，即使是完全有效的 SQL，也会在边界处被拒绝： ``` { "allowed": false, "decided_by": "guard", "reason": "table not on Gold allowlist: billing_raw" } ``` 模型也可以进行对抗性尝试。当被要求删除低评分医院或读取 PHI 表时，它永远无法生成能让 guard 运行的查询；guard 的执行机制是独立验证的，专门针对那些命中非允许列表表或系统 catalog 的有效但被禁止的 SQL。 ## 运行 ``` pip install -r requirements-api.txt $env:CEREBRAS_API_KEY="..." # the planner's model $env:API_KEY="..." # require a key on the data endpoints uvicorn app:app --port 8000 ``` 打开 `http://localhost:8000/docs` 查看交互式 API。`/health` 报告就绪状态，`/schema` 返回可查询的 schema，`/query` 接收自然语言，`/raw-sql` 接收受保护的 SQL。 受治理的 pipeline 也可以直接调用： ``` python nl_to_sql_planner.py # NL question -> model SQL -> guard -> rows, plus the guard-enforcement check python run_hospital_query.py # the same four care-navigation queries through the runtime ``` ## 适用范围与诚实的局限性 这是一个可运行的参考实现，在此如实描述： - 它以单实例运行，并在锁机制下序列化查询。这对于单实例是正确的；水平扩展是未来的工作。 - 当设置了 `API_KEY` 时，会强制执行认证；如果未设置，则服务会以开发模式开放运行（在启动时和 `/health` 中会有醒目标记）。在公开暴露之前，请务必设置密钥。 - 数据为公开的 CMS 数据。这里不包含 PHI，且系统尚未针对 PHI 或生产负载进行强化。 - Gold 是按需构建的，而不是按计划构建的。自我刷新的 pipeline 已列入规划。 ## 路线图 - 云端迁移：ADLS Gen2 + Databricks Workflows，将分析工具重新指向云端的 Gold。 - 通过 GitHub Actions 定期进行每月数据刷新，使 pipeline 实现自我维护。 - 基于 API 为非技术用户提供轻量级的查询 UI。 - 在进行任何公开的、敏感的部署之前，实现失败安全（Fail-closed）的认证和基于密钥的速率限制。 ## 许可证 MIT。

标签：AI安全, Chat Copilot, CISA项目, Kubernetes, SQL验证, XSS注入, 医疗健康数据分析, 合规治理, 审计日志, 权限控制, 逆向工具, 零日漏洞检测