Agent产品的下一步技术路线:从LLM到LAM(Large Action Model)的演进

一、LLM的边界:为什么纯粹的语言模型无法满足Agent的核心需求

当前主流的Agent产品几乎都建立在LLM之上。典型的模式是:用LLM解析用户意图、生成执行计划、拆解为工具调用、汇总输出结果。这个模式在简单任务上表现尚可,但在复杂操作场景下暴露出三个根本缺陷。

第一个缺陷是推理与行动的脱节。LLM生成一个步骤描述(比如"点击登录按钮"),但这个描述需要通过中间层的翻译才能转化为实际的UI操作。这个翻译过程是错误的高发区——文本描述"点击页面右上角的按钮"在实践中经常因为页面布局变化而失败。

第二个缺陷是执行过程缺乏记忆。LLM执行多步骤任务时,每步都是"无状态"的推理。如果第3步操作后页面跳转到一个新URL,LLM无法持续感知到这个环境变化,它只能被动地等待外部系统告诉它当前的状态。

第三个缺陷是错误恢复能力弱。LLM擅长生成"理想路径"的操作计划,但对"异常路径"的处理几乎空白。当某个中间步骤失败时,LLM的默认行为是重新生成整个过程,而不是基于当前状态从失败点恢复。

这三个缺陷的背后是一个共识:下一代Agent需要的不只是一个"语言模型",而是一个能理解操作上下文、执行具体行动、并能从执行结果中学习的"行动模型"(Large Action Model,LAM)。

二、LAM架构设计:感知、规划、执行与反思的四层闭环

LAM的核心思想是将Agent的智能拆解为四个独立但协同的层次:感知层负责理解当前的环境状态(页面DOM、API响应、系统状态),规划层负责制定行动序列,执行层负责将行动转化为实际操作,反思层负责评估操作结果并调整后续策略。

四层架构的关键设计原则:

感知层需要同时支持多种输入模态。传统的LLM-Agent只能处理文本输入,LAM必须能理解UI截图(通过Vision模型)、DOM结构变化(通过HTML解析器)、API响应状态码和系统日志。多模态感知的能力决定了Agent能处理多复杂的操作环境。

规划层与传统LLM推理的根本差异在于"行动序列的物理可行性"。LLM可以生成"删除数据库中所有过期数据"这样的描述,但LAM必须知道"过期数据的WHERE条件是什么、删错了怎么回滚、正在写入的事务怎么处理"。规划层需要引入操作的安全约束和回滚机制。

执行层的核心技术挑战是操作的可信度。对UI操作而言,元素定位的鲁棒性远比操作速度重要。基于CSS选择器或XPath的定位方式脆弱性极高,需要结合视觉定位(screenshot + 目标检测模型)和文本定位(accessibility label、placeholder text)的多策略融合。

反思层是LAM区别于LLM-Agent的最核心差异。反思层允许Agent将失败经验编码进一个持久化的记忆库,后续相似情况下可以直接检索并避免重蹈覆辙。这个记忆库不是简单的文本向量存储,而是结构化的"状态-操作-结果"三元组知识图谱。

三、LAM核心引擎实现:多模态感知与经验驱动的行动决策

以下是LAM核心引擎的简化实现,展示了感知、规划、执行和反思四层之间的数据流和决策逻辑。

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any, Tuple
from enum import Enum
import json
import hashlib

class ActionStatus(Enum):
    SUCCESS = "成功"
    FAILED = "失败"
    PARTIAL = "部分成功"

@dataclass
class EnvironmentState:
    """环境状态表征"""
    current_url: str = ""
    visible_elements: List[Dict] = field(default_factory=list)
    dom_hash: str = ""
    api_responses: List[Dict] = field(default_factory=list)
    
    def fingerprint(self) -> str:
        """生成环境状态指纹,用于经验检索"""
        raw = f"{self.current_url}_{self.dom_hash}"
        return hashlib.md5(raw.encode()).hexdigest()[:12]

@dataclass 
class ActionStep:
    """操作步骤定义"""
    step_id: str
    action_type: str  # click, type, scroll, api_call, wait
    target_selector: str
    target_visual_hint: Optional[str] = None  # 视觉定位辅助
    params: Dict = field(default_factory=dict)
    rollback_steps: List['ActionStep'] = field(default_factory=list)
    retry_count: int = 3
    timeout_ms: int = 5000

