Codex Harness × 企业 ERP：构建智能化企业系统完整方案

背景：2024年OpenAI将 Codex CLI（codex）在 GitHub 开源，其核心 Harness 框架将 LLM 包裹在安全的执行环境中，使 AI 能够受控地操作真实系统。结合企业 ERP（SAP/Oracle/用友/金蝶）的特点，Harness 正是构建企业智能化系统缺失的那块拼图。

---

一、OpenAI Codex Harness 核心解析

1.1 Harness 是什么？

Harness（执行框架）是 Codex 最核心的创新——它不是简单的 API 包装，而是一套完整的智能体执行基础设施：

组件	功能	ERP场景映射
Agent Loop	多轮推理-执行循环，持久化状态	处理跨多天的ERP业务流程
Tool Executor	file/shell/MCP 工具统一调度	调用SAP BAPI、OData、RFC接口
Approval Engine	操作分级审批，diff预览确认	ERP高风险操作四眼审批
Sandbox Manager	Docker隔离，资源限额	代码在DEV系统而非PRD执行
Context Engine	Token预算管理，滚动压缩	超长ERP流程的上下文保持

1.2 三种执行模式

suggest    →  仅给出建议，不执行任何操作（只读场景）
auto-edit  →  自动修改文件，Shell需确认（开发辅助场景）
full-auto  →  全自动执行（已授权的固定流程场景）

1.3 MCP：连接一切的标准协议

Model Context Protocol（MCP） 是 Harness 的扩展点，任何系统都可以暴露 MCP Server 接入 Harness：

Harness ←→ SAP MCP Server    ←→ SAP BAPI/RFC/OData
        ←→ Oracle MCP Server  ←→ Oracle REST API / PL-SQL
        ←→ DB MCP Server      ←→ HANA / PostgreSQL（只读）
        ←→ Qdrant MCP Server  ←→ 企业知识库向量检索
        ←→ 自定义 MCP Server  ←→ 任意内部系统

---

二、ERP 系统特点与 AI 化挑战

2.1 企业ERP的核心特征（为什么难以AI化）

① 数据结构极其复杂：SAP 标准表超过 90,000 张，业务对象之间存在数百个 BAPI/RFC 接口。传统 AI 无法理解 EKKO（采购头表）和 EKES（收货记录）之间的业务关联。

② 操作风险极高：误操作财务凭证可能导致合规问题，跨期操作影响财务报表，权限体系复杂（授权对象 + 角色 + 用户组）。

③ 系统高度定制：每家企业的 ERP 都经过大量客制化（ABAP Enhancement / BAdi / User Exit），标准文档无法覆盖企业特有逻辑。

④ 变更管控严格：SAP 的变更通过 ChaRM（Change Request Management）管理，DEV→QAS→PRD 三层传输。

2.2 Harness 解决的核心问题

ERP挑战	Harness 解决方案
接口复杂	MCP Server 封装，标准化工具接口
操作风险	Approval Engine 分级审批，diff预览
系统定制	RAG知识库 + 企业文档向量索引
变更管控	Sandbox隔离，只在DEV执行代码
审计合规	全链路操作日志，不可篡改

---

三、六大核心应用场景

3.1 场景一：ERP 代码生成（ABAP/PL-SQL）

这是 Harness 最直接的价值场景。传统 ABAP 开发需要专业顾问，开发周期长，成本高。

# ABAP 代码生成 Agent 示例
# 用户指令 → Harness → 自动生成/测试/传输 ABAP 程序

async def generate_abap_report(instruction: str, agent: ERPAgent):
    """
    用户: "生成一个报表，统计过去30天各供应商的到货准时率"
    """

    result = await agent.execute(instruction)

    # Harness 内部执行链：
    # 1. 读取 SAP 数据字典（EKKO采购头/EKPO行项目/EKES收货）
    #    → tool: sap_read_ddic(tables=["EKKO","EKPO","EKES"])
    #
    # 2. 生成 ABAP 程序（ALV报表）
    #    → tool: file_write("ZMM_VENDOR_OTD.abap", abap_code)
    #
    # 3. 在 DEV 系统上传并激活
    #    → tool: sap_run_abap(code=..., system="DEV")  # R2级，需确认
    #
    # 4. 执行测试并检查输出
    #    → tool: sap_execute_report("ZMM_VENDOR_OTD", params={...})
    #
    # 5. 如果测试通过，生成传输请求（Transport Request）
    #    → tool: sap_create_transport(objects=["ZMM_VENDOR_OTD"])
    #    注意：传输到PRD需要走正式ChaRM流程，Harness只创建传输请求

    print(result.output)
    # 输出示例：
    # "已成功创建ABAP报表 ZMM_VENDOR_OTD，
    #  在DEV系统测试通过（100条数据，执行时间0.3s），
    #  已创建传输请求 DE1K900123，请在ChaRM中发起PRD传输审批。"

    return result

