大量 ITSM 事件工单在关闭后便进入“只存不看”的归档区,但其中往往记录了真实故障、排查路径、错误尝试和最终修复方案。KnowledgeForge 的思路是把这些已解决工单重新送入一条自动化流水线:用 Amazon Bedrock 提炼知识,用 Amazon S3 Vectors 检索相似内容,再由 AWS Step Functions 编排去重、评分、改写和发布,最终形成一个多租户、可持续反馈的知识管理闭环。
工单不是文章,直接摘要远远不够
一张已关闭工单通常包含时间线、告警文本、客服回复、工程师内部备注以及重复粘贴的日志。即便工单状态是“已解决”,也不代表内容已经满足知识库要求。
从工单生成文章时,至少需要完成四类判断:
- 可复用性:问题是否可能再次发生,解决方案是否只对某一台机器有效。
- 完整性:是否包含症状、适用条件、根因、操作步骤和验证方法。
- 安全性:是否混入用户名、访问令牌、内部地址或客户数据。
- 新颖性:知识库中是否已有相同或高度相似的文章。
因此,Bedrock 在这里不应只是“摘要器”。更合适的做法是让模型输出结构化候选文章以及质量判断,例如:
你是 ITSM 知识库编辑。请根据已解决事件生成候选知识文章。
必须遵守:
1. 不补充工单中没有证据支持的根因或命令。
2. 删除个人信息、凭据、客户标识和内部主机名。
3. 区分“已验证步骤”和“待确认建议”。
4. 输出 JSON,字段为 title、symptoms、scope、cause、resolution、verification、warnings。
5. 如果信息不足以形成文章,将 publishable 设为 false,并列出 missing_information。
事件内容:
{{SANITIZED_TICKET}}
结构化输出让后续状态机能够明确分支,而不是依赖自然语言中的模糊结论。例如,publishable=false 的候选项可以进入人工补充队列,存在高风险警告的内容则必须经过审批。
S3 Vectors 解决的是“该不该新建”
生成一篇措辞流畅的文章并不难,难的是判断它应该成为新文章、合并到旧文章,还是仅用于改进旧文章。KnowledgeForge 使用 Amazon S3 Vectors 支持向量检索,正适合处理这种语义重复问题。
流水线可以在租户范围内检索最相似的现有文章,再根据相似度和内容差异选择动作:
| 检索结果 | 建议动作 |
|---|---|
| 没有相近文章 | 创建候选文章 |
| 高度相似,只有措辞差异 | 标记为重复,不发布 |
| 主题相同,但新工单包含额外步骤 | 生成旧文章的改进版本 |
| 相似度处于灰区 | 交给知识管理员复核 |
多租户环境中的关键点不是在检索后过滤结果,而是在写入和查询阶段都强制带上租户边界。向量记录、原始工单、候选文章和审批任务都应包含稳定的 tenant_id。如果底层索引设计无法可靠隔离租户,应使用独立索引、存储前缀或独立 AWS 账户,而不是只相信提示词中的“不要访问其他租户”。
向量相似度也不能独自决定发布。两篇文章可能描述相同错误码,却分别适用于不同产品版本;它们在语义上接近,在操作上却不能合并。版本、平台、服务名称和环境等结构化元数据必须参与判断。
用 Step Functions 把生成过程变成可审计工作流
AWS Step Functions 适合承载这类多阶段流程,因为每一步都有不同的失败方式和重试策略。一个典型执行链可以包含:
- 从归档区读取已解决工单并确认租户。
- 脱敏和规范化工单内容。
- 调用 Bedrock 生成结构化候选文章。
- 计算嵌入并在 S3 Vectors 中检索近邻。
- 执行重复检测和质量评分。
- 根据风险、分数和相似度选择发布、改进或人工复核。
- 发布后记录文章版本,并收集搜索点击、解决率和编辑反馈。
可以这样实践一个本地可运行的“前置决策器”。它不调用 AWS 服务,而是演示租户隔离、文本去重和质量门槛;接入生产环境时,可把 similarity 替换为 S3 Vectors 查询,把候选生成步骤替换为 Bedrock 调用。
将下面内容保存为 knowledge_gate.py,使用 Python 3.10 或更高版本运行:
import re
from dataclasses import dataclass
@dataclass
class Article:
tenant_id: str
title: str
symptoms: str
resolution: str
verification: str
def tokens(text: str) -> set[str]:
return set(re.findall(r"[a-z0-9_./-]+", text.lower()))
def similarity(left: Article, right: Article) -> float:
a = tokens(f"{left.title} {left.symptoms} {left.resolution}")
b = tokens(f"{right.title} {right.symptoms} {right.resolution}")
return len(a & b) / len(a | b) if a or b else 0.0
def quality_score(article: Article) -> int:
score = 0
score += 25 if len(article.title) >= 12 else 0
score += 25 if len(article.symptoms) >= 30 else 0
score += 30 if len(article.resolution) >= 50 else 0
score += 20 if len(article.verification) >= 25 else 0
return score
def decide(candidate: Article, library: list[Article]) -> dict:
# Tenant filtering must happen before similarity comparison.
same_tenant = [a for a in library if a.tenant_id == candidate.tenant_id]
nearest = max(same_tenant, key=lambda a: similarity(candidate, a), default=None)
nearest_score = similarity(candidate, nearest) if nearest else 0.0
quality = quality_score(candidate)
if quality < 70:
action = "request_more_information"
elif nearest_score >= 0.80:
action = "deduplicate_or_merge"
elif nearest_score >= 0.45:
action = "human_review"
else:
action = "publish_candidate"
return {
"tenant_id": candidate.tenant_id,
"quality_score": quality,
"nearest_title": nearest.title if nearest else None,
"similarity": round(nearest_score, 3),
"action": action,
}
library = [
Article(
tenant_id="tenant-a",
title="Restart a stalled invoice worker",
symptoms="Invoices remain queued and the worker health check times out.",
resolution="Drain the queue, restart the invoice worker, and confirm that new jobs are consumed.",
verification="Submit a test invoice and confirm completed status.",
)
]
candidate = Article(
tenant_id="tenant-a",
title="Recover an unresponsive invoice worker",
symptoms="Invoice jobs stay queued while worker health checks repeatedly time out.",
resolution="Pause intake, drain queued jobs, restart the invoice worker, and resume intake after health checks pass.",
verification="Submit one test invoice and verify that it reaches completed status without retrying.",
)
print(decide(candidate, library))
运行命令:
python3 knowledge_gate.py
这段代码中的阈值只是演示值,不能直接作为生产标准。生产系统应使用经过标注的历史样本校准阈值,并按文章类型、服务和语言分别评估误合并与漏检风险。
闭环的重点是修正知识,而不只是增加文章
KnowledgeForge 的价值不止是从工单中批量创建内容。它还会对现有知识库执行去重、质量评分和内容改进,这使流水线从一次性迁移工具变成持续维护机制。
闭环信号可以来自文章发布后的实际使用情况:搜索后是否点击、文章是否帮助事件解决、工程师是否回退修改、同类工单是否仍频繁升级。低使用率不一定意味着文章质量差,也可能是标题与用户搜索词不匹配;高点击率同样不代表解决步骤有效。因此,反馈指标应组合使用,并保留人工编辑记录。
上线时建议设置清晰边界:
- 默认生成草稿,只有低风险且达到质量门槛的内容才能自动发布。
- 在调用模型前完成凭据、个人信息和客户标识的确定性脱敏。
- 将原始工单、提示词版本、模型输出、相似文章和最终决策写入审计记录。
- 对删除、合并和覆盖旧文章采用版本化操作,保证能够回滚。
- 分租户监控成本、失败率、发布率和人工驳回率。
- 定期抽样检查“未生成文章”的工单,避免质量门槛长期过滤掉某类有效知识。
这类系统最危险的失败不是生成一篇文笔一般的文章,而是把未经验证的操作包装成权威步骤,或者在租户之间泄露内容。将 Bedrock 的生成能力置于检索、策略、审批和审计控制之中,才能真正把工单归档转化为可信、可维护的工程知识。