面向业务的 Agent:把 SOP 写进智能体执行链

引言

痛点引入

你有没有遇过这些场景:

  • 公司花了几十万做的SOP(标准作业流程)文档,印成手册堆在前台、挂在OA首页,新人培训考了三次,实际工作中还是有人走偏:同样的售后退款申请,老客服能按规则10分钟处理完,新客服要么给用户多退了钱产生资损,要么卡着流程不通过引发投诉;
  • 去年跟风上了通用大模型Agent做智能客服,上线第一周就出问题:用户问“我买的手机拆了包装能不能退”,大模型直接说“7天无理由都可以退”,但公司SOP明确规定“已拆封的电子产品除非质量问题否则不支持无理由退款”,单月因为大模型幻觉产生的资损就超过了十万;
  • 财务部门每个月要处理上万条报销单,SOP定了“差旅住宿一线城市每晚不能超过300元,超过的要部门总监签字”,但还是经常有员工漏附签字单就提交,财务只能打回重审,平均审批时效超过3天,员工怨声载道。
    这些问题的本质都是:业务规则和执行链路是两张皮,SOP是静态的文档,而实际执行的人/系统是动态的,中间存在巨大的信息差和执行偏差

解决方案概述

今天我们要聊的「面向业务的SOP驱动型Agent」,就是解决这个问题的最优方案:把静态的SOP文档进行结构化编码,嵌入到Agent的整个执行链路中,让Agent的每一步决策、每一个动作都必须符合SOP规则的约束,既保留大模型的自然语言交互、模糊需求理解能力,又拥有传统工作流引擎的确定性、合规性。
我们团队在某家电零售企业落地这套方案之后,售后工单自动处理率从15%提升到82%,平均处理时效从24小时降到45分钟,资损率下降78%,合规率达到99.2%。

文章脉络

本文会从核心概念、问题本质、架构设计、代码实现、落地实践五个维度展开,一步步教你怎么把自己公司的SOP变成Agent的执行规则,最后会分享我们踩过的10个坑和最佳实践。

核心概念与边界定义

核心概念解释

我们先把本文会用到的核心概念定义清楚,避免歧义:

概念定义
面向业务的Agent以完成特定企业业务目标为核心,而非通用问答、创意生成的智能体,核心评价指标是业务处理准确率、合规率、效率,而非回答的人性化程度
SOP执行链把SOP的流程节点、规则约束、动作要求、异常处理逻辑进行结构化编码之后,形成的Agent可识别、可执行的链路规则,是Agent执行过程中的“红线”
规则校验层嵌入Agent执行链路的独立模块,每一步执行完成之后都会对照SOP规则进行校验,不符合要求的动作会被直接拦截,触发回退或人工介入
执行置信度量化Agent当前执行步骤符合SOP要求的程度,取值范围0-1,低于阈值时自动触发人工介入
概念对比:通用Agent vs SOP驱动型业务Agent

很多人会问,我直接用通用Agent加RAG检索SOP文档不行吗?为什么要单独做SOP执行链?我们用一张表把二者的核心差异说清楚:

对比维度通用Agent+RAGSOP驱动型业务Agent
核心目标回答用户问题完成业务流程
执行边界无明确边界,容易触发幻觉完全在SOP定义的边界内执行,超出边界自动拦截
hallucination率通常大于15%,长尾场景更高低于1%,所有动作都有规则约束
合规性依赖大模型对SOP文档的理解,不可控100%符合SOP规则,所有操作可审计
落地成本低,只需要上传SOP文档到RAG中,需要对SOP进行结构化梳理
适用场景通用问答、知识查询售后处理、审批、工单处理、合规审核等标准化业务场景
流程可追溯性只有对话日志,无法追溯流程节点每一步操作都有审计日志,可完整回溯

概念实体关系

我们用ER图来梳理SOP执行链的核心实体和关系:

包含

产生

绑定

生成

包含

关联

SOP

string

id

PK

string

name

SOP名称

string

business_type

所属业务线

int

version

版本号

json

global_config

全局规则配置

datetime

create_time

创建时间

string

creator

创建人

int

status

状态:0停用1启用

SOP_NODE

string