3.2 场景二：智能数据分析

用户："比较今年和去年同期的销售毛利率，找出下降超过5%的产品类别"

Harness执行链：
  1. sap_read_data(table="VBAK/VBAP/COGS", period=["2025Q3","2026Q3"])
  2. file_write("analysis.py", pandas_code)   # 生成分析代码
  3. shell_exec("python analysis.py")          # 沙箱执行
  4. file_read("output.html")                  # 读取可视化结果
  → 输出：完整毛利率对比报告 + 预警产品清单

3.3 场景三：ERP 配置变更管理

传统痛点：ERP 配置变更（如修改审批阈值、价格策略、税码设置）需要顾问操作，缺乏变更影响分析。

Harness 方案：

1. 分析变更影响范围（受影响的业务对象、流程、报表）
2. 生成变更预览文档供审批
3. 在 DEV 系统执行并自动化回归测试
4. 通过 ChaRM 创建传输请求，走正式审批流到 PRD

---

四、Agent 工作流：SAP 采购到付款（P2P）全流程

4.1 MCP Server 实现

// SAP MCP Server 实现（TypeScript）
// 遵循 Model Context Protocol 标准，供 Codex Harness 调用

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SAPClient } from "@sap-cloud-sdk/connectivity";

const server = new Server({ name: "sap-erp-mcp", version: "1.0.0" }, {
  capabilities: { tools: {} }
});

// 工具1：查询SAP业务数据（只读，自动放行）
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "sap_read_po",
      description: "读取采购订单明细（只读操作，无需审批）",
      inputSchema: {
        type: "object",
        properties: {
          po_number: { type: "string", description: "采购订单号" },
          items:     { type: "boolean", description: "是否包含行项目" }
        },
        required: ["po_number"]
      }
    },
    {
      name: "sap_post_goods_receipt",
      description: "过账收货（R2级操作，需用户确认）",
      inputSchema: {
        type: "object",
        properties: {
          po_number:     { type: "string" },
          delivery_note: { type: "string" },
          posting_date:  { type: "string", format: "date" }
        },
        required: ["po_number", "posting_date"]
      }
    },
    {
      name: "sap_run_abap",
      description: "在DEV系统执行ABAP代码（沙箱隔离）",
      inputSchema: {
        type: "object",
        properties: {
          code:        { type: "string", description: "ABAP源代码" },
          system:      { type: "string", enum: ["DEV", "QAS"],  // 生产系统禁止直接执行
                        description: "目标系统（不允许PRD）" },
          max_runtime: { type: "integer", default: 30 }
        },
        required: ["code", "system"]
      }
    }
  ]
}));

4.2 审批引擎设计

// Harness 审批引擎：ERP操作风险分级
use std::sync::Arc;
use tokio::sync::oneshot;

#[derive(Debug, Clone, PartialEq)]
pub enum RiskLevel {
    R1ReadOnly,      // 只读查询 → 自动放行
    R2Modification,  // 数据修改 → 需用户确认
    R3Financial,     // 财务操作 → 双人审批
    R4CrossPeriod,   // 跨期操作 → 严格管控
    R5SystemAdmin,   // 系统管理 → 禁止Agent执行
}

pub struct ERPApprovalEngine {
    policy:    Arc<ApprovalPolicy>,
    notifier:  Arc<dyn ApprovalNotifier>,  // 钉钉/企微/邮件
    audit_log: Arc<AuditLogger>,
}

impl ERPApprovalEngine {
    pub async fn evaluate(&self, op: &ERPOperation) -> ApprovalDecision {
        let risk = self.classify_risk(op);

        match risk {
            RiskLevel::R1ReadOnly => ApprovalDecision::AutoApprove,

            RiskLevel::R2Modification => {
                // 展示 diff，等待用户点击确认
                let preview = op.generate_diff_preview();
                self.notifier.show_diff_and_wait(preview).await
            }

            RiskLevel::R3Financial => {
                // 四眼原则：发起人 + 审批人双人确认
                let (tx, rx) = oneshot::channel();
                self.notifier.send_approval_request(
                    op, vec![op.initiator.clone(), self.policy.find_approver(op)]
                ).await;
                rx.await.unwrap_or(ApprovalDecision::Reject)
            }

            RiskLevel::R4CrossPeriod | RiskLevel::R5SystemAdmin => {
                self.audit_log.record_blocked(op).await;
                ApprovalDecision::Reject  // 硬性拒绝
            }
        }
    }

