让聊天代理回答“哪些租约不合规”并不难，难的是证明它没有漏掉任何一份租约、没有临时改变判断标准，而且每个结论都能追溯到具体数据和规则。Adjudicated Query（裁决式查询）模式把 Amazon Quick 聊天代理、受限的 MCP 服务器与确定性规则引擎组合起来：代理负责理解问题和组织答案，规则引擎负责逐条裁决，MCP 则把代理能做的事情限制在可审计的边界内。
这种分工尤其适合租约、合同、许可证和供应商档案等批量合规场景。它追求的不是“模型看起来判断得不错”，而是相对于一个明确的数据快照和规则版本，给出完整、可复现、可辩护的结果。
不要让聊天模型直接充当裁决者
一个宽泛的实现可能会让代理查询数据库、阅读合同文本，然后直接生成答案。这样虽然灵活，却会带来几个难以接受的问题：
- 查询范围可能随提示词变化，无法证明所有记录都被检查过。
- 同一条款在不同对话中可能得到不同解释。
- 长结果集可能因分页、上下文窗口或超时而被截断。
- 审计人员很难还原“当时用了哪些数据和规则”。
- 如果代理拥有任意 SQL 或对象存储读取权限，提示注入还可能扩大数据访问范围。
裁决式查询把自然语言层和合规判定层分开。Amazon Quick 可以把“检查所有 2026 年到期、通知期不足 30 天的租约”转换成一个受约束的扫描请求，但不应自己决定某条记录是否合规。
一次可辩护的扫描至少需要固定以下内容：
- 数据范围：例如某个 S3 对象版本、清单文件或不可变快照。
- 规则版本：例如
lease-policy-2026.03，而不是一段会随时变化的提示词。 - 逐条结果：每条输入记录都必须得到
COMPLIANT、NON_COMPLIANT或NEEDS_REVIEW。 - 证据引用：记录触发的规则、原始字段和源文档位置。
- 覆盖证明：输入数量、已裁决数量、重复标识和失败数量都要进入清单。
这里的“可证明完整”有明确边界：它证明指定快照中的记录都经过了指定规则，而不是证明数据源从未漏收合同，也不代表规则本身就是最终法律意见。
参考架构：代理只通过受限 MCP 工具工作
在 AWS 上落地时，可以把架构拆成五层：
- 采集与标准化层：原始租约保存在 S3，并抽取成结构化记录。扫描前生成不可变清单及内容哈希。
- 确定性规则引擎：运行在 Lambda、ECS 或批处理任务中，对每条记录执行同一版本的规则。
- 结果与证据层：将扫描清单、逐条裁决和证据引用保存到 S3 或 DynamoDB，并使用 KMS 加密。
- 受限 MCP 服务器：只暴露少量面向合规任务的工具，不提供任意 SQL、Shell 或通用对象读取能力。
- Amazon Quick 对话层：启动扫描、轮询状态、读取摘要，并把例外项解释给业务人员。
MCP 工具可以控制在类似下面的范围内：
start_compliance_scan(snapshot_id, rule_set_version) -> scan_id
get_scan_summary(scan_id) -> counts, hashes, timestamps
list_exceptions(scan_id, status, cursor) -> paginated results
get_evidence(scan_id, lease_id) -> rule hits and source references
设计工具时，应让分页和完整性成为服务器端协议的一部分。例如 list_exceptions 必须返回 next_cursor 和 total_count；Quick 可以总结结果，但不能因为上下文窗口已满就把剩余页面当作不存在。
AWS CDK 可以负责部署存储桶、计算资源、结果表、日志、密钥以及最小权限角色。实际资源组合可按吞吐量选择：数千份结构化记录可能由 Lambda 完成，更大的文档集或计算密集型解析任务则可以放到 ECS 或批处理系统中。无论选择哪种计算服务，核心边界都相同：Quick 调用任务级工具，规则引擎读取固定快照，结果存储保留审计证据。
可以直接运行的最小裁决引擎
下面是一个本地可运行的参考实现。它检查两个示例规则：通知期不得少于 30 天，保险到期日不得早于租约结束日。缺少关键字段、日期格式错误或租约编号重复时，记录会进入人工复核，而不是被静默忽略。
这些规则只是演示假设，运行前应替换成经过法务或合规团队批准的规则。
将以下内容保存为 compliance_scan.py：
#!/usr/bin/env python3
import argparse, collections, datetime as dt, hashlib, json
from pathlib import Path
RULE_SET = "lease-policy-demo-1"
REQUIRED = {"lease_id", "notice_days", "term_end", "insurance_expiry"}
def parse_date(value):
return dt.date.fromisoformat(value)
def adjudicate(record, duplicate_ids):
lease_id = record.get("lease_id", "<missing>")
missing = sorted(REQUIRED - record.keys())
if missing:
return {"lease_id": lease_id, "status": "NEEDS_REVIEW",
"findings": [{"rule": "REQUIRED_FIELDS", "missing": missing}]}
if lease_id in duplicate_ids:
return {"lease_id": lease_id, "status": "NEEDS_REVIEW",
"findings": [{"rule": "UNIQUE_LEASE_ID"}]}
findings = []
try:
if int(record["notice_days"]) < 30:
findings.append({"rule": "MIN_NOTICE_DAYS", "expected": 30,
"actual": int(record["notice_days"])})
if parse_date(record["insurance_expiry"]) < parse_date(record["term_end"]):
findings.append({"rule": "INSURANCE_THROUGH_TERM",
"term_end": record["term_end"],
"insurance_expiry": record["insurance_expiry"]})
except (TypeError, ValueError) as exc:
return {"lease_id": lease_id, "status": "NEEDS_REVIEW",
"findings": [{"rule": "INVALID_FORMAT", "detail": str(exc)}]}
return {"lease_id": lease_id,
"status": "NON_COMPLIANT" if findings else "COMPLIANT",
"findings": findings}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("input", help="UTF-8 JSON Lines file")
parser.add_argument("output", help="output report JSON")
args = parser.parse_args()
raw = Path(args.input).read_bytes()
records = [json.loads(line) for line in raw.decode().splitlines() if line.strip()]
id_counts = collections.Counter(r.get("lease_id") for r in records)
duplicate_ids = {key for key, count in id_counts.items() if key and count > 1}
results = [adjudicate(record, duplicate_ids) for record in records]
status_counts = collections.Counter(item["status"] for item in results)
report = {
"manifest": {
"rule_set": RULE_SET,
"source_sha256": hashlib.sha256(raw).hexdigest(),
"source_records": len(records),
"evaluated_records": len(results),
"unique_lease_ids": len({r.get("lease_id") for r in records if r.get("lease_id")}),
"duplicate_lease_ids": sorted(duplicate_ids),
"status_counts": dict(status_counts)
},
"results": results
}
Path(args.output).write_text(json.dumps(report, indent=2), encoding="utf-8")
if __name__ == "__main__":
main()
创建一组测试数据并执行扫描：
cat > leases.jsonl <<'EOF'
{"lease_id":"L-1001","notice_days":60,"term_end":"2027-12-31","insurance_expiry":"2027-12-31"}
{"lease_id":"L-1002","notice_days":15,"term_end":"2028-06-30","insurance_expiry":"2028-01-31"}
{"lease_id":"L-1003","notice_days":30,"term_end":"2027-03-31"}
EOF
python3 compliance_scan.py leases.jsonl report.json
python3 -m json.tool report.json
输出清单中的 source_records 与 evaluated_records 应完全相等。示例中的 L-1001 会通过，L-1002 会触发两条违规规则，L-1003 因缺少保险到期日进入人工复核。
部署到 AWS 时，可以把输入文件替换为带版本号的 S3 对象，把 RULE_SET 绑定到不可变的构建版本，并将输出写入以 scan_id 分区的结果位置。MCP 服务器只需要启动这个工作流并读取已经生成的结果，不应让代理在请求中提交任意 Python 表达式或规则代码。
审计证据比聊天答案更重要
面向用户的答案可以很简洁，例如“共检查 8,412 份租约，137 份不合规，24 份需要人工复核”。但底层扫描记录应更严格，至少保留：
- 数据快照标识、对象版本或清单哈希；
- 规则集版本、程序构建版本和执行时间；
- 输入数、成功裁决数、解析失败数与重复记录数；
- 每条规则命中的实际值、期望值和源文档引用；
- Quick 发起的工具调用、调用者身份和扫描 ID；
- 人工复核后的决定、复核人和变更原因。
不要只保存最终摘要。摘要无法回答“这份租约为什么被判定为不合规”，也无法在规则升级后重放历史扫描。对于包含敏感商业条款的租约，还要限制证据接口返回的字段，并设置日志脱敏、加密和保留期限。
上线前的检查清单
采用这一模式时，可以用以下问题做设计评审：
- 扫描的数据范围是否由不可变快照定义？
- 是否能证明每条输入恰好得到一个结果？
- 规则是否确定、带版本且经过业务批准？
- MCP 是否只暴露任务级工具，而不是任意查询能力？
- 分页、超时和部分失败是否会被明确报告？
NEEDS_REVIEW是否有负责人和处理时限？- 能否根据扫描 ID 重建当时的输入、规则和证据？
- Quick 的回答是否区分事实结果、规则解释和法律判断？
最稳妥的采用路径是先选择少量结构化程度高、规则明确的条款进行影子扫描，与现有人工流程并行比较。确认覆盖率、误报率和证据质量后，再扩展到更多租约类型。聊天代理提升的是访问和解释效率；真正让结果经得起审计的，是固定范围、确定性裁决、受限工具和完整证据链。