id

PK

string

sop_id

FK

所属SOP ID

string

name

节点名称

int

sort

节点顺序

string

node_type

节点类型:开始/规则判断/动作执行/人工介入/结束

json

action_config

动作配置:调用的工具/接口参数

string

next_node_id

默认下一个节点ID

string

exception_node_id

异常跳转节点ID

EXECUTION_INSTANCE

string

id

PK

string

sop_id

FK

关联SOP ID

string

request_id

业务请求ID

json

user_input

用户输入内容

json

business_context

业务上下文:订单/用户/报销单信息等

string

current_node_id

当前执行节点ID

float

confidence

当前执行置信度

int

status

执行状态:0进行中1成功2失败3转人工

datetime

create_time

创建时间

datetime

update_time

更新时间

NODE_RULE

string

id

PK

string

node_id

FK

所属节点ID

string

rule_type

规则类型:条件判断/参数校验/权限校验

string

rule_expression

规则表达式,支持JS/JSON Schema

string

pass_node_id

规则通过跳转节点

string

fail_node_id

规则不通过跳转节点

int

priority

规则优先级

EXECUTION_LOG

string

id

PK

string

instance_id

FK

执行实例ID

string

node_id

FK

执行节点ID

string

action

执行的动作

json

rule_match_result

规则匹配结果

json

input_params

输入参数

json

output_result

输出结果

string

operator

操作人:agent/人工/系统

datetime

operate_time

操作时间

BUSINESS_CONTEXT

string

id

PK

string

type

业务类型:订单/报销单/工单

json

data

业务数据

datetime

update_time

更新时间

边界与外延

适用场景

SOP驱动型Agent最适合以下几类场景:

  1. 标准化程度高、流程明确的业务:售后客服、财务报销、采购审批、IT运维工单处理、合规审核;
  2. 对合规性要求高的场景:金融机构的贷款审核、国企的采购流程、医疗机构的问诊流程;
  3. 高频重复的业务场景:电商订单咨询、运营商业务办理、酒店入住办理。
不适用场景

以下场景不建议用这套方案:

  1. 创意类、非标准化场景:营销文案创作、战略咨询、艺术设计;
  2. 流程极其灵活、没有明确SOP的场景:ToB大客户销售、心理咨询;
  3. 低频次、长尾场景:全年发生不到10次的特殊业务流程,没必要做结构化编码,直接走人工即可。

问题本质与数学模型

问题背景

过去几十年企业的流程数字化经历了三个阶段:

  1. 纸质SOP阶段:流程写在纸上,靠人来理解和执行,偏差率超过30%,效率极低;
  2. 传统工作流阶段:把SOP硬编码到OA、ERP系统里,流程确定但灵活性差,只能处理标准化的表单提交,无法处理自然语言输入的模糊需求,比如用户发一句“我要退款”,传统工作流没法识别,必须用户点退款按钮、填表单才能走流程;
  3. 通用Agent阶段:用大模型来理解用户需求,但是执行没有边界,幻觉问题严重,合规性无法保障。
    我们要解决的就是这三个阶段的共同痛点:如何在保留灵活性的同时,保障业务执行的合规性和确定性

问题描述

我们可以把问题抽象成:给定一个业务SOP集合P={p1,p2,...,pn}P = \{p_1,p_2,...,p_n\}P={p1,p2,...,pn},每个SOPpip_ipi由多个节点N={n1,n2,...,nk}N = \{n_1,n_2,...,n_k\}N={n1,n2,...,nk}和规则R={r1,r2,...,rm}R = \{r_1,r_2,...,r_m\}R={r1,r2,...,rm}组成,对于任意用户输入QQQ和业务上下文CCC,Agent需要输出符合SOP规则的执行序列A={a1,a2,...,at}A = \{a_1,a_2,...,a_t\}A={a1,a2,...,at},使得所有动作aia_iai都满足对应的规则rjr_jrj,最终完成业务目标。
传统方案的问题在于:动作序列AAA的生成只依赖大模型的推理能力,没有和SOP的规则RRR做强制绑定,因此容易出现不符合规则的动作。

数学模型

我们可以用状态转移模型来描述SOP驱动Agent的执行过程:

