基于Strands框架和Bedrock AgentCore的SAP智能采购助手方案

# 系统提示词
SYSTEM_PROMPT = """你是一位专业的SAP可持续性采购顾问，专门从事供应链分析和可持续采购决策支持。

## 专业身份
作为采购领域的专家，你具备深厚的SAP系统经验和可持续性评估能力，能够为企业提供数据驱动的采购决策支持。

## 可用工具
1. **query_sap_data**: 查询SAP数据
   - 支持RFQ查询、报价统计等
   - 参数：query_type, year, month, days等
   - 示例：query_sap_data("monthly_rfqs", year=2025, month=7)

2. **query_sustainability_knowledge**: 查询可持续性知识库
   - 获取付款条件、认证信息等
   - 参数：query (查询问题)
   - 示例：query_sustainability_knowledge("供应商**的付款条件？")

3. **create_purchase_order**: 创建采购订单
   - 基于报价单创建采购订单
   - 参数：quotation_id (报价单编号)
   - 示例：create_purchase_order("6000000262")

## 工作流程
1. 理解用户需求
2. 选择合适的工具获取数据
3. **仔细检查和解析工具返回的数据**
4. 分析数据并提供洞察
5. 给出可执行的建议

# Agent配置
agent = Agent(
    model=model,
    tools=[
        query_sap_data,
        query_sustainability_knowledge,
        create_purchase_order
    ],
    system_prompt=SYSTEM_PROMPT
)
@app.entrypoint
def main(payload):
    """AgentCore Runtime入口点"""
    try:
        # 标准化输入处理
        user_input = payload.get("prompt") or payload.get("user_input", "")
        if not user_input:
            return "请提供您的问题或需求。
        logger.info(f"Processing request: {user_input[:100]}...")
        # 执行Agent
        result = agent(user_input)
        logger.info("Request processed successfully")
        return str(result)
    except Exception as e:
        error_msg = f"处理请求时发生错误: {str(e)}"
        logger.error(error_msg)
        return error_msg
if __name__ == "__main__":
    app.run()

""
SAP数据查询工具
"""
import boto3
import json
from typing import Dict, Any, Literal
from strands import tool
from ..config import settings
from ..utils import get_logger

logger = get_logger(__name__)

@tool
def query_sap_data(
    query_type: Literal[
        "monthly_rfqs", "recent_rfqs", "all_rfqs", "count_quotations", "quotation_summary",
        "lifecycle_status_query", "quotation_comparison", "material_detail_query", 
        "payment_terms_analysis", "data_validation_query", "purchasing_org_query",
        "amount_range_query", "incoterms_query", "manufacturer_query", "plant_query",
        "purchase_requisition_query", "exchange_rate_query"
    ],
    **kwargs
) -> Dict[str, Any]:
    """
    查询SAP数据API
    
    Args:
        query_type: 查询类型，支持的值：
            - monthly_rfqs: 查询指定年月的RFQ (需要year, month参数)
            - recent_rfqs: 查询最近的RFQ (需要days参数)  
            - all_rfqs: 查询所有RFQ (无需参数)
            - count_quotations: 统计报价数量 (需要requestforquotation参数)
            - quotation_summary: 报价汇总 (需要requestforquotation参数)
            - lifecycle_status_query: 生命周期状态查询 (需要status, limit参数)
            - quotation_comparison: 报价比较 (需要requestforquotation, limit参数)
            - material_detail_query: 物料详情查询 (需要requestforquotation, limit参数)
            - purchasing_org_query: 采购组织查询 (需要purchasingorganization, limit参数)
            - manufacturer_query: 制造商查询 (需要limit参数)
            - plant_query: 工厂查询 (需要limit参数)
            - purchase_requisition_query: 采购申请查询 (需要limit参数)
            - exchange_rate_query: 汇率查询 (需要limit参数)
        **kwargs: 查询参数
            - year (int): 年份 (用于monthly_rfqs)
            - month (int): 月份 (用于monthly_rfqs)
            - days (int): 天数 (用于recent_rfqs)
            - requestforquotation (str): RFQ编号 (用于quotations相关查询)
            - status (str): 生命周期状态 (用于lifecycle_status_query)
            - limit (int): 结果数量限制 (用于多个查询类型)
            - supplierquotation (str): 供应商报价单号 (用于data_validation_query)
            - purchasingorganization (str): 采购组织 (用于purchasing_org_query)
            - min_amount (float): 最小金额 (用于amount_range_query)
            - max_amount (float): 最大金额 (用于amount_range_query)
            - currency (str): 货币代码 (用于amount_range_query)

    Returns:
        Dict[str, Any]: 查询结果，包含success状态和data数据
        
    Example:
        >>> query_sap_data("monthly_rfqs", year=2025, month=7)
        >>> query_sap_data("recent_rfqs", days=30)
    """
    try:
        logger.info(f"Querying SAP data: {query_type} with params: {kwargs}")
        
        lambda_client = boto3.client('lambda', region_name=settings.AWS_REGION)
        
        # 处理Agent框架包装的kwargs参数
        if 'kwargs' in kwargs and isinstance(kwargs['kwargs'], str):
            import json
            actual_params = json.loads(kwargs['kwargs'])
        else:
            actual_params = kwargs
        
        # 构建查询参数
        payload = {
            "query_type": query_type,
            "parameters": actual_params
        }
        
        # 为需要limit参数的查询添加默认值
        limit_required_queries = [
            "lifecycle_status_query", "quotation_comparison", "material_detail_query",
            "payment_terms_analysis", "purchasing_org_query", "amount_range_query", 
            "incoterms_query", "manufacturer_query", "plant_query",
            "purchase_requisition_query", "exchange_rate_query"
        ]
        
        if query_type in limit_required_queries and "limit" not in actual_params:
            payload["parameters"]["limit"] = 50 
        
        response = lambda_client.invoke(
            FunctionName=settings.SAP_LAMBDA_NAME,
            Payload=json.dumps(payload)
        )
        
        result = json.loads(response['Payload'].read())
        
        logger.info(f"SAP query completed successfully: {query_type}")
        logger.info(f"Lambda result: {result}")
        
        # 直接返回Lambda的结果，让Agent处理
        return result
            
    except Exception as e:
        error_msg = f"查询SAP数据时发生错误: {str(e)}"
        logger.error(error_msg)
        return {
            "success": False,
            "error": error_msg
        }

