Agentic ERP 集成：利用 LangGraph 驱动古老的 SAP/Oracle 系统完成自动化入库
各位技术同仁，下午好！
今天，我们将深入探讨一个既充满挑战又极具前景的话题：如何利用前沿的 Agentic AI 技术，特别是 LangGraph 框架，来改造我们企业中那些“古老”而又核心的 ERP 系统，例如 SAP 和 Oracle EBS，以实现高效率的自动化入库流程。
我们都知道，ERP 系统是企业运营的基石，它们承载着从采购、生产到销售、财务等几乎所有核心业务流程。然而，这些系统，尤其是那些运行了数十年之久的版本，往往以其复杂性、僵硬的流程和对人工操作的高度依赖而闻名。尤其在物流环节，例如货物入库（Goods Receipt），即使有标准化的事务代码或 API，其背后的决策、数据校验、异常处理以及与其他模块的联动，仍然需要大量的人工介入，导致效率低下、错误频发。
而另一边，生成式 AI 和大型语言模型（LLM）的兴起，为我们带来了全新的自动化范式。它不再仅仅是预设规则的自动化，而是能够理解人类意图、进行推理、规划行动并自我纠正的“智能自动化”。今天的讲座，我们就是要探讨如何将这两者结合起来，构建一个能够自主驱动 ERP 复杂流程的智能代理系统。
一、遗留 ERP 系统的挑战与 Agentic AI 的破局之道
1.1 传统 ERP 入库流程的痛点
让我们以典型的 SAP/Oracle 入库流程为例。一个标准的货物入库流程通常包括以下关键步骤：
- 采购订单（PO）确认： 确认是否存在有效的采购订单，以及待收货的物料、数量、供应商等信息。
- 货物到达与核对： 物理货物到达仓库，与送货单（Delivery Note）进行核对。
- 创建收货凭证（Goods Receipt Note – GRN）： 在 ERP 系统中执行收货操作（SAP 中的 MIGO 事务，Oracle EBS 中的 Receiving 模块），生成物料凭证。
- 质量检验（可选）： 根据物料属性或业务规则，决定是否需要进行质量检验（SAP 中的 QM 模块，Oracle 中的 Quality 模块）。
- 库存地点与批次管理： 确定物料存放的库存地点，如果是批次管理物料，则需要输入批次信息。
- 库存类型转换： 将物料从收货区（或质量检验库存）转移到自由使用库存。
- 异常处理： 如货物数量不符、质量问题、PO 不存在等。
这些步骤中，每一步都可能涉及人工决策、数据输入、跨模块操作以及对系统返回信息的理解。例如，一个简单的收货操作，可能需要用户记住特定的事务代码、字段含义，甚至在遇到错误时，需要根据错误消息去查找解决方案或联系相关部门。这无疑消耗了大量人力，并且容易因人为失误而产生数据不一致或业务中断。
1.2 ERP 系统的集成接口
尽管存在上述痛点，但现代 ERP 系统也提供了多种与外部系统交互的接口：
- SAP：
- BAPI (Business Application Programming Interface)： 标准化的业务接口，通过 RFC（Remote Function Call）调用。
- RFC (Remote Function Call)： 允许外部程序调用 SAP 系统内的函数模块。
- IDoc (Intermediate Document)： 用于异构系统间异步数据交换。
- OData / REST API： 较新的 SAP S/4HANA 系统提供了更现代的 RESTful API。
- GUI Scripting / RPA： 通过模拟用户界面操作进行自动化，适用于没有 API 或 API 难以覆盖的复杂场景，但通常较为脆弱。
- Oracle EBS：
- Open Interfaces： 预定义的数据接口表，通过插入数据触发业务逻辑。
- API (Application Programming Interface)： PL/SQL 包或 Java API。
- REST/SOAP Web Services： 现代 Oracle Fusion Cloud 和较新 EBS 版本提供。
- RPA： 同样适用于复杂或无 API 的场景。
这些接口是实现自动化的基础，但它们本身是“哑”的，不具备理解业务意图、规划执行路径、处理复杂逻辑的能力。它们仅仅是工具，需要外部智能进行编排。
1.3 Agentic AI：从自动化到自主化
Agentic AI（代理式人工智能）旨在构建能够感知环境、进行推理、制定计划、采取行动并具备自我纠正能力的智能实体。其核心思想是将 LLM 作为“大脑”，赋予其使用“工具”（Tools）的能力，以实现特定目标。
一个 Agentic 系统通常包含以下核心组件：
- 感知器（Perception）： 接收来自用户或环境的输入（例如，用户指令、系统返回信息）。
- 规划器（Planner）： 基于目标和当前状态，利用 LLM 的推理能力，将复杂任务分解为一系列可执行的子任务，并选择合适的工具。
- 执行器（Executor）： 调用相应的工具执行子任务（例如，调用 ERP API、执行 RPA 脚本）。
- 记忆（Memory）： 存储上下文信息、过往经验、知识库等，以支持长期规划和学习。
- 反思器（Reflector）： 评估行动结果，识别错误或改进机会，并调整后续计划。
LangGraph 正是为构建这种复杂、有状态的多代理系统而设计的框架。它允许我们以图（Graph）的形式定义代理的工作流，其中每个节点（Node）代表一个步骤或一个代理，边（Edge）定义了节点之间的转换逻辑。这种显式的图结构使得构建复杂、可控、可调试的代理系统成为可能，尤其适合于像 ERP 流程这样具有明确阶段和条件分支的业务场景。
二、LangGraph 核心概念解析
在深入实战之前，我们先来回顾一下 LangGraph 的核心概念。理解它们是构建高效 Agentic ERP 集成的关键。
2.1 图（Graph）与节点（Node）、边（Edge）
LangGraph 的核心是一个有向图。
- 节点 (Node)： 图的基本组成单元。每个节点可以是一个函数、一个 LLM 调用、一个工具调用，或者是一个更复杂的子代理。在我们的 ERP 场景中，一个节点可能代表：
- 解析用户意图
- 查询采购订单详情
- 执行 SAP MIGO 事务
- 进行质量检验决策
- 处理错误并向用户反馈
- 边 (Edge)： 连接节点的线，定义了控制流。边可以是：
- 直接边 (Direct Edge)： 从一个节点无条件地转移到另一个节点。
- 条件边 (Conditional Edge)： 基于某个条件（例如，上一个节点的输出或当前状态）来决定转移到哪个节点。这是实现复杂决策逻辑的关键。
- 状态 (State)： 整个图在执行过程中共享的数据结构。它在节点之间传递，每个节点都可以读取和修改状态。在我们的 ERP 场景中，状态可以包含：
- 用户输入的原始指令
- 解析出的采购订单号、物料、数量
- ERP 系统返回的物料凭证号
- 当前执行步骤
- 遇到的错误信息
- 与 LLM 的对话历史
2.2 代理（Agent）与工具（Tools）
- 代理 (Agent)： 在 LangGraph 中，通常是一个 LLM，它被赋予了推理和使用工具的能力。代理会根据当前状态和目标，决定下一步应该调用哪个工具或执行哪个内部逻辑。
- 工具 (Tools)： 代理可以调用的外部函数或 API。在我们的 ERP 场景中，工具是与 SAP/Oracle 系统交互的桥梁，例如：
sap_get_po_details(po_number: str)sap_create_goods_receipt(data: dict)oracle_check_material_quality(material_id: str)rpa_login_and_navigate_to_tcode(tcode: str)
2.3 内存（Memory）与反思（Reflection）
- 记忆 (Memory)： LangGraph 通过
State机制天然地支持短时记忆（当前会话上下文）。对于长时记忆，可以集成向量数据库或其他知识库，让代理能够检索历史数据、文档、配置信息等。 - 反思 (Reflection)： 这是 Agentic AI 的一个高级特性。代理能够审查自己的行动和结果，识别错误，并根据这些洞察调整未来的行为。在 LangGraph 中，这通常通过一个专门的“反思节点”实现，该节点利用 LLM 来分析当前状态和结果，并决定下一步是重试、修正计划还是向用户寻求帮助。
三、Agentic ERP 集成架构设计
现在，让我们来构想一个基于 LangGraph 的 Agentic ERP 入库自动化系统的整体架构。
3.1 总体架构视图
+---------------------+ +----------------------------+
| User Interface | | LangGraph Orchestrator |
| (Chatbot, Web App) | | (Agentic Core) |
+----------+----------+ +-------------+--------------+
| User Intent | Agent State, Action Plan
| | Execution Results
v v
+-------------------------------------------------------------+
| ERP Integration Layer (Tools) |
| +---------------------+ +---------------------+ +---------------------+
| | SAP Connector | | Oracle Connector | | RPA Bridge |
| | (pyrfc, pysap, OData) | | (SQL, REST, SOAP) | | (UiPath, Power Auto) |
| +----------+----------+ +----------+----------+ +----------+----------+
| | | |
+------------+------------------------------+------------------------------+
| | |
v v v
+---------------------+ +---------------------+ +---------------------+
| SAP System | | Oracle EBS/Fusion | | External Systems |
| (ECC, S/4HANA) | | | | (WMS, QM, TMS) |
+---------------------+ +---------------------+ +---------------------+3.2 关键组件详解
-
用户界面 (User Interface)：
- 可以是聊天机器人（Slack, Teams, 自定义Web Chat），让业务用户以自然语言描述入库需求。
- 也可以是一个简单的 Web 表单，用户输入关键信息后，由 LangGraph Agent 填充和验证其余数据并执行。
- Prompt Engineering： 设计清晰的提示语，引导用户提供必要信息，并明确代理的任务边界。
-
LangGraph Orchestrator (Agentic Core)：
- 主代理 (Master Agent)： 负责接收用户意图，将其分解为子任务，并协调各个子节点或专业代理的执行。它通常是一个 LLM 调用，通过工具调用来驱动流程。
- 状态管理 (State Management)： 维护整个工作流的上下文，包括用户输入、中间结果、错误信息、对话历史等。
- 节点与边 (Nodes & Edges)： 定义了整个入库流程的逻辑流、决策点和异常处理路径。
-
ERP Integration Layer (Tools)： 这是 LangGraph Agent 与实际 ERP 系统交互的“手脚”。
- SAP Connector：
- Python
pyrfc库： 用于调用 SAP 的 RFC/BAPI。这是最常见且强大的集成方式。 pysap或自定义 OData/REST 客户端： 用于与 S/4HANA 或 Fiori 应用交互。- 封装为 Python 函数： 每个 BAPI/RFC 调用都被封装成一个 LangGraph Agent 可以调用的工具函数，带有清晰的输入参数和输出结构。
- Python
- Oracle Connector：
- Python DB API (cx_Oracle)： 对于直接数据库接口或 Open Interfaces。
requests库： 用于调用 Oracle REST/SOAP Web Services。- 封装为 Python 函数： 同 SAP Connector。
- RPA Bridge：
- 当没有可用的 API，或者 API 无法满足复杂业务逻辑时，RPA 机器人可以作为工具被 LangGraph Agent 调用。
- 例如，一个
rpa_perform_migo_with_screenshot(tcode, data)工具，可以触发一个 UiPath 机器人去执行 SAP GUI 操作，并在完成后返回截图或日志。
- SAP Connector：
-
知识库/记忆 (Knowledge Base/Memory)：
- ERP 配置数据： 物料主数据、工厂、库存地点、供应商、批次属性等。
- 业务规则： 例如，“特定物料必须进行质检”、“特定供应商的收货数量允许有 5% 的误差”。
- 历史操作记录： 成功和失败的入库案例，用于代理学习和优化。
- 错误码/解决方案： 当 ERP 返回错误时，代理可以查询知识库以提供更智能的解决方案建议。
- 这部分可以通过向量数据库 (Vector Database) 结合 RAG (Retrieval Augmented Generation) 技术实现。
3.3 安全性与合规性考量
- 凭证管理： 绝不能将 ERP 系统凭证直接暴露给 LLM。应使用安全的凭证管理系统（如 Vault、KMS），并通过代理集成层进行认证和授权。
- 最小权限原则： ERP 用户或服务账户应只拥有执行特定自动化任务所需的最小权限。
- 审计日志： 所有由 Agent 执行的 ERP 操作都必须有详细的审计日志，记录谁、何时、执行了什么操作，以及操作结果。
- 人机协作 (Human-in-the-Loop – HITL)： 对于高风险或异常情况，系统应能够暂停自动化流程，请求人工审核和批准。
四、实战演练：自动化 SAP MIGO (收货) 流程
现在，让我们通过一个具体的例子，来展示如何使用 LangGraph 驱动 SAP MIGO（收货）流程。我们将聚焦于一个简化的场景：根据采购订单号和物料详情，执行货物收货。
4.1 场景设定
用户输入： "帮我完成采购订单 4500000001 的收货，物料 MAT001 数量 100 个，存放到 0001 库存地点。"
目标：Agent 能够解析用户意图，查询 PO 详情，调用 SAP BAPI BAPI_GOODSMVT_CREATE 创建收货凭证，并返回结果。
4.2 环境准备
- Python 3.9+
langchain,langgraph,langchain_openai(或其他 LLM 客户端)pyrfc：SAP Python Connector- 一个可用的 SAP ECC 或 S/4HANA 系统，并配置好
pyrfc连接参数。
# 安装必要的库
# pip install langchain langgraph langchain_openai pyrfc4.3 定义 ERP 交互工具
我们将封装与 SAP 交互的函数作为 LangGraph 的 Tools。
import os
from typing import Dict, Any, List, Optional
from langchain_core.tools import tool
from pyrfc import Connection # 假设已配置好SAP连接参数
# 假设的SAP连接配置
SAP_CONNECTION_PARAMS = {
"ashost": os.getenv("SAP_ASHOST", "your.sap.host"),
"sysnr": os.getenv("SAP_SYSNR", "00"),
"client": os.getenv("SAP_CLIENT", "100"),
"user": os.getenv("SAP_USER", "your_sap_user"),
"passwd": os.getenv("SAP_PASSWD", "your_sap_password"),
"lang": os.getenv("SAP_LANG", "EN")
}
# --- SAP 连接管理 ---
def get_sap_connection():
"""建立并返回一个SAP连接"""
try:
conn = Connection(**SAP_CONNECTION_PARAMS)
return conn
except Exception as e:
print(f"Error connecting to SAP: {e}")
raise
# --- 工具函数定义 ---
@tool
def sap_get_po_details(po_number: str) -> Dict[str, Any]:
"""
根据采购订单号获取采购订单的详细信息，包括物料、数量、工厂等。
返回示例:
{
"success": True,
"po_number": "4500000001",
"items": [
{"po_item": "00010", "material": "MAT001", "quantity": "200.000", "unit": "PC", "plant": "1000", "storage_loc": "0001", "open_quantity": "200.000"},
{"po_item": "00020", "material": "MAT002", "quantity": "50.000", "unit": "PC", "plant": "1000", "storage_loc": "0001", "open_quantity": "50.000"}
]
}
"""
print(f"Tool: Calling sap_get_po_details for PO {po_number}")
try:
conn = get_sap_connection()
# 调用BAPI_PO_GETDETAIL 或 BAPI_PO_GETITEMS
# 这里我们模拟一个简化的BAPI调用结果
# 实际场景中需要调用真实的BAPI并解析其复杂的结构
result = conn.call('BAPI_PO_GETDETAIL',
PURCHASEORDER=po_number,
ITEMS='X',
ITEM_TEXT='X')
if result and 'PO_ITEMS' in result:
items = []
for item in result['PO_ITEMS']:
# 假设我们只关心一些基本字段
items.append({
"po_item": item['PO_ITEM'],
"material": item['MATERIAL'],
"quantity": item['QUANTITY'],
"unit": item['PO_UNIT'],
"plant": item['PLANT'],
"storage_loc": item['STGE_LOC'],
"open_quantity": item['OPEN_QTY'] # 这是一个简化，实际需计算
})
return {"success": True, "po_number": po_number, "items": items}
else:
return {"success": False, "po_number": po_number, "error": "PO not found or no items."}
except Exception as e:
return {"success": False, "po_number": po_number, "error": str(e)}
@tool
def sap_create_goods_receipt(
po_number: str,
po_item: str,
material: str,
quantity: float,
unit: str,
plant: str,
storage_loc: str,
document_date: Optional[str] = None, # YYYYMMDD
posting_date: Optional[str] = None, # YYYYMMDD
) -> Dict[str, Any]:
"""
在SAP中执行MIGO收货操作，调用BAPI_GOODSMVT_CREATE。
参数:
- po_number: 采购订单号
- po_item: 采购订单行项目号
- material: 物料号
- quantity: 收货数量
- unit: 计量单位
- plant: 工厂
- storage_loc: 库存地点
- document_date: 凭证日期 (YYYYMMDD, 默认为今天)
- posting_date: 过账日期 (YYYYMMDD, 默认为今天)
返回示例:
{"success": True, "material_document": "5000000001", "doc_year": "2023"}
"""
print(f"Tool: Calling sap_create_goods_receipt for PO {po_number}, Material {material}, Qty {quantity}")
try:
conn = get_sap_connection()
# 获取当前日期
import datetime
today = datetime.datetime.now().strftime("%Y%m%d")
document_date = document_date if document_date else today
posting_date = posting_date if posting_date else today
goodsmvt_header = {
'PSTNG_DATE': posting_date,
'DOC_DATE': document_date,
'REF_DOC_NO': po_number, # 通常参考采购订单号
'GM_CODE': '01', # 01 for Goods Receipt for Purchase Order
'UNAME': SAP_CONNECTION_PARAMS['user'] # 操作用户
}
goodsmvt_item = {
'MATERIAL': material,
'PLANT': plant,
'STGE_LOC': storage_loc,
'MOVE_TYPE': '101', # 101 for Goods Receipt for Purchase Order
'ENTRY_QTY': str(quantity), # 注意：BAPI通常需要字符串类型
'ENTRY_UOM': unit,
'PO_NUMBER': po_number,
'PO_ITEM': po_item,
'MVT_IND': 'B', # B for Purchase Order
'GR_QTY': str(quantity) # 收货数量
}
# 调用 BAPI
result = conn.call('BAPI_GOODSMVT_CREATE',
GOODSMVT_HEADER=goodsmvt_header,
GOODSMVT_ITEM=[goodsmvt_item],
GOODSMVT_CODE={'GM_CODE': '01'})
# 检查返回消息
if result and 'GOODSMVT_HEADRET' in result and 'MAT_DOC' in result['GOODSMVT_HEADRET']:
material_document = result['GOODSMVT_HEADRET']['MAT_DOC']
doc_year = result['GOODSMVT_HEADRET']['DOC_YEAR']
# BAPI_TRANSACTION_COMMIT 必须显式调用以提交更改
conn.call('BAPI_TRANSACTION_COMMIT')
return {"success": True, "material_document": material_document, "doc_year": doc_year}
else:
# 解析BAPI返回的RETURN表获取错误信息
error_messages = []
if 'RETURN' in result and isinstance(result['RETURN'], list):
for msg in result['RETURN']:
if msg['TYPE'] in ['E', 'A']: # Error or Abort
error_messages.append(f"{msg['ID']}-{msg['NUMBER']}: {msg['MESSAGE']}")
return {"success": False, "error": "SAP BAPI call failed.", "details": error_messages if error_messages else result}
except Exception as e:
return {"success": False, "error": str(e)}
# 注册工具
tools = [sap_get_po_details, sap_create_goods_receipt]4.4 定义 LangGraph 状态
状态是节点之间传递的共享数据。
from typing import TypedDict, Annotated, List, Dict, Any, Literal, Optional
import operator
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
"""
表示Agent执行流程的当前状态。
"""
messages: Annotated[List[BaseMessage], operator.add] # 对话历史
user_intent: str # 用户原始指令
parsed_po_number: Optional[str] # 从用户指令中解析出的PO号
parsed_material: Optional[str] # 从用户指令中解析出的物料号
parsed_quantity: Optional[float] # 从用户指令中解析出的数量
parsed_storage_loc: Optional[str] # 从用户指令中解析出的库存地点
po_details: Optional[Dict[str, Any]] # sap_get_po_details 工具返回的PO详情
gr_result: Optional[Dict[str, Any]] # sap_create_goods_receipt 工具返回的结果
error_message: Optional[str] # 记录当前步骤的错误信息
current_action_plan: List[str] # Agent的行动计划
current_step_status: Literal["PLANNING", "FETCHING_PO", "VALIDATING_DATA", "PERFORMING_GR", "REFLECTING", "COMPLETED", "FAILED"]4.5 构建 LangGraph 节点
我们将定义几个关键节点来构建入库流程。
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langgraph.graph import StateGraph, END
# 初始化LLM
# 确保你已经设置了 OPENAI_API_KEY 环境变量
llm = ChatOpenAI(model="gpt-4-0125-preview", temperature=0)
# --- 定义LLM代理 ---
# 这个代理将负责规划和选择工具
system_prompt = """
你是一个高级的企业资源计划（ERP）自动化助手，专注于处理SAP和Oracle系统的入库流程。
你的任务是根据用户的指令，理解其意图，规划一系列步骤，并利用提供的工具与ERP系统交互，最终完成入库操作。
你有以下工具可以使用:
{tools}
根据当前的状态和用户意图，你需要决定下一步做什么。
你的输出应该是一个JSON对象，包含以下键：
1. 'action': 你要执行的下一步动作，可以是 'call_tool' 或 'finish' 或 'replan'。
2. 'tool_name': 如果 'action' 是 'call_tool'，这里是你要调用的工具名称。
3. 'tool_args': 如果 'action' 是 'call_tool'，这里是传递给工具的参数字典。
4. 'response': 如果 'action' 是 'finish'，这里是给用户的最终响应。
5. 'plan': 如果 'action' 是 'replan'，这里是你的新的行动计划。
6. 'reasoning': 解释你为什么选择这个动作。
如果需要调用工具，你必须严格按照工具的签名和参数要求来构建 tool_args。
如果完成了任务，使用 'finish' 动作。
如果发现当前计划无法继续或遇到错误，使用 'replan' 动作，并说明原因。
当前状态: {state}
"""
prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
("placeholder", "{messages}"),
]
).partial(tools="n".join([f"{tool.name}: {tool.description}" for tool in tools]))
# 创建一个LLM工具调用链
agent_runnable = prompt | llm.bind_tools(tools) | JsonOutputParser()
def call_tool_node(state: AgentState):
"""
根据Agent的决策调用相应的工具。
"""
messages = state['messages']
last_message = messages[-1]
if not isinstance(last_message, dict) or 'action' not in last_message:
return {"messages": [("assistant", "Error: Agent did not produce a valid action.")], "current_step_status": "FAILED"}
action_data = last_message
action_type = action_data['action']
reasoning = action_data.get('reasoning', 'No specific reasoning provided.')
if action_type == 'call_tool':
tool_name = action_data.get('tool_name')
tool_args = action_data.get('tool_args', {})
print(f"Agent decided to call tool: {tool_name} with args: {tool_args}")
print(f"Reasoning: {reasoning}")
try:
# 找到并执行工具
chosen_tool = next(t for t in tools if t.name == tool_name)
tool_output = chosen_tool.invoke(tool_args)
return {"messages": [("tool", tool_output)], "current_step_status": "TOOL_EXECUTED"}
except Exception as e:
return {"messages": [("tool", {"error": str(e)})], "current_step_status": "FAILED", "error_message": str(e)}
elif action_type == 'finish':
return {"messages": [("assistant", action_data.get('response', 'Task completed.'))], "current_step_status": "COMPLETED"}
elif action_type == 'replan':
return {"messages": [("assistant", f"Re-planning: {action_data.get('plan', 'No specific plan provided.')}")], "current_step_status": "PLANNING"}
else:
return {"messages": [("assistant", "Error: Unknown action type from agent.")], "current_step_status": "FAILED"}
# --- 节点定义 ---
def agent_node(state: AgentState):
"""
LLM代理节点，负责解析用户意图、规划行动或选择工具。
"""
# 将当前的AgentState转换为LLM可以理解的格式
# 移除不可序列化的部分或对LLM不重要的部分
state_for_llm = state.copy()
state_for_llm['messages'] = [m.content for m in state_for_llm['messages']] if state_for_llm['messages'] else []
# 确保 messages 总是列表
current_messages = state['messages'] if state['messages'] else []
# 模拟LLM思考和输出
# 在实际LangGraph中，这里会直接调用LLM
llm_output = agent_runnable.invoke({
"state": state_for_llm,
"messages": current_messages
})
# LLM的输出是JSON，我们需要将其转换为BaseMessage类型才能添加到messages历史中
return {"messages": [("assistant", llm_output)], "current_step_status": "PLANNING"}
def parse_and_validate_node(state: AgentState) -> AgentState:
"""
解析用户意图，从消息中提取PO号、物料、数量等信息，并进行初步验证。
"""
user_message = state['user_intent'] # 假设user_intent在初始状态中已设置
po_number = None
material = None
quantity = None
storage_loc = None
error_message = None
# 简单的正则表达式或NLP解析
import re
po_match = re.search(r'POs*(d+)', user_message, re.IGNORECASE)
if po_match:
po_number = po_match.group(1)
else:
error_message = "无法识别采购订单号。请提供正确的PO号。"
material_match = re.search(r'物料s*(w+)', user_message, re.IGNORECASE)
if material_match:
material = material_match.group(1)
else:
error_message = (error_message + "n" if error_message else "") + "无法识别物料号。"
quantity_match = re.search(r'数量s*(d+(.d+)?)s*(个|PC|EA)', user_message, re.IGNORECASE)
if quantity_match:
quantity = float(quantity_match.group(1))
else:
error_message = (error_message + "n" if error_message else "") + "无法识别数量。"
storage_loc_match = re.search(r'库存地点s*(w+)', user_message, re.IGNORECASE)
if storage_loc_match:
storage_loc = storage_loc_match.group(1)
else:
storage_loc = "0001" # 假设默认库存地点
# 进一步的业务逻辑验证，例如检查数量是否为正数
if quantity is not None and quantity <= 0:
error_message = (error_message + "n" if error_message else "") + "收货数量必须大于0。"
if error_message:
return {
"error_message": error_message,
"current_step_status": "FAILED",
"messages": [("assistant", f"解析用户指令失败：{error_message}")]
}
else:
return {
"parsed_po_number": po_number,
"parsed_material": material,
"parsed_quantity": quantity,
"parsed_storage_loc": storage_loc,
"current_step_status": "VALIDATING_DATA",
"messages": [("assistant", f"已成功解析用户意图：PO号 {po_number}, 物料 {material}, 数量 {quantity}, 库存地点 {storage_loc}. 准备查询PO详情。")]
}
def reflection_node(state: AgentState):
"""
反思节点：评估当前流程状态，尤其是在工具调用失败或任务完成后。
决定是结束、重试、重新规划还是请求人工干预。
"""
messages = state['messages']
last_message = messages[-1]
# 假设tool output是最后一个message
if last_message.type == "tool":
tool_output = last_message.content
if isinstance(tool_output, dict) and not tool_output.get("success"):
error_details = tool_output.get("error", "未知错误")
# LLM可以根据错误信息进行更智能的反思
reflection_text = f"工具调用失败，错误信息：{error_details}。我应该如何处理？是重试，还是通知用户，或者调整计划？"
return {
"messages": [("assistant", reflection_text)],
"error_message": error_details,
"current_step_status": "REFLECTING"
}
elif isinstance(tool_output, dict) and tool_output.get("success") and "material_document" in tool_output:
response = f"入库成功！物料凭证号：{tool_output['material_document']}，年度：{tool_output['doc_year']}。"
return {
"messages": [("assistant", response)],
"gr_result": tool_output,
"current_step_status": "COMPLETED"
}
elif isinstance(tool_output, dict) and tool_output.get("success") and "po_details" in state:
response = f"已成功获取PO {state['parsed_po_number']} 的详情。现在准备执行收货。"
return {
"messages": [("assistant", response)],
"current_step_status": "REFLECTING"
}
# 其他情况，比如agent_node的输出
if state["current_step_status"] == "COMPLETED":
return state # 任务已完成，无需进一步反思
# 默认情况下，如果一切顺利，继续到下一个代理决策
return {"current_step_status": "REFLECTING"} # 标记为反思中，等待agent_node决定下一步4.6 构建 LangGraph 图
现在，我们将这些节点连接起来，形成一个完整的自动化工作流。
from langgraph.graph import StateGraph, END
# 定义图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("parse_validate", parse_and_validate_node) # 初始解析和验证
workflow.add_node("agent", agent_node) # LLM Agent，负责规划和工具选择
workflow.add_node("call_tool", call_tool_node) # 执行工具调用的节点
workflow.add_node("reflect", reflection_node) # 反思和错误处理节点
# 设置入口点
workflow.set_entry_point("parse_validate")
# 定义边
# 1. 解析验证后，如果成功，交给 Agent 规划；如果失败，则结束（或者可以设计一个错误处理节点）
workflow.add_conditional_edges(
"parse_validate",
lambda state: "agent" if not state.get("error_message") else END,
{"agent": "agent", END: END}
)
# 2. Agent 节点后的决策
# 如果 Agent 决定调用工具，则转到 'call_tool' 节点
# 如果 Agent 决定结束，则转到 'END'
# 如果 Agent 决定重新规划，则转回 'agent' (这里简化为直接再次调用agent_node)
def agent_decision(state: AgentState):
last_message = state['messages'][-1]
if isinstance(last_message, dict) and last_message.get('action') == 'call_tool':
return "call_tool"
elif isinstance(last_message, dict) and last_message.get('action') == 'finish':
return "end_task"
elif isinstance(last_message, dict) and last_message.get('action') == 'replan':
return "agent" # Re-plan
else:
# Fallback for unexpected agent output, can be improved
return "reflect"
workflow.add_conditional_edges("agent", agent_decision, {
"call_tool": "call_tool",
"end_task": END,
"agent": "agent",
"reflect": "reflect"
})
# 3. 工具调用后，转到 'reflect' 节点进行结果评估
workflow.add_edge("call_tool", "reflect")
# 4. 反思节点后的决策
def reflection_decision(state: AgentState):
if state["current_step_status"] == "COMPLETED":
return END
elif state["current_step_status"] == "FAILED":
return END # 暂时以失败结束，实际可重试或通知人
else:
# 反思后，通常会再次交给 Agent 决定下一步
return "agent"
workflow.add_conditional_edges("reflect", reflection_decision, {
END: END,
"agent": "agent"
})
# 编译图
app = workflow.compile()4.7 运行 Agent
# 初始输入
user_input_1 = "帮我完成采购订单 4500000001 的收货，物料 MAT001 数量 100 个，存放到 0001 库存地点。"
user_input_2 = "收货PO 4500000002 的 MAT002 20个，库存地点 0002." # 另一个例子
user_input_3 = "错误测试：收货PO 4500000001 的 MAT001 数量 -10 个。" # 错误输入
initial_state = AgentState(
messages=[],
user_intent=user_input_1,
parsed_po_number=None,
parsed_material=None,
parsed_quantity=None,
parsed_storage_loc=None,
po_details=None,
gr_result=None,
error_message=None,
current_action_plan=[],
current_step_status="PLANNING"
)
print("--- Running Agent for User Input 1 ---")
for s in app.stream(initial_state):
print(s)
print("---")
# 假设用户输入后，agent_node会解析并生成一个包含action的字典，
# 然后call_tool_node会执行该action。
# 这是一个简化的示例，实际运行时，agent_node的LLM调用会根据prompt和tools生成action。
# 为了让这个示例在没有真实LLM调用时也能运行，我们可能需要模拟LLM的输出。
# 模拟LLM输出的逻辑，以在没有OpenAI Key时也能演示流程
# 假设 LLM 在 agent_node 中会输出类似这样的 JSON
# {
# "action": "call_tool",
# "tool_name": "sap_get_po_details",
# "tool_args": {"po_number": "4500000001"},
# "reasoning": "用户要求收货，我需要先获取PO详情来验证物料和数量。"
# }
# {
# "action": "call_tool",
# "tool_name": "sap_create_goods_receipt",
# "tool_args": {
# "po_number": "4500000001",
# "po_item": "00010", # 假设已从po_details获取
# "material": "MAT001",
# "quantity": 100.0,
# "unit": "PC",
# "plant": "1000", # 假设已从po_details获取
# "storage_loc": "0001"
# },
# "reasoning": "已获取PO详情并验证数据，现在执行收货。"
# }
# {
# "action": "finish",
# "response": "入库操作已成功完成，物料凭证号：...",
# "reasoning": "所有步骤已完成。"
# }
# 由于篇幅和复杂性，这里无法完全模拟整个LLM决策过程并使其在无LLM情况下运行。
# 但上述代码结构展示了LangGraph如何组织节点和工具，
# 实际运行时，'agent_node' 会通过 'agent_runnable' 调用真正的LLM来生成这些决策。4.8 流程概览（表格形式）
| 步骤编号 | 节点名称 | 节点类型/功能 | 输入（来自 State） | 输出（更新 State） | 决策/转换条件 |
|---|---|---|---|---|---|
| 1 | parse_validate |
解析用户意图，提取关键信息，初步校验 | user_intent |
parsed_po_number, parsed_material, parsed_quantity, parsed_storage_loc, error_message |
成功: 转至 agent 节点 失败 (有 error_message): 转至 END (或专门的错误处理) |
| 2 | agent |
LLM Agent，根据状态和工具决定下一步行动 | messages, parsed_..., po_details, gr_result, error_message |
messages (包含 LLM 决策: action, tool_name, tool_args 或 response) |
LLM 决策 ‘call_tool’: 转至 call_tool 节点 LLM 决策 ‘finish’: 转至 END LLM 决策 ‘replan’: 转至 agent 节点 (重新规划) 其他: 转至 reflect 节点 (处理意外情况) |
| 3 | call_tool |
执行由 agent 节点选择的工具 |
messages (包含 LLM 的工具调用决策) |
messages (包含工具执行结果), po_details (如果调用 sap_get_po_details), gr_result (如果调用 sap_create_goods_receipt), error_message |
工具执行完毕: 转至 reflect 节点 |
| 4 | reflect |
反思工具执行结果或当前流程状态 | messages, gr_result, error_message |
messages (包含反思结果或用户反馈), current_step_status |
任务完成 (current_step_status == "COMPLETED"): 转至 END 任务失败 ( current_step_status == "FAILED"): 转至 END (或重新规划) 其他 (需进一步决策): 转至 agent 节点 |
| (结束) | END |
流程结束 | – | – | – |
五、高级场景与增强功能
上述示例展示了一个基础的入库流程，但 Agentic ERP 集成的潜力远不止于此。
5.1 多代理协作
在更复杂的场景中，单一代理可能难以处理所有事务。我们可以构建多个专业代理，每个代理负责一个特定的业务领域或流程阶段：
- 采购代理 (PO Agent)： 负责采购订单的创建、修改、查询。
- 收货代理 (GR Agent)： 负责 MIGO/Receiving 操作。
- 质检代理 (QM Agent)： 负责创建质检批、记录检验结果、做使用决定。
- 库存管理代理 (IM Agent)： 负责库存转移、盘点、库存查询。
- 异常处理代理 (Exception Agent)： 专门负责识别和处理业务流程中的异常情况，例如数量差异、质量问题、系统错误等。
这些代理可以在 LangGraph 中作为独立的节点或子图存在，并由一个主协调代理 (Orchestrator Agent) 进行调度和通信。
5.2 人机协作 (Human-in-the-Loop, HITL)
对于高风险操作、模糊不清的指令或无法自动解决的异常，引入人工干预至关重要。
- 批准节点： 在执行关键 ERP 写入操作前，Agent 可以暂停流程，将相关信息发送给指定用户进行审核和批准。
- 澄清节点： 当 Agent 无法完全理解用户意图或所需信息不完整时，它可以向用户提问以获取更多上下文。
- 异常上报： 当 Agent 遇到其无法处理的复杂错误时，可以自动创建工单或发送通知给支持团队。
5.3 知识检索增强生成 (RAG)
将企业的内部知识库（ERP 配置手册、常见问题解答、SOP 文档、历史错误日志）向量化，并通过 RAG 技术集成到 Agent 中。
- 当 Agent 需要了解某个物料的特殊处理规则时，它可以从知识库中检索相关文档。
- 当 Agent 遇到 SAP 错误码时，它可以查询知识库以获取详细的错误描述和建议解决方案。
- 这极大地增强了 Agent 的推理能力和解决问题的自主性，减少了“幻觉”的可能性。
5.4 智能错误处理与自愈
LLM 的推理能力使其能够超越简单的错误代码匹配。
- 语义化错误分析： Agent 可以理解 ERP 返回的错误消息的含义，并尝试根据上下文进行智能纠正（例如，自动调整数量单位、尝试不同的库存地点）。
- 重试策略： 对于瞬时性错误（如网络问题、系统负载），Agent 可以实施带指数退避的重试机制。
- 回滚或补偿： 在某些情况下，如果一个事务失败，Agent 可以尝试执行回滚操作或启动一个补偿流程来撤销已完成的部分。
5.5 安全性与性能优化
- API Gateway / Middleware： 在 LangGraph Agent 与 ERP 系统之间增加一个 API Gateway 或中间件层，以集中管理认证、授权、请求限流、审计和数据转换。
- 数据缓存： 对于不经常变动的 ERP 主数据（如物料描述、工厂信息），可以进行缓存，减少对 ERP 系统的频繁调用。
- 异步处理： 对于耗时较长的 ERP 操作，可以设计异步工具，并在 LangGraph 中使用回调或状态轮询来处理。
六、挑战与未来展望
Agentic ERP 集成无疑为企业带来了巨大的机遇，但同时也伴随着挑战。
6.1 当前挑战
- LLM 幻觉 (Hallucination)： LLM 有时会生成不准确或虚假的信息，这在操作关键业务系统时是不可接受的。需要通过严格的工具设计、数据验证、RAG 增强和 HITL 来缓解。
- 遗留系统的复杂性： 深入理解 SAP/Oracle 复杂的业务逻辑、数据模型和配置仍然是构建有效工具的关键。LLM 无法替代对业务专家的依赖。
- 性能与成本： LLM 推理的延迟和成本可能成为实时、高并发场景的瓶颈。选择合适的模型、优化提示、使用缓存和批处理是必要的。
- 数据安全与隐私： 如何安全地处理敏感的企业数据，确保 LLM 不会泄露或误用信息，是至关重要的。
- 信任与采纳： 业务用户和 IT 团队需要时间来建立对 Agentic 系统的信任，理解其能力边界。
- 可解释性： 当 Agent 做出某个决策时，如何提供清晰的解释，对于审计和故障排除至关重要。
6.2 未来展望
尽管存在挑战，Agentic ERP 集成的未来一片光明：
- 更智能的自动化： 代理将能够处理更复杂、更模糊的业务场景，减少人工干预。
- 自适应流程： 代理将能够根据实时数据和业务变化，动态调整其工作流程，实现真正的柔性自动化。
- 低代码/无代码 Agent 构建： 出现更多用户友好的平台，让业务分析师也能参与到 Agent 的设计和训练中。
- 强化学习与持续优化： 代理可以通过与 ERP 系统的持续交互和反馈，不断学习和优化其决策和行动策略。
- 事件驱动的 Agent： 代理不再是被动等待指令，而是能够主动监控 ERP 事件（例如，新的 PO 创建、库存水平变化），并触发相应的自动化流程。
结论
Agentic ERP 集成，通过 LangGraph 这样的框架赋能，正将我们从简单的任务自动化推向自主、智能的业务流程管理。它不再仅仅是连接系统，而是赋予系统理解、推理、规划和执行复杂业务逻辑的能力。通过将 LLM 的智能与 SAP/Oracle 等核心 ERP 系统的稳定性和数据完整性结合，企业将能够解锁前所未有的运营效率、显著降低人工错误，并使宝贵的人力资源聚焦于更具战略价值的创新工作。这是一个激动人心的时代，智能代理正在重塑企业自动化的未来。