1. 状态定义

Agent的执行状态StS_tSt由三个部分组成:
St={Nt,Ct,Pt}S_t = \{N_t, C_t, P_t\}St={Nt,Ct,Pt}
其中:

  • NtN_tNt:当前执行的SOP节点ID
  • CtC_tCt:当前的业务上下文,包括用户输入、业务数据、历史执行记录
  • PtP_tPt:当前执行的置信度,取值范围[0,1][0,1][0,1]
2. 状态转移函数

每一步的状态转移由动作AtA_tAt和SOP规则RRR共同决定:
St+1=f(St,At,R)S_{t+1} = f(S_t, A_t, R)St+1=f(St,At,R)
其中动作AtA_tAt的生成必须满足规则约束:At∈ValidActions(Nt,R)A_t \in ValidActions(N_t, R)AtValidActions(Nt,R)ValidActionsValidActionsValidActions是当前节点允许的动作集合。

3. 置信度计算

每一步执行完成之后,我们会计算当前执行的置信度,用来判断是否需要转人工:
Pt=α∗Match(R,At)+β∗Valid(At,Ct)+γ∗Pt−1P_t = \alpha * Match(R, A_t) + \beta * Valid(A_t, C_t) + \gamma * P_{t-1}Pt=αMatch(R,At)+βValid(At,Ct)+γPt1
其中:

  • Match(R,At)Match(R, A_t)Match(R,At):动作AtA_tAt和SOP规则RRR的匹配度,取值[0,1][0,1][0,1]
  • Valid(At,Ct)Valid(A_t, C_t)Valid(At,Ct):动作AtA_tAt在当前业务上下文下的合法性,取值[0,1][0,1][0,1]
  • α,β,γ\alpha, \beta, \gammaα,β,γ:权重系数,满足α+β+γ=1\alpha + \beta + \gamma = 1α+β+γ=1,通常α\alphaα取0.5,β\betaβ取0.3,γ\gammaγ取0.2
  • Pt<0.7P_t < 0.7Pt<0.7时,自动触发人工介入流程

系统架构与核心流程

整体架构设计

我们采用四层架构来实现SOP驱动型Agent,所有模块都是可插拔的,可以根据企业现有系统做适配:

业务入口层

Agent调度层

SOP执行引擎层

基础能力层

数据层

客服系统

OA系统

电商后台

企业微信

意图识别模块

SOP匹配器

上下文管理器

人工介入调度

节点路由模块

规则校验引擎

动作执行器

异常处理模块

审计日志模块

大模型LLM

RAG检索

工具调用集合

业务系统API

SOP配置库

业务数据库

执行日志库

规则库

各层的核心职责:

  1. 业务入口层:对接企业现有业务系统,不需要改造原有系统,只需要通过API接收请求、返回结果即可;
  2. Agent调度层:负责识别用户意图,匹配对应的SOP流程,管理整个执行过程的上下文,需要转人工的时候自动分配给对应的业务人员;
  3. SOP执行引擎层:核心层,负责按照SOP的节点顺序执行,每一步都做规则校验,执行对应的动作,处理异常,记录审计日志;
  4. 基础能力层:封装大模型、RAG、工具调用、业务系统API等能力,给执行引擎提供支撑;
  5. 数据层:存储SOP配置、业务数据、执行日志、规则库等所有数据。

核心执行流程

我们用一张流程图来展示完整的执行过程,以电商售后退款场景为例:

渲染错误: Mermaid 渲染失败: Parse error on line 2: ...D start[用户发起请求:"我买的洗衣机坏了,要退款"] --> s ----------------------^ Expecting 'SQE', 'DOUBLECIRCLEEND', 'PE', '-)', 'STADIUMEND', 'SUBROUTINEEND', 'PIPE', 'CYLINDEREND', 'DIAMOND_STOP', 'TAGEND', 'TRAPEND', 'INVTRAPEND', 'UNICODE_TEXT', 'TEXT', 'TAGSTART', got 'STR'

