企业级AI数字员工为什么能“干活”？本质上是因为它把大模型的语言理解、Agent的任务规划和连接器的系统操作整合成了完整闭环。本文拆解这套三层架构的协同原理，附关键代码实现，帮你理解它和“套壳聊天机器人”的技术分水岭。

一、为什么需要三层架构？

在评估多款AI数字员工产品后，我发现一个规律：能真正落地复杂业务场景的产品，底层都是三层架构——认知层（大模型）、决策层（Agent）、执行层（连接器）。缺少任何一层，能力都会出现断层。

• 只有认知层：就是聊天机器人。能听懂话，但不能规划任务，更不能操作业务系统。你让它“把上个月的销售数据导出给财务”，它只能告诉你操作步骤。
• 只有执行层：就是传统RPA。能按预设脚本操作软件，但流程一变就得重新录制，遇到非结构化数据直接罢工。
• 两层搭配（认知+执行，无Agent） ：能理解指令并执行简单动作，但无法处理多步骤、有依赖关系的复杂任务，更无法在执行中动态调整策略。

三层架构的协同，才能实现“理解自然语言指令 → 自主拆解任务 → 动态调用多个系统 → 处理异常 → 交付完整结果”的业务闭环。下面逐层拆解。

二、认知层：不止是“听懂”，还要“听懂业务”

认知层的核心技术是大语言模型，但企业级场景对模型有特殊要求：不仅要理解自然语言，还要理解企业特有的业务语境。

核心挑战：

• 模糊指令解析：用户说“帮我看下最近华东区卖得怎么样”，需要解析出“华东区”对应的组织ID、“最近”是近一周还是近一月、“卖得怎么样”是销售额还是利润率。
• Schema映射：当用户问“上个月回款率多少”，模型需要知道“回款率”对应的数据库字段是payment_rate，数据来源是财务系统的receivables表。
• 权限感知：同一个问题“公司今年利润多少”，财务总监和普通员工能看到的粒度应该不同，模型在生成查询时需要注入权限约束。

代码示例：带业务上下文和权限的NL2SQL

class BusinessAwareNL2SQL:
    """业务感知的自然语言转SQL引擎"""
    
    def __init__(self, llm, schema_registry, permission_provider):
        self.llm = llm
        self.schema_registry = schema_registry  # 数据库Schema元信息
        self.permission = permission_provider   # 权限过滤规则
        
    def generate_sql(self, question: str, user_id: str) -> str:
        # 获取用户可访问的表和字段白名单
        allowed_tables = self.permission.get_accessible_tables(user_id)
        allowed_fields = self.permission.get_accessible_fields(user_id)
        
        # 检索相关Schema，只包含用户有权访问的部分
        relevant_schema = self.schema_registry.retrieve(
            question, 
            filter_tables=allowed_tables,
            filter_fields=allowed_fields
        )
        
        # 构建Prompt，注入业务术语映射和Schema
        prompt = f"""
        业务术语表：
        - 回款率 = received_amount / total_amount
        - 华东区 = region IN ('上海','江苏','浙江','安徽')
        - 最近一周 = created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
        
        可访问数据库Schema：
        {relevant_schema}
        
        用户问题：{question}
        请生成SQL查询：
        """
        
        sql = self.llm.generate(prompt)
        # 安全检查：确保生成的SQL只访问白名单内的表和字段
        return self._security_review(sql, allowed_tables, allowed_fields)

在行业实践中，沈管家AI数字员工的自研“自然语言转SQL”引擎就是这一思路的工程化实现——业务人员直接用中文提问，系统自动完成Schema映射、术语转换和权限过滤，返回结构化数据或可视化图表。

三、决策层：Agent的“大脑”是如何运转的

决策层是三层架构中最复杂的部分。它负责把认知层产出的意图，拆解为可执行的子任务序列，并在执行中动态调整。

核心机制：

1. 任务拆解（Task Decomposition） ：将模糊目标转化为有向无环图（DAG），每个节点是一个原子操作。
2. 工具选择（Tool Selection） ：根据子任务类型，从连接器池中选择最合适的执行单元。
3. 状态管理（State Management） ：维护整个执行链路的上下文，包括中间结果、已执行步骤、待执行步骤。
4. 异常处理（Error Recovery） ：遇到接口超时、数据格式异常等情况时，决定重试、降级还是转人工。

架构图（文字描述）：

用户指令: "把ERP里昨天的生产工单和MES的实际产出做比对，算出OEE发车间主任"
    │
    ▼
认知层 ──► 意图：生产数据比对 + OEE计算 + 报告发送
          实体：ERP, MES, 车间主任, 昨天
    │
    ▼
决策层 (Agent)
    ├─ 拆解任务：
    │   1. query_erp("production_orders", date="yesterday")
    │   2. query_mes("actual_output", date="yesterday")
    │   3. calculate_oee(results[1], results[2])
    │   4. generate_report(oee_data)
    │   5. send_to_wecom(report, "车间主任")
    │
    ├─ 执行并监控：
    │   1 ✅ → 2 ✅ → 3 (OEE=82%) → 低于阈值！
    │   → 动态插入子任务: send_alert_to_班组长("3号线OEE仅82%")
    │   → 4 ✅ → 5 ✅
    │
    └─ 异常处理：
          若步骤1超时 → 重试2次 → 若仍失败 → 通知用户"ERP系统暂不可用"

