用 Amazon Bedrock AgentCore 搭一个农机维修 AI 助手

用户语音/文字输入
    → AgentCore Runtime (Strands Agent 编排)
        → 调用 Knowledge Base 检索维修文档 (RAG)
        → 调用 Memory 读取历史对话上下文
        → Nova 2 Lite 生成诊断建议 + 零件编号 + 维修步骤
    → 返回结构化回复

# repair_agent.py — 农机维修助手 Agent 定义
# 前置依赖：pip install strands-agents-sdk boto3

from strands_agents_sdk import Agent, tool
import boto3
import json

# ---------- 工具定义 ----------

@tool(name="query_knowledge_base", description="从维修手册和零件目录中检索相关信息")
def query_knowledge_base(query: str, top_k: int = 5) -> str:
    """调用 Bedrock Knowledge Base 进行 RAG 检索"""
    kb_client = boto3.client("bedrock-agent-runtime")

    response = kb_client.retrieve(
        knowledgeBaseId="YOUR_KB_ID",       # 替换为你的 Knowledge Base ID
        retrievalQuery={"text": query},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": top_k
            }
        }
    )

    results = []
    for r in response.get("retrievalResults", []):
        content = r["content"]["text"]
        source = r.get("location", {}).get("s3Location", {}).get("uri", "未知来源")
        results.append(f"【来源: {source}】\n{content}")

    return "\n---\n".join(results) if results else "未找到相关维修信息。"


@tool(name="lookup_part_number", description="根据设备型号和故障描述查找替换零件编号")
def lookup_part_number(equipment_model: str, symptom: str) -> str:
    """从零件目录中匹配零件编号"""
    # 组合更精准的检索查询
    refined_query = f"{equipment_model} {symptom} 替换零件 零件编号"
    return query_knowledge_base(refined_query, top_k=3)


@tool(name="get_repair_procedure", description="获取厂商批准的维修操作步骤")
def get_repair_procedure(equipment_model: str, component: str) -> str:
    """检索标准维修流程"""
    procedure_query = f"{equipment_model} {component} 维修步骤 操作规程"
    return query_knowledge_base(procedure_query, top_k=5)


# ---------- Agent 定义 ----------

repair_agent = Agent(
    name="equipment-repair-assistant",
    model="us.amazon.nova2-lite-v1:0",       # Nova 2 Lite 模型 ID
    instructions=[
        "你是一名专业的农机设备维修助手，服务于农民和现场维修技师。",
        "诊断流程：先确认设备型号和故障现象，再检索相关维修文档，最后给出诊断结论和操作建议。",
        "所有维修步骤必须引用厂商批准的规程，不得自行编造。",
        "如果信息不足，主动追问用户：设备型号、故障代码、异常声音/烟雾等细节。",
        "回复中必须包含：1) 可能的故障原因 2) 所需替换零件编号 3) 维修步骤摘要 4) 安全警告。",
    ],
    tools=[query_knowledge_base, lookup_part_number, get_repair_procedure],
)

# ---------- 运行交互 ----------

if __name__ == "__main__":
    # 单轮测试
    result = repair_agent.run(
        "我的 John Deere 8R 拖拉机液压系统压力不足，仪表显示故障代码 E-HYD-042"
    )
    print(result)

    # 多轮对话（Memory 由 AgentCore Runtime 在部署模式下自动管理）
    # 本地开发模式下可手动传入 conversation_id 模拟持久化

# 创建 Knowledge Base（需先在 S3 上传文档并创建向量存储）
aws bedrock-agent create-knowledge-base \
  --name "equipment-repair-kb" \
  --description "农机维修手册与零件目录知识库" \
  --knowledge-base-configuration '{
    "type": "VECTOR",
    "vectorKnowledgeBaseConfiguration": {
      "embeddingModelArn": "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0"
    }
  }' \
  --roleArn "arn:aws:iam::YOUR_ACCOUNT:role/BedrockKBRole"

# 记下返回的 knowledgeBaseId，填入上面的 Python 代码

# 关联数据源（指向 S3 文档目录）
aws bedrock-agent create-data-source \
  --knowledge-base-id "YOUR_KB_ID" \
  --name "repair-manuals" \
  --data-source-configuration '{
    "type": "S3",
    "s3Configuration": {
      "bucketArn": "arn:aws:s3:::kb-docs",
      "prefixes": ["manuals/", "parts-catalog/", "procedures/"]
    }
  }'

# 触发同步（将文档向量化入库）
aws bedrock-agent start-ingestion-job \
  --knowledge-base-id "YOUR_KB_ID" \
  --data-source-id "YOUR_DS_ID"

# 在 Agent 定义中增加 Memory 配置（部署模式）
from strands_agents_sdk import MemoryConfig

repair_agent = Agent(
    name="equipment-repair-assistant",
    model="us.amazon.nova2-lite-v1:0",
    instructions=[...],
    tools=[...],
    memory=MemoryConfig(
        strategy="CONVERSATION_SUMMARY",   # 对话摘要策略，节省 token
        max_history_turns=20,              # 保留最近 20 轮
    ),
)

# 创建 AgentCore Runtime 应用
aws bedrock-agent-core create-app \
  --name "repair-assistant-app" \
  --runtime-type "STRANDS_AGENT"

# 上传 Agent 代码包
aws bedrock-agent-core deploy-agent \
  --app-id "YOUR_APP_ID" \
  --agent-entry-point "repair_agent.py" \
  --s3-code-uri "s3://agent-code/repair-agent.zip"

# 获取调用端点
aws bedrock-agent-core get-app-endpoint --app-id "YOUR_APP_ID"