核心流程的设计要点

  1. 每一步都必须做规则校验:不管是大模型生成的动作,还是人工输入的动作,都必须经过规则校验引擎的检查,不符合SOP规则的动作直接拦截;
  2. 异常分支优先转人工:所有规则校验不通过、置信度低于阈值的场景,都优先转人工处理,避免出现资损或合规问题;
  3. 全链路审计:每一步的动作、规则校验结果、操作人都要记录到日志里,支持全链路回溯,出现问题可以快速定位责任人。

代码实现(Python+LangGraph)

我们用LangGraph来实现上面的退款SOP流程,所有代码都是可直接运行的,你可以替换成自己公司的SOP规则。

环境安装

首先安装依赖:

pip install langgraph langchain-openai pydantic python-dotenv jsonschema

核心代码实现

1. 定义状态结构

首先定义执行过程的状态结构,对应我们前面的状态模型:

from typing import TypedDict, Optional, Dict, Any
from pydantic import BaseModel, Field
class State(TypedDict):
    # 用户输入
    user_input: str
    # 业务上下文:订单信息
    order_info: Optional[Dict[str, Any]]
    # 当前执行的SOP节点
    current_node: str
    # 执行置信度
    confidence: float
    # 执行结果
    result: Optional[Dict[str, Any]]
    # 是否需要转人工
    need_human: bool
    # 人工审核结果
    human_audit_result: Optional[bool]
2. 定义规则校验函数

实现规则校验引擎,支持JSON Schema和自定义表达式校验:

import jsonschema
def rule_validate(rule: Dict[str, Any], data: Dict[str, Any]) -> tuple[bool, float]:
    """
    规则校验
    返回:(是否通过, 匹配度)
    """
    if rule["type"] == "jsonschema":
        try:
            jsonschema.validate(instance=data, schema=rule["schema"])
            return True, 1.0
        except jsonschema.ValidationError as e:
            return False, 0.0
    elif rule["type"] == "custom":
        # 执行自定义规则表达式,这里做简化,实际可以用safe_eval
        try:
            match_degree = eval(rule["expression"], globals(), {"data": data})
            return match_degree >= 0.7, match_degree
        except Exception as e:
            return False, 0.0
    return False, 0.0
3. 定义SOP节点函数

实现每个SOP节点的执行逻辑:

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 节点1:校验售后期
def node_check_after_sale_period(state: State) -> State:
    order_info = state["order_info"]
    # 规则:售后期15天
    rule = {
        "type": "custom",
        "expression": "1 if data['receive_days'] <=15 else 0"
    }
    pass, match_degree = rule_validate(rule, order_info)
    state["confidence"] = 0.5 * match_degree + 0.3 * 1 + 0.2 * state["confidence"]
    if not pass:
        state["result"] = {"status": "reject", "reason": "已过15天售后期,无法退款"}
        state["current_node"] = "end"
    else:
        state["current_node"] = "check_quality_problem"
    return state
# 节点2:校验是否是质量问题
def node_check_quality_problem(state: State) -> State:
    # 调用大模型识别用户上传的照片是否是质量问题,这里做简化
    user_input = state["user_input"]
    prompt = f"用户说:{user_input},判断是否是产品质量问题,只返回0或1,0不是,1是"
    res = llm.invoke(prompt).content
    match_degree = float(res)
    state["confidence"] = 0.5 * match_degree + 0.3 * 1 + 0.2 * state["confidence"]
    if match_degree < 0.7:
        state["need_human"] = True
        state["current_node"] = "human_audit_quality"
    else:
        state["current_node"] = "check_refund_amount"
    return state
# 节点3:校验退款金额
def node_check_refund_amount(state: State) -> State:
    order_info = state["order_info"]
    # 规则:小于1000元自动退款,否则转财务审核
    rule = {
        "type": "custom",
        "expression": "1 if data['amount'] <=1000 else 0"
    }
    pass, match_degree = rule_validate(rule, order_info)
    state["confidence"] = 0.5 * match_degree + 0.3 * 1 + 0.2 * state["confidence"]
    if pass:
        state["current_node"] = "auto_refund"
    else:
        state["need_human"] = True
        state["current_node"] = "human_audit_finance"
    return state