@dataclass
class ExperienceEntry:
    """经验记录:状态-操作-结果三元组"""
    state_fingerprint: str
    action_type: str
    result: ActionStatus
    success_pattern: Optional[Dict] = None
    failure_reason: Optional[str] = None
    timestamp: float = 0.0

class LAMEngine:
    """LAM四层引擎"""
    
    def __init__(self):
        self.experience_memory: Dict[str, List[ExperienceEntry]] = {}
        self.current_state: Optional[EnvironmentState] = None
    
    def perceive(self, raw_input: Dict) -> EnvironmentState:
        """感知层:从原始输入构建环境状态表征"""
        state = EnvironmentState()
        
        # 解析多模态输入
        if "url" in raw_input:
            state.current_url = raw_input["url"]
        if "dom" in raw_input:
            state.visible_elements = self._parse_visible_elements(
                raw_input["dom"]
            )
            state.dom_hash = hashlib.md5(
                json.dumps(state.visible_elements, sort_keys=True).encode()
            ).hexdigest()
        if "api_responses" in raw_input:
            state.api_responses = raw_input["api_responses"]
        
        if not state.current_url:
            raise ValueError("环境状态缺少URL信息")
        
        self.current_state = state
        return state
    
    def plan(self, goal: Dict) -> List[ActionStep]:
        """规划层:生成行动序列,结合经验记忆"""
        if not self.current_state:
            raise ValueError("必须先调用perceive获取环境状态")
        
        # 查询经验记忆,避免在相似状态下执行已知会失败的操作
        fp = self.current_state.fingerprint()
        past_failures = self._query_failures(fp)
        steps = self._generate_plan(goal)
        
        # 过滤掉已知的失败操作,替换为替代方案
        filtered_steps = []
        for step in steps:
            if step.action_type not in past_failures:
                filtered_steps.append(step)
            else:
                alt = self._find_alternative(step, past_failures)
                if alt:
                    filtered_steps.append(alt)
        return filtered_steps if filtered_steps else steps[:1]
    
    def execute(self, step: ActionStep, 
                executor: Any) -> Tuple[ActionStatus, Optional[str]]:
        """执行层:执行单步操作,含重试和回滚"""
        for attempt in range(step.retry_count):
            try:
                result = executor.execute(step)
                if result.success:
                    return ActionStatus.SUCCESS, None
                
                # 重试时调整策略(如切换选择器类型)
                if attempt < step.retry_count - 1:
                    step = self._adjust_retry_strategy(step, attempt)
            except Exception as e:
                if attempt == step.retry_count - 1:
                    return ActionStatus.FAILED, str(e)
        return ActionStatus.FAILED, "重试次数用尽"
    
    def reflect(self, step: ActionStep, status: ActionStatus,
                error_info: Optional[str]) -> None:
        """反思层:将执行结果编码到经验记忆"""
        if not self.current_state:
            return
        
        fp = self.current_state.fingerprint()
        entry = ExperienceEntry(
            state_fingerprint=fp,
            action_type=step.action_type,
            result=status,
            failure_reason=error_info,
            success_pattern=step.params if status == ActionStatus.SUCCESS else None
        )
        
        if fp not in self.experience_memory:
            self.experience_memory[fp] = []
        self.experience_memory[fp].append(entry)
    
    def _parse_visible_elements(self, dom: Dict) -> List[Dict]:
        """DOM解析:提取可见且可交互的元素"""
        elements = []
        for node in self._traverse_dom(dom):
            if node.get("visible", True) and node.get("interactive", False):
                elements.append({
                    "tag": node.get("tag", ""),
                    "text": node.get("text", ""),
                    "selector": node.get("css_selector", ""),
                    "bbox": node.get("bounding_box", {}),
                })
        return elements
    
    def _traverse_dom(self, node: Dict) -> List[Dict]:
        """递归遍历DOM树"""
        result = [node]
        for child in node.get("children", []):
            result.extend(self._traverse_dom(child))
        return result
    
    def _query_failures(self, fingerprint: str) -> List[str]:
        """查询指定状态下的历史失败操作类型"""
        entries = self.experience_memory.get(fingerprint, [])
        return list(set(
            e.action_type for e in entries 
            if e.result == ActionStatus.FAILED
        ))
    
    def _generate_plan(self, goal: Dict) -> List[ActionStep]:
        """基于目标生成初始行动计划"""
        # 生产环境此处接入微调后的行动规划模型
        # 这里展示简化的规则生成逻辑
        return [ActionStep(
            step_id=f"step_{i}",
            action_type=goal.get("type", "click"),
            target_selector=goal.get("selector", ""),
            target_visual_hint=goal.get("visual_hint"),
            params=goal.get("params", {}),
        ) for i, goal in enumerate(goal.get("steps", [goal]))]
    
    def _find_alternative(
        self, step: ActionStep, failures: List[str]
    ) -> Optional[ActionStep]:
        """为已知失败的操作寻找替代策略"""
        alternatives = {
            "click": "keyboard_navigate",
            "type": "paste_clipboard",
        }
        alt_type = alternatives.get(step.action_type)
        if alt_type and alt_type not in failures:
            return ActionStep(
                step_id=f"{step.step_id}_alt",
                action_type=alt_type,
                target_selector=step.target_selector,
                params=step.params,
            )
        return None
    
    def _adjust_retry_strategy(
        self, step: ActionStep, attempt: int
    ) -> ActionStep:
        """重试时调整策略"""
        if attempt == 0:
            step.timeout_ms *= 2
        elif attempt == 1:
            step.target_selector = step.target_selector.replace(
                "css", "xpath"
            )
        return step

