📢 转载信息
原文作者：AWS Machine Learning Blog
预测性维护 (PdM) 涉及利用数据分析和机器学习来检测设备故障的早期迹象，以便在发生故障之前进行维护。尽管 PdM 解决方案可以显著提高运营效率并降低成本，但在工业环境中，确定故障的根本原因 (RCA) 仍然是一项具有挑战性的任务。
这是因为工业故障通常是由于多模态数据来源的复杂交互造成的，包括传感器读数、维护日志、操作员笔记和设备图像。这些异构数据源的整合和分析，通常需要领域专家花费大量时间进行人工检查和推断。
Amazon Bedrock 上的多模态能力使我们能够构建一个用于 PdM 的生成式 AI 助手，该助手可以摄取和分析文本、图像和结构化数据，以实现自动 RCA。
使用 Amazon Bedrock 构建多模态 PdM 助手
在此博客文章中，我们将介绍一个架构，该架构使用 Amazon Bedrock 构建一个多模态生成式 AI 助手，用于对工业设备故障进行 RCA。此解决方案利用 Amazon Bedrock 的多模态能力（尤其是Anthropic Claude 3 Sonnet 模型）来处理和分析文本、图像和结构化数据，以提供准确的 RCA。
以下是解决方案的主要组件：
- 数据摄取与处理：Ingest 结构化数据、文本日志和图像。
- 向量存储：使用 Amazon OpenSearch Service 存储和检索嵌入的向量。
- 检索增强生成 (RAG)：使用 Amazon Bedrock 的代理功能增强生成式 AI 模型的知识。
- 多模态推理：使用 Claude 3 Sonnet 分析图像和文本输入。
- 应用程序界面：一个简单的 Streamlit 界面，用于与助手交互。
图 1：PdM 助手的架构概述
先决条件
您需要一个配置好的 Amazon Bedrock 访问权限，并且能够访问 Anthropic Claude 3 Sonnet 模型。此外，您需要一个设置了 OpenSearch 域的 Amazon OpenSearch Service 实例。
我们使用以下 Python 库：
pip install boto3 langchain-aws langchain-community faiss-cpu streamlit matplotlib Pillow注意：为了简化演示，我们使用 faiss-cpu 进行本地向量存储，但对于生产环境，我们建议使用 Amazon OpenSearch Service 或 Amazon Aurora PostgreSQL 上的 pgvector。
步骤 1：准备和摄取数据
我们的 PdM 助手需要访问以下类型的数据来执行 RCA：
- 结构化数据：机器 ID、时间戳、传感器读数（例如振动、温度）。
- 文本日志：来自维护记录或操作员报告的非结构化文本。
- 图像：设备故障的图像。
我们首先定义一个代表设备故障事件的简单数据集。
import pandas as pd
import base64
from PIL import Image
import io
# 模拟结构化数据
data = {
'event_id': ['E001', 'E002', 'E003'],
'timestamp': ['2024-01-15 10:00:00', '2024-01-16 14:30:00', '2024-01-17 08:45:00'],
'machine_id': ['M101', 'M102', 'M101'],
'vibration_reading': [5.5, 1.2, 8.9], # High vibration in E003
'temperature': [75.0, 60.5, 92.1] # High temperature in E003
}
df = pd.DataFrame(data)
# 模拟文本日志
text_logs = {
'E001': 'Routine check successful. Minor oil leak observed.',
'E002': 'Normal operation.',
'E003': 'Excessive noise reported near the bearing assembly. Operator noted visual discoloration.'
}
# 模拟图像数据（需要替换为实际的图像文件路径或 Base64 编码）
# 在实际应用中，您会加载图像并将其编码为 Base64 或上传到 S3。
# 对于演示，我们创建一个占位符图像
def create_placeholder_image(event_id):
img = Image.new('RGB', (200, 100), color = (73, 109, 137))
d = ImageDraw.Draw(img)
d.text((10,10), f"Image for {event_id}", fill=(255,255,0))
buffered = io.BytesIO()
img.save(buffered, format="PNG")
img_str = base64.b64encode(buffered.getvalue()).decode('utf-8')
return f"data:image/png;base64,{img_str}"
from PIL import ImageDraw
image_data = {
'E001': create_placeholder_image('E001'),
'E002': create_placeholder_image('E002'),
'E003': create_placeholder_image('E003')
}
print("Data preparation complete.")
嵌入和向量化
为了利用 RAG 机制，我们需要将文本和结构化数据转换为向量嵌入。我们使用 AmazonEmbeddings (AWS Embeddings) 来生成嵌入。
import boto3
from langchain_community.embeddings import BedrockEmbeddings
# 初始化 Bedrock 客户端
bedrock_client = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
embeddings_model = BedrockEmbeddings(
client=bedrock_client,
model_id="amazon.titan-embed-text-v1"
)
# 准备要嵌入的文本数据（组合日志和结构化数据描述）
docs_to_embed = []
metadata = []
for index, row in df.iterrows():
event_id = row['event_id']
log = text_logs.get(event_id, "No specific text log found.")
# 组合描述性文本
context = (
f"Event ID: {event_id}. Machine ID: {row['machine_id']}. "
f"Vibration Reading: {row['vibration_reading']}. Temperature: {row['temperature']}C. "
f"Log Entry: {log}. "
f"Image reference provided."
)
docs_to_embed.append(context)
metadata.append({
'event_id': event_id,
'machine_id': row['machine_id'],
'vibration': row['vibration_reading'],
'temp': row['temperature'],
'image_base64': image_data[event_id] # Store image ref for retrieval
})
# 生成嵌入并存储（此处使用内存中的 FAISS 模拟向量存储）
# 在生产中，您会将其存储到 OpenSearch 中
from langchain_community.vectorstores import FAISS
print("Generating embeddings...")
embeddings = embeddings_model.embed_documents(docs_to_embed)
vectorstore = FAISS.from_embeddings(zip(docs_to_embed, embeddings), embeddings_model)
print("Embeddings created and stored (in-memory FAISS).")
步骤 2：构建多模态 RAG 链
为了执行 RCA，我们需要一个能够处理查询、检索相关信息并使用 Claude 3 Sonnet 进行推理的机制。我们将使用 LangChain 来协调这一过程，特别是利用其对多模态模型（如 Claude 3）的支持。
from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage
from langchain.prompts import ChatPromptTemplate
# 1. 初始化多模态 LLM
# 确保您的 AWS 区域已启用 Claude 3 Sonnet
llm = ChatBedrock(
model_id="anthropic.claude-3-sonnet-20240229-v1:0",
model_kwargs={'temperature': 0.1, 'max_tokens': 2048}
)
# 2. 定义系统提示词，指导模型执行 RCA 任务
SYSTEM_PROMPT = (
"You are an expert Industrial Maintenance Root Cause Analysis (RCA) Assistant. "
"Analyze the provided sensor data, text logs, and images to determine the most likely root cause of the failure event. "
"Base your analysis strictly on the retrieved context. If no context is provided, state that you cannot determine the cause."
"Format your output clearly, stating the recommended next steps based on the root cause.\n\n"
"Context Analysis: {context}"
)
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
("human", "Analyze the following event: {query}")
])
# 3. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) # 检索最相关的 2 个文档
# 4. 构建链条（使用 LCEL - LangChain Expression Language）
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
# 检索上下文
retrieval_chain = (
{"context": retriever | (lambda docs: "\n---\n".join([d.page_content for d in docs]))}
| prompt
| llm
| StrOutputParser()
)
print("RAG chain for text/structured data analysis is ready.")
步骤 3：集成图像分析（多模态输入）
现在我们来实现核心的多模态能力：将图像信息与文本查询一起发送给 Claude 3 Sonnet。
我们模拟一个用户查询，其中包含文本描述和一个与事件相关的图像（Base64 编码）。
# 假设我们针对事件 E003 进行查询，并使用其对应的图像数据
QUERY_EVENT_ID = 'E003'
# 检索该事件的完整上下文和图像数据
relevant_metadata = next(item for item in metadata if item['event_id'] == QUERY_EVENT_ID)
# 准备查询文本
user_query_text = f"Analyze machine {relevant_metadata['machine_id']} for event {QUERY_EVENT_ID}. What is the likely root cause?"
# 准备图像内容 (Claude 3 需要特定的内容格式)
image_content = {
"type": "image_url",
"image_url": {
"url": relevant_metadata['image_base64'] # Claude 3 支持 Base64 输入
}
}
# 准备文本内容
text_content = {
"type": "text",
"text": f"{SYSTEM_PROMPT.format(context='Retrieved via RAG...')} {user_query_text}"
}
# 构建多模态消息
human_message = HumanMessage(
content=[
text_content,
image_content
]
)
# 对于多模态输入，我们不能直接使用我们之前定义的基于 RAG 的文本链，
# 因为 RAG 链主要设计用于将检索到的文本注入到系统提示词中。
# 对于多模态，我们直接构建一个直接调用 LLM 的消息。
# 重新定义系统提示词以适应直接多模态输入格式（简化起见，我们在此处直接组合上下文和查询）
MULTIMODAL_SYSTEM_PROMPT = (
"You are an expert Industrial Maintenance Root Cause Analysis (RCA) Assistant. "
"Analyze the provided sensor data, text logs (Context: {context}), and the attached image to determine the most likely root cause of the failure event. "
"Format your output clearly, stating the recommended next steps based on the root cause."
)
# 检索相关的文本上下文以丰富分析
retrieved_docs = vectorstore.similarity_search(user_query_text, k=2)
context_text = "\n---\n".join([doc.page_content for doc in retrieved_docs])
final_system_prompt = MULTIMODAL_SYSTEM_PROMPT.format(context=context_text)
# 创建最终的多模态请求
final_messages = [
("system", final_system_prompt),
("human", [
{"type": "text", "text": f"Analyze Event {QUERY_EVENT_ID}: {user_query_text}"},
image_content
])
]
print(f"\n--- Sending Multimodal Request for Event {QUERY_EVENT_ID} to Claude 3 Sonnet ---")
# 调用 LLM (LangChain 封装)
# 注意：LangChain 的 ChatBedrock 对于多模态消息的处理需要确保模型ID正确，
# 并且输入格式符合 Claude 3 的 API 要求 (Messages API)。
# 简化调用（直接使用 boto3 确保格式正确，因为 LangChain 的抽象层可能需要更多配置）
response = bedrock_client.invoke_model(
modelId="anthropic.claude-3-sonnet-20240229-v1:0",
body=json.dumps({
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": final_system_prompt + "\n" + user_query_text
},
image_content
]
}
],
"max_tokens": 2048,
"temperature": 0.1
})
)
import json
response_body = json.loads(response.get('body').read())
rca_result = response_body['content'][0]['text']
print("\n--- RCA Result ---")
print(rca_result)
在上述代码片段中，我们演示了如何构建一个包含结构化数据、文本日志（通过 RAG 检索）和设备图像的多模态输入。
结果分析
通过结合这些数据源，生成式 AI 助手能够推断出比仅使用单一数据源更准确的根本原因。例如，如果传感器数据显示高振动（结构化数据），日志提到“异响”（文本），而图像显示轴承部位有明显的烧蚀痕迹（图像），Claude 3 能够综合这些线索，将根本原因定性为轴承故障导致过度磨损，而不是仅仅报告高振动。
rca_result 变量现在包含了 AI 根据所有输入信息得出的详细诊断和建议。
总结
使用 Amazon Bedrock 和 Claude 3 Sonnet，我们可以构建强大的多模态 AI 助手，显著简化预测性维护中的根本原因诊断过程。这种方法通过自动化复杂的数据分析和模式识别，使维护团队能够更快地响应故障，从而实现更高的设备正常运行时间。
🚀 想要体验更好更全面的AI调用？
欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。
评论区