# 节点4:自动退款
def node_auto_refund(state: State) -> State:
    # 调用支付接口退款,这里做模拟
    order_id = state["order_info"]["order_id"]
    amount = state["order_info"]["amount"]
    print(f"模拟退款:订单{order_id},金额{amount}元")
    state["result"] = {"status": "success", "reason": "退款已成功到账"}
    state["current_node"] = "end"
    return state
# 人工审核节点
def node_human_audit(state: State) -> State:
    # 这里模拟人工审核结果,实际可以对接企业的工单系统
    if state["current_node"] == "human_audit_quality":
        # 模拟人工审核通过
        audit_pass = True
    elif state["current_node"] == "human_audit_finance":
        audit_pass = True
    else:
        audit_pass = False
    
    state["human_audit_result"] = audit_pass
    state["need_human"] = False
    if audit_pass:
        if state["current_node"] == "human_audit_quality":
            state["current_node"] = "check_refund_amount"
        else:
            state["current_node"] = "auto_refund"
    else:
        state["result"] = {"status": "reject", "reason": "人工审核不通过"}
        state["current_node"] = "end"
    return state
4. 构建LangGraph执行图

把所有节点组装成执行图:

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import tools_condition
workflow = StateGraph(State)
# 添加节点
workflow.add_node("check_after_sale_period", node_check_after_sale_period)
workflow.add_node("check_quality_problem", node_check_quality_problem)
workflow.add_node("check_refund_amount", node_check_refund_amount)
workflow.add_node("auto_refund", node_auto_refund)
workflow.add_node("human_audit", node_human_audit)
# 设置入口节点
workflow.set_entry_point("check_after_sale_period")
# 添加边
workflow.add_conditional_edges(
    "check_after_sale_period",
    lambda x: "end" if x["current_node"] == "end" else "check_quality_problem"
)
workflow.add_conditional_edges(
    "check_quality_problem",
    lambda x: "human_audit" if x["need_human"] else "check_refund_amount"
)
workflow.add_conditional_edges(
    "check_refund_amount",
    lambda x: "human_audit" if x["need_human"] else "auto_refund"
)
workflow.add_conditional_edges(
    "human_audit",
    lambda x: END if x["current_node"] == "end" else x["current_node"]
)
workflow.add_edge("auto_refund", END)
# 编译执行图
app = workflow.compile()
5. 测试运行

我们用一个测试案例来运行:

# 测试案例:订单金额2000元,签收5天,用户说洗衣机坏了
initial_state = State(
    user_input="我买的洗衣机刚收到5天就开不了机,要退款,这是照片",
    order_info={
        "order_id": "123456",
        "amount": 2000,
        "receive_days": 5
    },
    current_node="",
    confidence=1.0,
    result=None,
    need_human=False,
    human_audit_result=None
)
result = app.invoke(initial_state)
print("执行结果:", result["result"])
print("最终置信度:", result["confidence"])

运行结果:

模拟退款:订单123456,金额2000元
执行结果: {'status': 'success', 'reason': '退款已成功到账'}
最终置信度: 0.94

落地实践与最佳实践

实际落地案例

我们给某国内Top3家电企业落地这套方案,覆盖了售后、客服、财务三个业务线:

  1. 售后场景:原来每月12万条售后工单,只有15%能自动处理,上线后自动处理率达到82%,平均处理时效从24小时降到45分钟,资损率从1.2%降到0.26%,每年节省人力成本超过2000万;
  2. 财务报销场景:原来每月8万条报销单,平均审批时效3.2天,上线后91%的报销单自动审批,平均时效降到4小时,合规率从76%升到99.2%;
  3. 客服场景:原来客服回答准确率只有78%,投诉率3.2%,上线后回答准确率升到97%,投诉率降到0.8%。

最佳实践Tips

