电商智能运营洞察自动化生成方案

from typing import List, TypedDict, Annotated
from langgraph.graph import StateGraph
from langgraph.checkpoint.aiopostgres import PostgresSaver
import operator
 
# 1. 定义中心状态对象
# 这个 TypedDict 定义了在整个工作流中传递的数据结构。
class AgentState(TypedDict):
    # 初始用户请求
    initial_request: str
    # 数据分析结果
    analysis_result: dict
    # 知识检索的上下文
    retrieved_knowledge: List[str]
    # 最终报告内容
    final_report: str
    # 用于记录中间步骤的消息列表
    intermediate_steps: Annotated[List[str], operator.add]
 
# 2. 初始化数据库支持的 Checkpointer
# 这使得工作流的每一步状态都能被持久化到 PostgreSQL 数据库中
db_conn_info = "postgresql+psycopg://user:password@host:port/dbname"
memory = PostgresSaver.from_conn_string(db_conn_info)
 
# 3. 创建 StateGraph 实例
# 将状态定义和 Checkpointer 传入，构建一个有状态、可持久化的工作流图
workflow = StateGraph(AgentState)
 
#... 在这里添加节点 (nodes) 和边 (edges)...
 
# 4. 编译图
# 编译后的 app 对象就是一个可执行的、具备持久化能力的应用
app = workflow.compile(checkpointer=memory)

from typing import Literal
 
# 假设 data_analysis_node 节点执行后，会更新 state['analysis_result']
# analysis_result 可能包含一个 'status' 字段
 
def route_after_analysis(state: AgentState) -> Literal["generate_report", "clarify_with_user", "__end__"]:
    """
    这是一个路由函数，用于实现条件边。
    它检查数据分析节点的结果，并决定下一步的走向。
    """
    print("---DECISION: ROUTING AFTER DATA ANALYSIS---")
    analysis_status = state.get("analysis_result", {}).get("status")
 
    if analysis_status == "success":
        # 数据完整且分析成功，进入报告生成环节
        print("Decision: Analysis successful. Proceeding to report generation.")
        return "generate_report"
    elif analysis_status == "insufficient_data":
        # 数据不足，需要与用户交互以获取更多信息
        print("Decision: Insufficient data. Routing to user clarification.")
        return "clarify_with_user"
    else:
        # 出现无法处理的错误，结束工作流
        print("Decision: Analysis failed with an unrecoverable error. Ending workflow.")
        return "__end__"
 
# 在构建图时，将这个路由函数与一个条件边关联起来
# workflow.add_node("data_analyzer", data_analysis_node)
# workflow.add_conditional_edges(
#     "data_analyzer",
#     route_after_analysis,
#     {
#         "generate_report": "report_generator",
#         "clarify_with_user": "user_interaction_node",
#         "__end__": END
#     }
# )

启动开发环境
推荐使用 Amazon Linux进行开发和测试，因为它与 Lambda 运行时环境一致。

安装 uv（快速 Python 包管理器）
curl -LsSf https://astral.sh/uv/install.sh | sh
export PATH="$HOME/.cargo/bin:$PATH"
uv --version

创建项目
mkdir langgraph-agentcore-postgres
cd langgraph-agentcore-postgres
uv venv
source .venv/bin/activate

安装依赖
uv pip install bedrock-agentcore bedrock-agentcore-starter-toolkit langgraph langchain-aws psycopg2-binary sqlalchemy
agentcore --help
python -c "import langgraph; print('LangGraph version:', langgraph.version)"

安装系统依赖（Amazon Linux）
sudo dnf install -y postgresql15
sudo dnf install -y git vim

创建 RDS 实例并记录 RDS 信息
aws rds create-db-instance
--db-instance-identifier langgraph-memory-db
--db-instance-class db.t3.micro
--engine postgres
--engine-version 15.4
--master-username postgres
--master-user-password YourSecurePassword123
--allocated-storage 20
--vpc-security-group-ids sg-xxxxx
--db-subnet-group-name your-subnet-group
--publicly-accessible false

连接到 RDS：
psql -h langgraph-memory-db.xxxxx.us-east-1.rds.amazonaws.com -U postgres -d postgres
创建数据库：
CREATE DATABASE langgraph_memory;

# ============================================================================
# 定义 State
# ============================================================================

class AgentState(TypedDict):
    """Agent 状态定义"""
    messages: Annotated[list, operator.add]  # 消息历史
    user_input: str  # 用户输入
    response: str  # AI 响应

# ============================================================================
# 定义 Agent 节点
# ============================================================================