# 使用示例
if __name__ == "__main__":
    engine = LAMEngine()
    
    # 感知阶段
    raw_input = {
        "url": "https://example.com/dashboard",
        "dom": {
            "tag": "body",
            "visible": True,
            "interactive": False,
            "children": [
                {
                    "tag": "button",
                    "text": "创建新任务",
                    "visible": True,
                    "interactive": True,
                    "css_selector": "#create-task-btn",
                }
            ]
        }
    }
    engine.perceive(raw_input)
    
    # 规划阶段
    goal = {
        "steps": [
            {"type": "click", "selector": "#create-task-btn"},
            {"type": "type", "selector": "#task-name", 
             "params": {"text": "日报生成"}},
        ]
    }
    plan = engine.plan(goal)
    for step in plan:
        print(f"计划步骤: {step.action_type} -> {step.target_selector}")
    
    # 模拟执行并反思
    engine.reflect(ActionStep("s1", "click", "#create-task-btn"),
                   ActionStatus.SUCCESS, None)

核心实现要点:经验记忆的查询不是简单的全字匹配,而是基于环境状态指纹的模糊匹配。当指纹完全相同时(说明环境状态一模一样),直接使用历史经验。当指纹相似度在80%以上时,将经验作为参考但允许重新规划。这是避免"过度依赖旧经验"和"重复犯同种错误"之间平衡的关键机制。

四、LAM当前的能力边界与现实约束

多模态感知的延迟问题:在每一步操作前后都进行完整的环境感知(截图+DOM解析+API状态查询)会引入显著的延迟。对于需要10步以上的复杂任务,额外延迟可能达到5-10秒。生产优化需要引入"必要性判断"——只在状态可能发生变化(跳转新页面、弹窗出现、元素刷新)时才重新感知。

经验记忆的泛化能力:不同网站的操作经验很难互相迁移。"在某电商平台的添加购物车操作"与"在另一个平台的添加购物车操作"虽然逻辑相同,但元素定位完全不同。经验记忆的泛化需要更抽象的表征——不是记录CSS选择器,而是记录操作语义("找到页面中文本为'加入购物车'的button元素并点击")。

安全边界与权限控制:LAM的自主操作能力越强,安全风险越高。自动操作删除按钮、提交表单、甚至发起转账——这些都必须经过"人工确认门槛"的配置。建议将操作分为三个安全级别:L1(读取操作,无需确认)、L2(修改操作,需单次确认)、L3(敏感操作,需双重确认并留审计日志)。

结论

从LLM到LAM的演进不是替代关系,而是能力扩展。LAM = LLM(推理能力)+ Perception(环境感知)+ Execution(行动执行)+ Reflection(经验反思)。当前阶段,LAM最适合的场景是操作路径相对固定但需要高可靠性的自动化任务——比如数据录入、报表生成、系统巡检。对于高度创新和开放性的任务(如产品设计、战略分析),LLM依然是更合适的选择。

落地建议:不要在Agent产品中一步到位地实现完整的LAM架构。先从反思层开始——给现有Agent增加经验记忆能力,记录每次操作的结果,在下次相似操作时给出提示。这一步的性价比最高,投资回报最明确。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