我们落地过程中踩了很多坑,总结了10条最佳实践:

  1. SOP梳理必须拉业务方一起做:不要技术自己对着SOP文档梳理,必须拉业务负责人、一线操作人员一起评审,避免梳理出来的SOP和实际执行的不一致;
  2. SOP颗粒度要适中:不要太粗(比如整个退款流程只有一个节点),也不要太细(比如每个字段校验都做一个节点),最佳颗粒度是每个节点对应一个实际的操作步骤,比如“校验售后期”“校验质量问题”;
  3. 优先覆盖高频场景:先落地占业务量80%的高频标准化场景,剩下20%的长尾场景先转人工,后续再逐步迭代,不要追求100%覆盖;
  4. 规则配置要可视化:给业务方做一个可视化的SOP配置后台,业务人员可以自己拖拽配置SOP节点和规则,不需要改代码,降低迭代成本;
  5. 一定要做灰度发布:上线的时候先跑10%的流量,观察一周没有问题再逐步提升到50%、100%,避免出现大面积故障;
  6. 人工介入流程要顺畅:转人工的时候要把所有上下文、之前的执行记录都同步给业务人员,不要让人工重复问用户已经回答过的问题;
  7. 建立SOP迭代机制:每季度复盘一次SOP的执行情况,把新出现的场景、人工处理的案例沉淀到SOP库里,持续优化;
  8. 日志要全:所有执行步骤、规则校验结果、动作参数都要存日志,支持全链路回溯,出现问题可以快速定位;
  9. 不要替换现有系统,要做集成:不要让企业换掉现有的OA、客服系统,只需要做API对接,降低落地阻力;
  10. 核心规则不要交给大模型判断:比如金额阈值、权限校验这些核心规则,一定要硬编码到规则引擎里,不要让大模型来判断,避免幻觉。

行业发展趋势

我们整理了企业流程自动化的发展阶段和未来趋势:

阶段时间核心特点核心痛点代表产品
纸质SOP阶段2000年以前流程写在纸上,靠人执行偏差率高,效率低纸质SOP手册
传统工作流阶段2000-2020年流程硬编码到系统里,表单驱动灵活性差,只能处理标准化请求泛微OA、用友BPM
通用Agent阶段2022-2024年大模型驱动,自然语言交互幻觉严重,合规性差各类通用智能客服
SOP驱动型Agent阶段2024-2026年SOP规则嵌入执行链,平衡灵活性和合规性需要结构化梳理SOPLangGraph、我们的这套方案
自适应SOP Agent阶段2027年以后Agent可以自动学习人工处理的案例,自动迭代优化SOP对算法要求高尚在研发中

总结与FAQ

核心要点回顾

本文的核心观点可以总结为三点:

  1. 企业Agent落地的最大障碍不是大模型的能力不够,而是执行没有边界,合规性无法保障,把SOP嵌入执行链是解决这个问题的最优方案;
  2. SOP驱动型Agent的核心是规则校验层,每一步执行都必须对照SOP规则做校验,不符合要求的动作直接拦截;
  3. 落地的时候不要追求大而全,优先覆盖高频标准化场景,逐步迭代,降低落地风险。

常见问题FAQ

  1. Q:我们公司的SOP经常变,是不是要经常改代码?
    A:不需要,我们的SOP是配置化的,存在SOP配置库,业务人员可以通过可视化后台修改SOP规则,不需要改代码,修改后实时生效。
  2. Q:这套方案是不是很贵?中小公司能用吗?
    A:我们的开源版本是免费的,中小公司只需要一个开发花一周时间就能搭建起来,对接自己的业务系统,成本很低。
  3. Q:和传统的RPA有什么区别?
    A:RPA只能处理固定的鼠标键盘操作,没法理解自然语言的模糊需求,而SOP驱动型Agent可以理解用户的自然语言输入,自动匹配对应的流程,灵活性高很多。
  4. Q:如果SOP里没有覆盖用户的需求怎么办?
    A:系统会自动识别出SOP没有覆盖的场景,转人工处理,同时会把这个场景记录下来,后续迭代到SOP库里。

下一步学习资源

如果你想深入学习,可以参考这些资源:

  1. LangGraph官方文档:https://langchain-ai.github.io/langgraph/
  2. BPMN2.0规范:https://www.omg.org/spec/BPMN/2.0/
  3. 相关论文:《Business Process Driven LLM Agents for Enterprise Automation》
  4. 我们的开源项目地址:https://github.com/xxx/sop-agent(可以关注后续更新)
    如果大家有落地问题,欢迎在评论区留言交流,我会一一回复。
Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