这个系列写了二十三篇文章。
从架构设计写到稳定性工程，从分布式集群写到编排引擎，从策略中台写到安全体系，从商品生命周期写到营销体系，从风控适配写到内容素材工程，从组织博弈写到新手指南。
几乎所有技术维度都覆盖了。
但有一个方向，我在之前的文章里提过好几次，却从来没有系统性地展开。
AI Agent。
不是“大模型API调用”，不是“AI生成标题”，不是“智能定价”——这些只是AI能力在特定场景的应用。
AI Agent是不同的。
它是一个自主决策的智能体——能够理解目标、拆解任务、调用工具、执行操作、反思结果、迭代优化。
传统RPA是“执行者”——你告诉它每一步怎么做，它照做。
AI Agent是“决策者”——你告诉它“我要什么结果”，它自己决定“怎么达成”。
这个区别，是质变。
今天这篇文章，我们聊聊店群自动化系统中的AI Agent工程实践。
不是“趋势展望”，不是“概念介绍”。
是已经跑在生产环境里的真实方案。
一、先说说“规则引擎的天花板”在哪
在聊AI Agent之前，先搞清楚：现有的规则引擎为什么不够？
前面文章里写的策略中台、决策引擎、定价规则——本质上都是“if-else”的集合。
运营配置规则：“如果库存低于10件，就涨价5%。”
系统执行规则：“库存低于10件 → 涨价5%。”
这套逻辑在确定性场景里很好用。
但店群运营的很多场景，恰恰是不确定的。
竞品突然降价了——你只知道“降价了”，但不知道“为什么降价”、“会降多久”、“该不该跟”。
平台突然出了新活动——你只知道“有活动”，但不知道“报不报”、“报什么价”、“报多少库存”。
买家突然大量退货——你只知道“退货多了”，但不知道“是质量问题还是恶意退款”、“该怎么做才能止损”。
规则引擎能处理“已知的已知”——你提前想好了所有可能的情况，写好了对应的规则。
但真实世界充满了“未知的未知”——你根本想不到的情况，规则引擎处理不了。
AI Agent填补的，正是这个空白。
二、从“规则引擎”到“AI Agent”的架构演进
传统自动化系统的架构是：
用户输入 → 规则引擎（if-else） → 执行引擎（RPA） → 输出结果。
AI Agent架构是：
用户输入（自然语言目标） → Agent（理解+规划+决策） → 工具调用（API/RPA/数据库） → 结果评估 → Agent（反思+优化） → 输出结果。
多了三个关键环节：理解、规划、反思。
理解层：把用户的自然语言目标翻译成机器可执行的任务。
“帮我优化一下店铺A的转化率” → “分析店铺A的流量数据、商品数据、竞品数据 → 找出转化率低的原因 → 生成优化方案 → 执行优化操作 → 评估效果。”
规划层：把大任务拆解成子任务，确定执行顺序和依赖关系。
“分析店铺A的流量数据” → 调取GA数据 → 分析流量来源 → 识别异常流量 → 生成分析报告。
反思层：执行完成后，评估结果是否达到预期，如果没有，调整策略再试。
这三个环节，传统RPA完全没有。
只有大模型能胜任。
这就是为什么说“AI Agent重构了自动化的决策层”——它让系统从“执行指令”变成了“理解目标”。
三、AI Agent在店群自动化中的三个落地场景
说了这么多概念，说点实际的。
我们在店群自动化系统里已经跑通了三个AI Agent场景。
场景一：商品标题与描述的智能生成与优化Agent。
这是最早跑通的场景。
传统做法：运营写标题，或者用模板生成。
AI Agent做法：输入商品基本信息（品类、规格、卖点），Agent自动分析该品类的搜索关键词趋势、竞品标题特征、平台SEO规则，然后生成10个标题候选，按预估点击率排序，供运营选择。
更进一步：标题上线后，Agent持续监控该商品的搜索曝光和点击数据，自动检测哪些关键词表现好、哪些表现差，然后迭代生成新的标题版本。
这不是“一次生成”，而是“持续优化”。
场景二：竞品分析与定价策略Agent。
传统做法：系统定时采集竞品价格，规则引擎判断是否调价。
AI Agent做法：Agent不仅要“看到”竞品降价了，还要“理解”为什么降价——是清库存、是冲销量、还是应对平台活动？
然后基于理解做出决策：如果竞品是清库存（库存低于阈值且降价幅度大），Agent判断这是短期行为，不建议跟价。如果竞品是冲销量（销量上升且降价幅度适中），Agent判断这是长期策略，建议调整定价。
更进一步：Agent会分析历史数据，学习“什么样的定价策略在什么样的市场环境下最有效”，然后自动优化定价规则。
这不是“规则执行”，而是“策略学习”。
场景三：异常诊断与自愈Agent。
传统做法：系统检测到异常（如任务失败率突然上升），触发告警，人工介入排查。
AI Agent做法：Agent检测到异常后，自动开始诊断——采集异常任务的日志、截图、系统状态 → 分析可能的根因（平台改版？网络波动？资源耗尽？） → 生成修复方案 → 尝试自动修复 → 验证修复效果。
如果修复成功，Agent记录这次异常和处理方式，作为“案例”存入知识库。下次遇到类似异常，Agent可以直接调用历史方案。
如果修复失败，Agent生成详细的诊断报告，推送给技术人员，大大缩短了人工排查的时间。
这不是“告警后等人修”，而是“系统自己尝试修”。
四、工程实现：一个轻量级AI Agent框架
下面是我们生产环境中使用的AI Agent核心框架。
# agent_core.py - AI Agent核心框架
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import json
import time
import uuid
from datetime import datetime
class AgentRole(Enum):
"""Agent角色"""
PLANNER = "planner" # 规划者 - 拆解任务
EXECUTOR = "executor" # 执行者 - 调用工具
REFLECTOR = "reflector" # 反思者 - 评估结果
ORCHESTRATOR = "orchestrator" # 编排者 - 协调多个Agent
@dataclass
class AgentContext:
"""Agent上下文 - 贯穿整个执行过程"""
session_id: str
user_goal: str # 用户原始目标
current_task: str # 当前正在执行的任务
completed_tasks: List[str] = field(default_factory=list)
failed_tasks: List[str] = field(default_factory=list)
tool_results: Dict[str, Any] = field(default_factory=dict)
reflections: List[str] = field(default_factory=list)
iteration: int = 0
max_iterations: int = 10
@dataclass
class Tool:
"""Agent可调用的工具"""
name: str
description: str
parameters: Dict[str, Any] # JSON Schema
handler: Callable
class AgentOrchestrator:
"""
AI Agent编排器
核心工作流：理解 → 规划 → 执行 → 反思 → 迭代
"""
def __init__(self, llm_client, tool_registry):
self.llm = llm_client
self.tools = tool_registry
self._max_iterations = 10
def run(self, user_goal: str, context: Optional[Dict] = None) -> Dict[str, Any]:
"""
执行Agent工作流
输入：用户目标（自然语言）
输出：执行结果
"""
# 1. 初始化上下文
ctx = AgentContext(
session_id=f"agent_{uuid.uuid4().hex[:8]}",
user_goal=user_goal
)
# 2. 主循环 - 计划-执行-反思
while ctx.iteration < ctx.max_iterations:
ctx.iteration += 1
# 2.1 规划阶段 - 生成执行计划
plan = self._plan(ctx)
if not plan:
break
# 2.2 执行阶段 - 调用工具
for step in plan:
result = self._execute_step(ctx, step)
if not result.get("success", False):
ctx.failed_tasks.append(step.get("task", "unknown"))
else:
ctx.completed_tasks.append(step.get("task", "unknown"))
ctx.tool_results[step.get("task")] = result.get("data")
# 2.3 反思阶段 - 评估是否达到目标
if self._should_continue(ctx):
continue
else:
break
# 3. 返回最终结果
return {
"session_id": ctx.session_id,
"iterations": ctx.iteration,
"completed_tasks": ctx.completed_tasks,
"failed_tasks": ctx.failed_tasks,
"results": ctx.tool_results,
"reflections": ctx.reflections
}
def _plan(self, ctx: AgentContext) -> List[Dict]:
"""
规划阶段 - 调用LLM生成执行计划
输入：用户目标、当前上下文、可用工具列表
输出：有序的任务步骤列表
"""
prompt = self._build_planning_prompt(ctx)
response = self.llm.chat(prompt)
plan = self._parse_plan(response)
return plan
def _execute_step(self, ctx: AgentContext, step: Dict) -> Dict:
"""
执行阶段 - 调用工具执行单个任务
输入：任务描述 + 参数
输出：执行结果
"""
tool_name = step.get("tool")
params = step.get("params", {})
tool = self.tools.get(tool_name)
if not tool:
return {"success": False, "error": f"工具不存在: {tool_name}"}
try:
result = tool.handler(**params)
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}
def _should_continue(self, ctx: AgentContext) -> bool:
"""反思阶段 - 判断是否需要继续迭代"""
# 调用LLM评估是否达到目标
prompt = self._build_reflection_prompt(ctx)
response = self.llm.chat(prompt)
decision = self._parse_reflection(response)
ctx.reflections.append(decision.get("reflection", ""))
return decision.get("continue", False)
def _build_planning_prompt(self, ctx: AgentContext) -> str:
"""构建规划阶段的Prompt"""
tools_desc = "\n".join([
f"- {name}: {tool.description}"
for name, tool in self.tools.items()
])
return f"""
你是一个店群运营自动化Agent。用户的目标是：{ctx.user_goal}
当前已完成的任务：{ctx.completed_tasks}
当前失败的任务：{ctx.failed_tasks}
你可以使用以下工具：
{tools_desc}
请生成一个执行计划，以JSON格式输出，包含一个任务步骤列表。
每个步骤包含：task（任务描述）、tool（使用的工具）、params（参数）。
计划格式：
{{
"plan": [
{{"task": "分析竞品价格", "tool": "get_competitor_prices", "params": {{"product_id": "123"}}}},
{{"task": "计算建议价格", "tool": "calculate_price", "params": {{"strategy": "competitive"}}}}
]
}}
"""
def _build_reflection_prompt(self, ctx: AgentContext) -> str:
"""构建反思阶段的Prompt"""
return f"""
你是一个店群运营自动化Agent。用户的目标是：{ctx.user_goal}
已完成的任务：{ctx.completed_tasks}
失败的任务：{ctx.failed_tasks}
工具执行结果：{json.dumps(ctx.tool_results, ensure_ascii=False)[:500]}
请评估是否已经达到用户的目标。
如果已达到，输出 {{"continue": false, "reflection": "已达成目标"}}
如果未达到，输出 {{"continue": true, "reflection": "还需要继续...", "next_steps": "..."}}
"""
五、Agent与RPA的集成：让Agent拥有“手脚”
上面说的Agent框架，核心能力是“思考”和“决策”。
但思考完了，谁来执行？
答案是：影刀RPA。
在我们的架构中，Agent通过两个方式调用RPA：
方式一：API调用。
影刀RPA支持HTTP API触发。Agent生成执行计划后，通过API调用对应的RPA流程。
比如Agent决定“上架商品A”，它会调用/api/run_flow，传入flow_name=upload_product和params={product_id: "123"}。
影刀RPA收到请求后执行对应的流程，执行完成后回调Agent汇报结果。
方式二：流程触发。
对于一些需要实时交互的复杂流程，Agent会生成“流程触发指令”，写入消息队列。
影刀RPA监听队列，收到指令后执行对应的流程。
这个模式和之前的调度系统类似，但触发逻辑从“定时调度”变成了“Agent决策驱动”。
举个例子：
运营说：“帮我处理一下今天所有的异常订单。”
Agent理解目标 → 调用get_abnormal_orders工具获取异常订单列表 → 分析每个订单的异常类型 → 对于“物流异常”的订单，调用sync_logistics RPA流程 → 对于“支付异常”的订单，调用check_payment RPA流程 → 对于无法自动处理的异常，生成报告推送给运营。
整个链路，从理解到执行到反馈，全部自动化。
六、工具注册表：Agent能调用的一切
为了让Agent能够调用系统能力，我们维护了一个“工具注册表”。
所有Agent可调用的能力——API、RPA流程、数据库查询、外部服务——都注册在工具表中。
# tool_registry.py - Agent工具注册表
from typing import Dict, Any, Callable
from dataclasses import dataclass
@dataclass
class ToolDefinition:
"""工具定义"""
name: str
description: str
parameters: Dict[str, Any] # JSON Schema
handler: Callable
timeout: int = 60
retry: int = 3
class ToolRegistry:
"""工具注册表 - Agent可调用的所有能力"""
def __init__(self):
self._tools: Dict[str, ToolDefinition] = {}
def register(self, name: str, description: str,
parameters: Dict[str, Any], handler: Callable):
"""注册工具"""
self._tools[name] = ToolDefinition(
name=name,
description=description,
parameters=parameters,
handler=handler
)
def get(self, name: str) -> Optional[ToolDefinition]:
"""获取工具"""
return self._tools.get(name)
def list_all(self) -> List[Dict]:
"""列出所有工具（供LLM理解）"""
return [
{
"name": t.name,
"description": t.description,
"parameters": t.parameters
}
for t in self._tools.values()
]
def invoke(self, name: str, params: Dict) -> Dict:
"""调用工具"""
tool = self._tools.get(name)
if not tool:
return {"success": False, "error": f"工具不存在: {name}"}
try:
result = tool.handler(**params)
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}
# 初始化工具注册表
registry = ToolRegistry()
# 注册系统工具
registry.register(
name="get_competitor_prices",
description="获取竞品价格数据，输入商品ID，返回竞品价格列表",
parameters={"product_id": {"type": "string"}},
handler=lambda product_id: {"price": 99.0, "competitors": [{"name": "竞品A", "price": 89.0}]}
)
registry.register(
name="calculate_price",
description="根据策略计算建议价格，输入商品信息和策略类型，返回建议价格",
parameters={
"product_id": {"type": "string"},
"strategy": {"type": "string", "enum": ["competitive", "premium", "discount"]}
},
handler=lambda product_id, strategy: {"suggested_price": 95.0}
)
registry.register(
name="update_price",
description="更新商品价格，输入商品ID和目标价格，返回执行结果",
parameters={
"product_id": {"type": "string"},
"target_price": {"type": "number"}
},
handler=lambda product_id, target_price: {"success": True}
)
有了这个工具注册表，Agent的“知识”就不仅是语言模型中的通用知识，还包括了系统特有的业务能力。
七、持续学习：让Agent越用越聪明
传统的LLM应用有一个问题：每次调用都是“从头开始”。
它不记得上次的决策、不学习历史的成功经验、不积累业务知识。
我们给Agent加了一层“记忆系统”：
短期记忆： 当前会话的上下文，包括之前的决策、执行结果、反思记录。存储在Redis中，会话结束后清理。
长期记忆： 跨会话积累的经验——哪些策略有效、哪些工具好用、哪些模式值得复用。存储在向量数据库中，Agent在规划时会检索相关经验。
举个例子：
第一次遇到“竞品降价”场景，Agent需要从头开始分析。
第十次遇到“竞品降价”场景，Agent会从长期记忆中检索到“上一次成功的处理策略是：先判断降价原因，再决定是否跟价”，然后直接应用这个策略。
Agent在持续“学习”。
这就是“经验驱动”和“规则驱动”的根本区别。
规则是死的，经验是活的。
八、写在最后
AI Agent不是“RPA的替代品”，也不是“大模型的简单包装”。
它是一个新的架构范式。
在传统RPA架构中，人是“决策者”，机器是“执行者”。
在AI Agent架构中，机器也开始参与“决策”——理解目标、生成计划、调用工具、反思结果。
人从“决策者”变成了“监督者”——审核Agent的计划、修正Agent的偏差、确认Agent的结果。
这不是“机器替代人”。
这是“人机协同”的新形态。
这个方向的工程化落地，还有很多挑战需要解决——Agent的稳定性、可解释性、成本控制、异常处理。
但方向已经很明确了。
希望这篇文章能给你一些启发。
哪怕只是让你意识到“原来还可以这样做”，这篇文章的价值就达到了。
作者：林焱