def process_message(state: AgentState) -> AgentState:
    """
    处理用户消息的节点
    这里可以添加你的 AI 逻辑，比如调用 LLM
    """
    user_input = state["user_input"]
    
    # 简单的响应逻辑（替换为你的 AI 模型调用）
    response = f"收到您的消息: {user_input}"
    
    # 更新状态
    return {
        "messages": [
            HumanMessage(content=user_input),
            AIMessage(content=response)
        ],
        "response": response
    }

# ============================================================================
# 创建 LangGraph
# ============================================================================

def create_agent_graph():
    """创建 Agent Graph"""
    
    # 创建 Graph
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("process", process_message)
    
    # 定义流程
    workflow.add_edge(START, "process")
    workflow.add_edge("process", END)
    
    # 编译 Graph（使用内存 Checkpointer）
    checkpointer = MemorySaver()
    graph = workflow.compile(checkpointer=checkpointer)
    
    return graph

# ============================================================================
# BedrockAgentCore 应用
# ============================================================================

app = BedrockAgentCoreApp()

# 创建 Graph（全局单例）
agent_graph = create_agent_graph()

@app.entrypoint
def invoke(payload):
    """
    AI Agent 入口函数
    
    Args:
        payload: 包含 prompt 的字典
        
    Returns:
        包含 result 的字典
    """
    # 获取用户消息
    user_message = payload.get("prompt", "Hello! How can I help you today?")
    
    # 获取或生成 thread_id（用于会话管理）
    thread_id = payload.get("thread_id", "default_thread")
    
    # 调用 Graph
    result = agent_graph.invoke(
        {
            "messages": [],
            "user_input": user_message,
            "response": ""
        },
        config={
            "configurable": {
                "thread_id": thread_id
            }
        }
    )
    
    # 返回响应
    return {
        "result": result["response"]
    }

# ============================================================================
# 运行
# ============================================================================

if __name__ == "__main__":
    app.run()

bedrock-agentcore
langgraph>=0.2.0
langchain-aws>=0.1.0
psycopg2-binary>=2.9.9
sqlalchemy>=2.0.0

设置环境变量：
export DATABASE_URL="postgresql://postgres:password@host.docker.internal:5432/langgraph_memory"
export AWS_REGION="us-east-1"

启动 agent：
python my_agent.py

在另一个终端测试：
curl -X POST http://localhost:8080/invocations -H "Content-Type: application/json" -d '{"prompt": "你好！"}'

验证输出
成功的输出应该类似：
{
"result": "你好！我是一个友好的 AI 助手...",
"thread_id": "default"
}

基础配置
agentcore configure -e my_agent.py
在配置过程中：
Region: 选择你的 RDS 所在区域（如 us-east-1）
Memory: 选择 "No" （我们使用自己的 PostgreSQL）
Deployment mode: 选择 direct_code_deploy（推荐）或 container
配置 VPC
重要：根据 AWS 官方文档，AgentCore Runtime 必须使用私有子网。
编辑生成的 .bedrock_agentcore.yaml 文件，添加 VPC 配置：
bedrock_agentcore:
agent_name: my_agent
region: us-east-1
deployment_mode: direct_code_deploy
VPC 配置（使用私有子网）
vpc_config:
subnet_ids:
- subnet-private-1a # 必须是私有子网（无直接 Internet 访问）
- subnet-private-1b # 建议多 AZ 部署
security_group_ids:
- sg-lambda-langgraph # Lambda 安全组
环境变量
environment_variables:
DATABASE_URL: "postgresql://postgres:YourSecurePassword123@langgraph-memory-db.xxxxx.us-east-1.rds.amazonaws.com:5432/langgraph_memory"
AWS_REGION: "us-east-1"
MODEL_ID: "us.amazon.nova-pro-v1:0"
Lambda 配置
timeout: 60
memory_size: 512

agentcore launch
打包你的代码和依赖
上传到 S3（direct_code_deploy）或 ECR（container）
创建 Lambda 函数
配置 VPC 和安全组
设置环境变量
配置 IAM 角色和权限

记录输出中的：
Agent ARN: arn:aws:bedrock-agentcore:us-east-1:xxxxx:agent-runtime/my-agent/DEFAULT
CloudWatch Logs: /aws/bedrock-agentcore/runtimes/my-agent-DEFAULT

检查状态
agentcore status

测试调用
测试基础对话：
agentcore invoke '{"prompt": "你好！"}'
测试多轮对话（使用 thread_id）：
agentcore invoke '{"prompt": "我叫张三", "thread_id": "user_001"}'
agentcore invoke '{"prompt": "你还记得我的名字吗？", "thread_id": "user_001"}'
测试会话隔离：
agentcore invoke '{"prompt": "你知道张三是谁吗？", "thread_id": "user_002"}'