    fn classify_risk(&self, op: &ERPOperation) -> RiskLevel {
        match op {
            op if op.is_read_only()          => RiskLevel::R1ReadOnly,
            op if op.involves_financial()    => RiskLevel::R3Financial,
            op if op.is_cross_period()       => RiskLevel::R4CrossPeriod,
            op if op.requires_basis_access() => RiskLevel::R5SystemAdmin,
            _                                => RiskLevel::R2Modification,
        }
    }
}

4.3 Agent 主循环

# Codex Harness ERP Agent 主循环（Python SDK）
from openai import OpenAI
from typing import AsyncIterator

class ERPAgent:
    """基于 Codex Harness 的 ERP 智能体"""

    def __init__(self, erp_mcp_servers: list[str]):
        self.client = OpenAI()
        self.mcp_servers = erp_mcp_servers   # SAP/Oracle MCP Server 地址
        self.context = ERPContext()           # 持久化业务上下文
        self.approval = ERPApprovalEngine()  # 风险审批引擎
        self.audit    = AuditLogger()        # 不可篡改审计日志

    async def execute(self, instruction: str, user: ERPUser) -> AgentResult:
        """执行自然语言ERP指令"""

        # 1. 构建携带业务上下文的提示
        system_prompt = self._build_erp_system_prompt(user)

        # 2. 启动 Harness Agent 循环
        async with self.client.beta.realtime.connect(model="codex-1") as harness:
            harness.config(
                tools        = self.mcp_servers,    # MCP工具集
                sandbox      = "erp-sandbox",        # 隔离环境
                approval_mode= "supervised",         # 监督模式
                max_turns    = 30,                   # 最大迭代轮数
                token_budget = 100_000,              # Token预算
            )

            # 3. 每次工具调用前过审批引擎
            @harness.on_tool_call
            async def before_tool(tool_call: ToolCall):
                erp_op = ERPOperation.from_tool_call(tool_call, user)
                decision = await self.approval.evaluate(erp_op)
                if decision == ApprovalDecision.Reject:
                    raise ToolCallRejected(f"操作被拒绝: {erp_op.risk_reason}")
                await self.audit.record(erp_op, decision)  # 审计记录

            result = await harness.run(instruction, system=system_prompt)

        return AgentResult(
            output      = result.text,
            trace       = self.audit.get_session_trace(),  # 完整操作链路
            erp_changes = result.tool_results,             # ERP变更汇总
        )

    def _build_erp_system_prompt(self, user: ERPUser) -> str:
        return f"""
你是企业ERP智能助手。当前用户: {user.name}，角色: {user.role}。
ERP系统: {self.context.erp_type}，公司代码: {self.context.company_code}。

操作权限范围:
- 只读操作: 自动执行无需确认
- 数据修改: 展示变更预览后等待用户确认
- 财务凭证: 必须触发双人审批流程
- 跨期操作: 严格禁止，需走正式变更流程

始终用中文回复。每个操作后说明已变更内容和影响范围。
"""

---

五、ERP 知识库 RAG 系统

ERP 系统最大的壁垒是隐性知识：企业特有的业务规则、客制化逻辑、历史配置原因。RAG 系统将这些知识向量化，让 Harness 具备企业特有的上下文。

# ERP 知识库 RAG 系统（向量检索增强）
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer

class ERPKnowledgeBase:
    """企业ERP知识库：操作手册 + 历史经验 + 配置文档"""

    def __init__(self):
        self.qdrant = QdrantClient("localhost", port=6333)
        self.encoder = SentenceTransformer("BAAI/bge-m3")  # 中文优化

    async def build_index(self, sources: list[ERPDocument]):
        """索引企业ERP文档"""
        for doc in sources:
            chunks = self._chunk_document(doc)
            vectors = self.encoder.encode([c.text for c in chunks])
            self.qdrant.upsert(
                collection_name = "erp_knowledge",
                points = [
                    PointStruct(
                        id      = chunk.id,
                        vector  = vec.tolist(),
                        payload = {
                            "text":     chunk.text,
                            "source":   doc.source,   # "SAP操作手册v3.2"
                            "module":   doc.module,   # "MM", "FI", "SD"...
                            "doc_type": doc.type,     # "procedure", "config", "bapi_doc"
                        }
                    )
                    for chunk, vec in zip(chunks, vectors)
                ]
            )

    async def retrieve(self, query: str, module: str = None) -> list[str]:
        """语义检索相关ERP知识"""
        vector = self.encoder.encode(query).tolist()
        filter_condition = (
            Filter(must=[FieldCondition(key="module", match=MatchValue(value=module))])
            if module else None
        )
        results = self.qdrant.search(
            collection_name = "erp_knowledge",
            query_vector    = vector,
            query_filter    = filter_condition,
            limit           = 5,
            score_threshold = 0.75,
        )
        return [r.payload["text"] for r in results]

