基于LangGraph的电商AI Agent系统架构设计与实践

weixin_30566111

383人浏览 · 2026-07-25 15:21:25

weixin_30566111 · 2026-07-25 15:21:25 发布

1. 项目背景与挑战解析

在电商运营领域，购物场景生成系统正经历着从传统人工配置向AI智能生成的范式转变。我们团队负责的购物场景生成AI Agent系统，核心功能是让运营人员通过自然语言描述（如"生成一个冬日红豆年糕汤场景"），系统就能自动完成从场景理解、内容生成到商品匹配的全流程工作。

1.1 原有架构的痛点分析

旧版系统基于低代码流程编排平台构建，采用线性流程设计：

意图识别节点 → 2. 场景生成节点 → 3. 商品搜索节点 → 4. 结果组装节点

这种架构在初期快速验证阶段表现尚可，但随着业务复杂度提升，暴露出以下关键问题：

扩展性瓶颈 ：

新增功能需要修改整个流程链路
条件分支逻辑难以用图形化方式清晰表达
与内部系统（如商品搜索、知识库）的集成方式不统一

维护成本高 ：

错误处理逻辑分散在各节点
调试时需要跟踪整个流程状态
性能优化空间有限

智能化不足 ：

缺乏动态规划能力
多轮对话上下文管理困难
工具调用缺乏标准化接口

1.2 技术选型决策过程

经过对主流AI Agent框架的评估，我们最终选择LangGraph作为新架构的核心，主要基于以下考量：

框架能力矩阵对比 ：

特性	LangChain	AutoGPT	LangGraph
状态管理	弱	中等	强
可视化调试	无	基础	完善
分布式支持	有限	有限	完善
协议标准化	无	无	支持MCP
开发效率	中等	低	高

关键决策因素 ：

电商场景需要处理复杂的状态流转（如多轮对话中的上下文维护）
需要与企业内部多种协议（HSF、HTTP、MCP）无缝集成
未来需要支持动态扩展的业务场景

2. 新架构设计与核心创新

2.1 LangGraph架构总览

新系统采用分层架构设计：

应用层
├── 用户接口
├── 会话管理
└── 结果渲染

工作流层（LangGraph）
├── Planner节点
├── 技能执行节点
└── 状态检查点

技能层
├── 场景生成Skill
├── 商品服务Skill
└── 持久化Skill

服务层
├── LLM服务
├── 向量数据库
└── 商品搜索

持久层
├── 场景存储
└── 工作流状态存储

2.2 核心创新点实现

2.2.1 Agent Skills标准化封装

我们将系统能力拆分为独立的Skill模块，每个Skill包含：

SKILL.md ：接口文档和调用示例
handler.py ：核心业务逻辑
adapter.py ：协议转换层
test/ ：单元测试用例

场景生成Skill示例 ：

class SceneGenerationSkill:
    @skill_tool
    async def generate_scene_title(
        self, 
        user_input: str,
        context: dict
    ) -> dict:
        """
        生成场景标题
        参数:
            user_input: 用户自然语言输入
            context: 包含品类等上下文信息
        返回:
            {"title": str, "tags": List[str]}
        """
        prompt = self._build_prompt(user_input, context)
        result = await self.llm_service.generate(prompt)
        return self._parse_result(result)

2.2.2 Planner智能规划机制

Planner节点的执行流程：

接收用户输入和当前状态
调用LLM生成执行计划
验证计划完整性
动态加载所需Skills

Planner提示词设计要点 ：

PLANNER_PROMPT = """请生成JSON格式的执行计划，必须包含：
- 场景理解（使用scene-understanding Skill）
- 内容生成（使用content-generation Skill）
- 商品匹配（使用product-matching Skill）
- 结果持久化（使用persistence Skill）

示例输出：
{
  "steps": [
    {
      "name": "场景理解",
      "skill": "scene-understanding",
      "inputs": {"user_input": "..."},
      "outputs": ["category", "attributes"]
    },
    ...
  ]
}"""

2.2.3 状态管理优化方案

采用TypedDict定义严格的状态结构：

class SceneGuideState(TypedDict):
    user_input: str
    scene_data: NotRequired[dict]
    product_list: NotRequired[List[dict]]
    error: NotRequired[str]

状态检查点设计：

每完成一个关键步骤自动持久化
支持从任意检查点恢复执行
状态压缩：自动清理历史对话中的冗余信息

3. 工程化落地实践

3.1 AI辅助开发流程

我们采用"双工具+知识库"的开发模式：

工具矩阵 ：

场景	工具	使用技巧
架构设计	Cursor	上传DSL文件生成初始代码骨架
内部协议开发	AoneCopilot	关联内部文档自动补全API调用
代码审查	双工具并行	比较不同工具给出的优化建议

知识库建设 ：

/docs
  ├── architecture.md    # 架构设计文档
  ├── dsl-spec.yaml      # DSL规范
  └── protocols/         # 各协议接口文档

3.2 性能优化关键点

商品搜索优化 ：

建立标签-商品ID的倒排索引
实现两级缓存：
- 内存缓存：存储热点场景商品
- 分布式缓存：存储通用品类商品

LLM调用优化 ：

请求合并：将多个小请求合并为批量请求
结果缓存：对相同参数的生成结果缓存5分钟
流式传输：逐步返回生成结果

4. 效果评估与经验总结

4.1 量化指标对比

指标	旧架构	新架构	提升幅度
任务完成率	68%	88%	+20%
平均响应时间	2.4s	1.7s	-29%
代码复用率	15%	45%	+200%
异常恢复成功率	60%	95%	+58%

4.2 关键经验总结

架构设计经验 ：

状态字段要预留扩展空间，我们最初设计的State在迭代3次后就需重构
Skill接口要包含版本号，便于后续兼容性处理
检查点数据需要定期清理，避免存储膨胀

AI辅助开发心得 ：

给AI工具提供足够的上下文（如上传架构图）
对生成代码必须进行人工复核，特别是异常处理逻辑
建立项目专属的提示词库，持续优化交互效率

避坑指南 ：

不要过度依赖Planner的自动规划，关键路径需要硬编码保障
Skill之间的数据依赖要显式声明，避免隐式耦合
分布式环境下要处理好状态锁的问题

5. 典型问题排查手册

5.1 商品匹配异常

症状：

生成的场景与商品不相关
商品数量不足

排查步骤 ：

检查Planner输出的商品搜索参数
验证商品搜索Skill的输入输出
检查商品索引的更新时间戳

解决方案 ：

async def debug_product_matching(state: SceneGuideState):
    logger.info(f"当前状态: {state}")
    if "scene_data" not in state:
        return {"error": "缺少scene_data"}
    
    # 手动触发商品搜索
    test_result = await product_skill.search(
        tags=state["scene_data"]["tags"],
        limit=10
    )
    return {"debug_result": test_result}

5.2 状态恢复失败

症状：

多轮对话中丢失上下文
从检查点恢复后流程错乱

排查步骤 ：

检查Checkpoint存储的实现
验证State的序列化/反序列化
检查节点间的状态传递

解决方案 ：

class SafeCheckpointer(Checkpointer):
    async def save(self, state: dict):
        # 压缩历史状态
        compressed = {
            k: v for k, v in state.items()
            if not k.startswith('_')
        }
        await self.backend.save(compressed)