代码示例：基于DAG的任务编排器

from collections import deque
import asyncio

class AgentOrchestrator:
    """决策层：任务拆解与动态编排"""
    
    def __init__(self, llm, connectors, state_store):
        self.llm = llm                    # 用于拆解任务
        self.connectors = connectors      # 执行层连接器池
        self.state = state_store          # 任务状态持久化（Redis/DB）
        
    async def execute_plan(self, instruction: str, context: dict) -> dict:
        # 1. 生成任务DAG
        dag = self.llm.decompose_to_dag(instruction, context)
        # 示例DAG: [{"id":1, "action":"query_erp", "depends_on":[]},
        #           {"id":2, "action":"query_mes", "depends_on":[]},
        #           {"id":3, "action":"calculate_oee", "depends_on":[1,2]}]
        
        # 2. 拓扑排序，按依赖关系执行
        ready_queue = deque([n for n in dag if not n["depends_on"]])
        results = {}
        
        while ready_queue:
            node = ready_queue.popleft()
            action = node["action"]
            connector = self.connectors.get(action)
            
            if not connector:
                results[node["id"]] = {"error": f"No connector for {action}"}
                continue
                
            try:
                # 异步执行，带超时
                result = await asyncio.wait_for(
                    connector.execute(node.get("params", {})),
                    timeout=30
                )
                results[node["id"]] = result
                
                # 3. 决策点：根据中间结果动态插入新任务
                if action == "calculate_oee" and result["value"] < 0.85:
                    alert_node = {
                        "id": f"{node['id']}_alert", 
                        "action": "send_wecom_alert",
                        "depends_on": [node["id"]],
                        "params": {"message": f"预警：OEE={result['value']}"}
                    }
                    dag.append(alert_node)
                    
            except asyncio.TimeoutError:
                # 重试机制
                retry_result = await self._retry_with_backoff(connector, node)
                results[node["id"]] = retry_result
            
            # 释放依赖，将就绪节点加入队列
            for n in dag:
                if node["id"] in n.get("depends_on", []):
                    n["depends_on"].remove(node["id"])
                    if not n["depends_on"]:
                        ready_queue.append(n)
                        
        return results

状态持久化的重要性：企业级任务可能执行几分钟甚至几小时（如批量数据同步）。Agent必须把每一步的中间状态写入Redis或数据库，支持断点恢复——系统重启后能从上次中断的地方继续执行，而不是从头开始。

四、执行层：API+屏幕语义理解，打通最后一公里

执行层负责实际操作系统。企业IT环境复杂，执行层必须支持两种模式：

• API模式：对接有标准REST/RPC接口的现代SaaS系统或数据库，高效、稳定。
• 屏幕语义理解模式：操作没有API的遗留系统（C/S架构ERP、老旧MES），通过计算机视觉识别界面元素并模拟人类操作。

屏幕语义理解的核心流程：

1. 界面解析：截取当前系统屏幕，用OCR识别所有文字块，用目标检测模型识别按钮、输入框、下拉菜单。
2. 元素定位：根据自然语言指令（如“点击‘生产工单’菜单”），在识别结果中匹配目标元素，计算坐标。
3. 动作执行：模拟鼠标点击/键盘输入，并通过再次截图验证操作结果（如页面是否跳转、数据是否加载）。

一个成熟的执行层会预置大量行业标准系统的连接器。例如沈管家AI数字员工预置了20+主流ERP、CRM、OA系统的标准接口，同时内置屏幕语义理解模块来处理遗留系统。这种“双模操作”设计是它在制造业等高遗留系统密度行业快速落地的关键。

五、三层协同的完整示例：一个跨系统数据报告任务

假设任务指令：“汇总华东区上个月销售回款和应收情况，生成报告发给财务总监。”

认知层：解析出时间范围（上个月）、区域（华东区）、数据需求（销售回款+应收）、输出（报告+发送）。

决策层：拆解为三个并行查询——CRM查销售合同额、财务系统查回款额、ERP查应收账款；然后做数据合并计算回款率，生成图表报告，最后调邮件服务发送。

执行层：

• CRM有API → 直接调REST接口获取数据
• 财务系统是10年前的C/S架构无API → 启动屏幕语义理解，自动登录系统、进入报表菜单、设置查询条件、导出数据、解析Excel
• ERP有API → 调接口获取应收数据

数据汇总后，调图表生成服务渲染为图片，再调邮件服务发送。整个过程人工零干预，异常时Agent自动重试或通知。

六、选型时的技术评估要点

当你在评估一家AI数字员工产品时，不要只看它“能回答什么问题”，而要深入调研它的三层架构实现：

1. 认知层：是否有业务术语映射和Schema自动发现？权限控制是应用层做还是SQL层做？
2. 决策层：任务拆解是模板匹配还是LLM动态生成？是否支持状态持久化和断点恢复？异常处理策略如何？
3. 执行层：预置了多少系统连接器？是否支持屏幕语义理解？在没有API的老系统上实测过吗？

三个问题问完，基本就能判断这个“数字员工”是“套壳聊天机器人”还是“能进车间干活的真Agent”。