# Harness 调用示例：
# tool: rag_retrieve(query="如何处理SAP中的负库存", module="MM")
# → 返回相关操作步骤 → 注入 Agent 上下文 → 生成精准回答

知识库建设范围

知识类型	来源	更新频率
ERP操作手册	官方文档 + 企业SOP	季度更新
BAPI/RFC文档	SE37导出 + 自动解析	随ERP升级
历史错误案例	工单系统 + 邮件记录	实时增量
配置说明文档	实施顾问输出	变更时更新
业务规则库	合规/财务部门文档	按需更新

---

六、安全管控框架与实施路线图

6.1 ERP 操作分级管控矩阵

操作类型	风险级别	Harness处理方式	示例
查询/报表	R1 只读	自动执行	ME23N查看采购单
主数据修改	R2 低风险	diff预览+确认	修改供应商地址
财务凭证	R3 中风险	双人审批	付款凭证过账
期末结账	R4 高风险	严格管控	月末关账操作
系统配置	R5 极高风险	禁止Agent直接操作	Basis系统参数

6.2 与 ERP 原有权限体系集成

Harness 权限 = min(AI角色权限, ERP账号权限)

用户A（SAP账号）: 有MM60查看权限，无MIGO过账权限
→ Harness 只能读取库存数据，无法创建入库单

四眼原则：付款指令必须由 initiator ≠ approver
→ Harness 自动检测，拒绝同一人发起+审批

6.3 审计追踪格式

{
  "trace_id": "hrn-2026-0819-001",
  "session_user": "zhang.san@company.com",
  "instruction": "处理采购订单PO-2026-0819",
  "executed_tools": [
    { "tool": "sap_read_po", "risk": "R1", "approval": "auto", "ts": "14:32:01" },
    { "tool": "sap_post_gr", "risk": "R2", "approval": "user_confirmed", "ts": "14:32:45" },
    { "tool": "sap_post_fi_doc", "risk": "R3", "approval": "dual_approved", "ts": "14:45:12" }
  ],
  "erp_changes": [
    "物料凭证: 5000012345",
    "会计凭证: 0100000234",
    "应付账款: AP-2026-08-0891"
  ],
  "rollback_point": "snap-hrn-2026-0819-001"
}

---

七、部署架构

┌─────────────────────────────────────────────────────────┐
│  企业内网（On-Premise）                                    │
│                                                          │
│  用户终端（浏览器/钉钉/VSCode）                              │
│       ↓ HTTPS                                            │
│  ┌──────────────────────────────┐                        │
│  │  Codex Harness 服务          │  ← 私有化部署            │
│  │  （K8s，3副本，高可用）         │     不上传数据到OpenAI   │
│  │  - Agent Loop Service        │                        │
│  │  - Approval Engine           │                        │
│  │  - Audit Logger (ClickHouse) │                        │
│  │  - Knowledge Base (Qdrant)   │                        │
│  └──────────────┬───────────────┘                        │
│                 ↓ MCP Protocol（内网）                     │
│  ┌──────────────┴───────────────┐                        │
│  │  MCP Server 集群              │                        │
│  │  - SAP MCP Server（RFC/BAPI） │                        │
│  │  - Oracle MCP Server         │                        │
│  │  - DB MCP Server（只读）       │                        │
│  └──────────────┬───────────────┘                        │
│                 ↓                                        │
│  ┌──────────────┴───────────────┐                        │
│  │  ERP 系统层                   │                        │
│  │  SAP（DEV/QAS/PRD）           │                        │
│  │  Oracle EBS                  │                        │
│  └──────────────────────────────┘                        │
│                                                          │
│  LLM 推理（私有化 or API）：企业级部署可用国内大模型            │
│  - 私有化：Qwen / DeepSeek / 百川（满足数据安全要求）          │
│  - OpenAI API：仅传输脱敏的指令和工具调用结构（不含业务数据）     │
└─────────────────────────────────────────────────────────┘

---

八、实施关键成功因素

1. MCP Server 质量是核心：ERP接口封装的完整性和稳定性决定 Agent 能力上限
2. 知识库建设先行：没有高质量的企业上下文，Agent 会产生幻觉。至少准备 500+ 文档
3. 审批流程不能省：财务类操作必须保留人工审批，这是合规红线，也是用户信任基础
4. 从只读场景起步：先上线查询/报表/分析（零风险），建立信任后逐步开放写操作
5. 变更管控集成：代码类变更必须走 ChaRM/DevOps 正式流程，Harness 负责生成，不负责绕过

---

2026-08-19 | 参考：openai/codex GitHub、Model Context Protocol官方文档、SAP BAPI文档