import boto3
from strands import tool
from ..config import settings
from ..utils import get_logger

logger = get_logger(__name__)

@tool
def query_sustainability_knowledge(query: str) -> str:
    """
    查询供应商可持续性知识库
    Args:
        query: 查询问题，如"某供应商的ESG评级"、"可持续发展认证要求"等
    Returns:
        str: 知识库查询结果
    Example:
        >>> query_sustainability_knowledge("什么是ESG评级标准？")
    """
    try:
        logger.info(f"Querying sustainability knowledge: {query[:50]}...")
        bedrock_client = boto3.client('bedrock-agent-runtime', region_name=settings.AWS_REGION) 
        response = bedrock_client.retrieve_and_generate(
            input={'text': query},
            retrieveAndGenerateConfiguration={
                'type': 'KNOWLEDGE_BASE',
                'knowledgeBaseConfiguration': {
                    'knowledgeBaseId': settings.KNOWLEDGE_BASE_ID,
                    'modelArn': settings.KNOWLEDGE_BASE_MODEL_ARN
                }
            }
        )
        result = response['output']['text']
        logger.info("Sustainability knowledge query completed successfully")
        return result
        
    except Exception as e:
        error_msg = f"知识库查询失败: {str(e)}"
        logger.error(error_msg)
        return error_msg       

import boto3
import json
from typing import Dict, Any
from strands import tool
from ..config import settings
from ..utils import get_logger
logger = get_logger(__name__)
@tool
def create_purchase_order(
    quotation_id: str
) -> Dict[str, Any]:
    """
    创建SAP采购订单
    Args:
        quotation_id: 报价单编号 (如: 6000000262)
    Returns:
        Dict[str, Any]: 采购订单创建结果
    Example:
        >>> create_purchase_order("6000000262")
    """
    try:
        logger.info(f"Creating purchase order for quotation: {quotation_id}")
        
        lambda_client = boto3.client('lambda', region_name=settings.AWS_REGION)
        
        payload = {
            "apiPath": "/createPO",
            "httpMethod": "POST",
            "requestBody": {
                "QuotationID": quotation_id
            }
        }
        response = lambda_client.invoke(
            FunctionName="SAPPOC-action-sap-gai-west-new",
            Payload=json.dumps(payload)
        )
        result = json.loads(response['Payload'].read())
        logger.info(f"Purchase order creation completed for quotation: {quotation_id}")
        return result
    except Exception as e:
        error_msg = f"创建采购订单时发生错误: {str(e)}"
        logger.error(error_msg)
        return {
            "success": False,
            "error": error_msg
        }