公共部门积累的证据往往不是整齐的数据库记录，而是执法记录仪视频、扫描表格、PDF、图片和录音。真正困难的环节不只是“让模型读懂文件”，而是把提取结果转换成可追溯、受权限控制的结构化数据，再让业务人员通过 Salesforce Agentforce 用自然语言查询。
一种可行的组合方式是：使用 Amazon Bedrock Data Automation 处理非结构化材料，通过 Model Context Protocol（MCP）将结构化结果暴露为受控工具，最后由 Agentforce 调用这些工具回答问题或触发后续流程。
从文件处理走向可追溯的查询链路
这套架构可以拆成五个环节：
- 证据进入受控存储：视频、扫描件和其他文件进入 Amazon S3 等受控存储，并记录案件编号、上传者、时间和保留策略。
- 提取结构化信息：Amazon Bedrock Data Automation 分析材料，产出文本、字段、事件或其他机器可处理的结果。
- 统一证据模型：将不同来源的输出规范化为统一 JSON，并保留原始对象位置、页码、时间点和处理版本。
- 通过 MCP 发布工具：查询服务不直接把数据库交给智能体，而是提供
search_evidence、get_evidence_item等边界清晰的工具。 - Agentforce 编排交互：用户用自然语言提问，Agentforce 选择工具、携带案件上下文，并将结果组织成业务可读的回答。
这里最重要的设计不是模型本身，而是证据与回答之间的追溯关系。每条结构化记录至少应包含来源标识；视频结果可以保留时间范围，扫描件结果可以保留页码。模型生成的总结不能替代原始证据，也不应抹掉来源信息。
一个推荐的数据形态可以这样设计：
{
"evidence_id": "EV-1042",
"case_id": "CASE-2025-017",
"source_type": "body_camera",
"captured_at": "2025-02-18T14:32:10Z",
"summary": "A red vehicle stopped near the east entrance.",
"entities": ["red vehicle", "east entrance"],
"source_reference": {
"object_uri": "s3://evidence-bucket/cases/CASE-2025-017/cam-03.mp4",
"start_second": 184,
"end_second": 211
},
"processing_version": "v1"
}
这只是可以采用的规范化模型，并不代表必须使用这些字段。实际项目还应加入分类级别、访问范围、保留期限、校验值和人工复核状态。
为什么要在 Agentforce 与数据层之间加入 MCP
如果智能体可以自由拼接 SQL、读取任意对象，权限控制和审计会迅速失控。MCP 更适合充当能力边界：它把后端功能封装成带名称、参数和返回结构的工具，智能体只能调用明确开放的操作。
例如，与其开放一个通用数据库查询入口，不如定义下面几类窄工具：
search_evidence：在用户有权访问的案件中搜索摘要和实体；get_evidence_item：按证据编号读取详情和来源引用；build_case_timeline：生成带时间和出处的事件序列；request_human_review：把低置信度结果提交给审核人员，而不是自动写回正式记录。
这种设计也方便实施字段级过滤。普通办事人员可能只能看到摘要，调查人员可以看到来源位置，而下载原始文件则需要更高权限和额外审计。
MCP 工具返回的内容也应该尽量是事实记录，而不是直接生成结论。最终措辞可以由 Agentforce 组织，但事实、引用和权限判断应由后端服务控制。
可运行的最小 MCP 证据查询服务
下面的示例不直接调用 Amazon Bedrock Data Automation，而是假设其结果已经被规范化为本地 JSON。这样可以先验证 MCP 工具契约，再把文件读取替换为 DynamoDB、OpenSearch 或其他受控数据服务。
创建目录并安装 Python MCP SDK：
mkdir evidence-mcp && cd evidence-mcp
python -m venv .venv
source .venv/bin/activate
python -m pip install "mcp[cli]"
创建 evidence.json：
[
{
"evidence_id": "EV-1042",
"case_id": "CASE-2025-017",
"source_type": "body_camera",
"summary": "A red vehicle stopped near the east entrance.",
"entities": ["red vehicle", "east entrance"],
"source_reference": {
"object_uri": "s3://example-evidence/CASE-2025-017/cam-03.mp4",
"start_second": 184,
"end_second": 211
}
},
{
"evidence_id": "EV-1043",
"case_id": "CASE-2025-017",
"source_type": "scanned_document",
"summary": "The intake form records the vehicle color as red.",
"entities": ["red vehicle", "intake form"],
"source_reference": {
"object_uri": "s3://example-evidence/CASE-2025-017/intake.pdf",
"page": 2
}
}
]
然后创建 server.py：
import json
from pathlib import Path
from typing import Any
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("public-sector-evidence")
DATA_FILE = Path(__file__).with_name("evidence.json")
def load_records() -> list[dict[str, Any]]:
return json.loads(DATA_FILE.read_text(encoding="utf-8"))
@mcp.tool()
def search_evidence(
query: str,
case_id: str,
limit: int = 5,
) -> list[dict[str, Any]]:
"""Search normalized evidence records within one authorized case."""
terms = [term.lower() for term in query.split() if term.strip()]
matches = []
for record in load_records():
if record["case_id"] != case_id:
continue
searchable = " ".join(
[record["summary"], *record.get("entities", [])]
).lower()
if all(term in searchable for term in terms):
matches.append(
{
"evidence_id": record["evidence_id"],
"summary": record["summary"],
"source_type": record["source_type"],
"source_reference": record["source_reference"],
}
)
return matches[: max(1, min(limit, 20))]
@mcp.tool()
def get_evidence_item(
evidence_id: str,
case_id: str,
) -> dict[str, Any]:
"""Return one evidence item, restricted to the supplied case."""
for record in load_records():
if (
record["evidence_id"] == evidence_id
and record["case_id"] == case_id
):
return record
return {"error": "Evidence item not found or access not permitted"}
if __name__ == "__main__":
mcp.run()
可用 MCP 开发工具启动并检查服务：
mcp dev server.py
测试时可以调用：
search_evidence(
query="red vehicle",
case_id="CASE-2025-017",
limit=5
)
这个演示只实现了案件编号过滤，不应被视为生产级授权。接入 Agentforce 时，需要按照所选 MCP 传输方式和 Salesforce 环境支持的连接机制注册工具；不同部署方式的认证与网络配置可能不同。生产系统还应从经过验证的身份令牌中解析案件权限，而不能相信模型传来的 case_id。
把提取结果交给智能体之前，还要补齐什么
1. 让引用成为强制返回字段
如果工具只返回一段摘要，使用者无法验证答案。应让 source_reference 成为固定字段，并要求 Agentforce 在回答中展示证据编号、页码或视频时间点。找不到出处时，智能体应明确说明“现有结构化结果不足以回答”。
2. 把提示注入视为数据安全问题
扫描文档或视频字幕中可能出现“忽略之前指令”之类文本。它们属于证据内容，不是系统指令。MCP 服务应返回结构化字段，Agentforce 的系统规则也应明确禁止将证据文本解释为工具调用命令。
3. 分开原始数据、提取结果与生成结论
三者应使用不同的数据层和保留策略：
- 原始证据用于复核与合规留存；
- 自动提取结果允许重新处理和版本升级；
- 智能体回答属于派生内容，需要记录所用工具、参数和引用。
这种分层可以避免模型升级后覆盖历史结果，也能解释某次回答为何与后续回答不同。
4. 为高风险操作保留人工确认
自然语言查询适合检索、汇总和定位材料，但不应直接决定执法、福利资格、调查结论或其他高影响事项。凡是会修改正式记录、向外部发送通知或触发处置的操作，都应加入人工确认和独立授权。
上线前的检查清单
采用 Agentforce、Amazon Bedrock Data Automation 与 MCP 时，可以按下面的顺序推进：
- 先选择一个材料类型和一个明确问题，不要一开始覆盖所有档案；
- 定义统一的证据 JSON，并保留对象位置、页码或视频时间点；
- 让 MCP 工具保持窄接口，禁止任意 SQL 和任意对象读取；
- 在服务端实施用户、角色、案件和字段级授权；
- 记录每次工具调用的身份、参数、结果版本和关联会话；
- 对敏感字段做脱敏，并验证数据驻留与保留要求；
- 用人工标注样本评估漏检、误提取和错误关联；
- 对外部动作、正式结论和低置信度结果设置人工审批。
这套组合的价值不在于让聊天界面“知道所有事情”，而在于建立一条从自然语言问题到结构化记录、再回到原始证据的受控路径。只有当答案可追溯、权限可执行、操作可审计时，智能体才适合进入公共部门的真实工作流。