各位同仁，各位技术爱好者，大家好！
今天，我们齐聚一堂，探讨一个正在颠覆传统自动化领域的前沿话题——Agentic RPA。在数字化的浪潮中，我们已经见证了机器人流程自动化（RPA）所带来的巨大变革。它解放了人类双手，让枯燥重复的任务得以高效执行。然而，当自动化面对最棘手的挑战——那些形态各异、结构不一的非标表单时，传统RPA的局限性便暴露无遗。
今天，我将作为一名编程专家，带领大家深入解析Agentic RPA的核心概念，并将其与我们熟悉的传统RPA（以UiPath为例）进行对比，重点探讨基于LangGraph的“视觉驱动RPA”如何在这场对抗非标表单的战役中，展现出前所未有的优势。我们将通过严谨的逻辑、丰富的代码示例和实际的应用场景，揭示Agentic RPA的魔力。
1. 传统RPA的辉煌与局限：UiPath为例
首先，让我们回顾一下传统RPA的基石。以UiPath为例，它无疑是当前RPA市场上的领导者之一。UiPath的核心思想是通过模拟人类在图形用户界面（GUI）上的操作，实现业务流程自动化。
1.1. UiPath的工作原理简述
UiPath机器人通过一系列预定义的活动（Activities）来执行任务。这些活动通常包括：
- UI自动化： 点击按钮、输入文本、选择下拉菜单项等。它依赖于选择器（Selectors）——一种XPath或CSS选择器风格的字符串，用于唯一标识屏幕上的UI元素。
- 数据抓取： 从网页或桌面应用程序中提取结构化数据（如表格）。
- 图像与文本自动化： 当选择器不可用时，使用图像识别或光学字符识别（OCR）来识别和交互元素。
- 应用程序集成： 通过API或直接集成与特定应用程序交互。
- 流程控制： 条件判断（If/Else）、循环（Loop）、异常处理（Try/Catch）。
1.2. 传统RPA在处理非标表单时的挑战
传统RPA在处理高度结构化、重复性高、UI界面稳定的任务时表现卓越。然而，一旦我们遇到“非标表单”，例如：
- 来自不同供应商的发票，布局各异。
- 客户提交的纸质申请表，经过扫描后可能存在歪斜、字体不一。
- 金融机构的历史合同文档，格式多样。
- 医疗领域的患者病历，手写内容或不同系统的导出格式。
这时，传统RPA的局限性就显现出来了：
- 选择器的脆弱性： 非标表单意味着其UI布局、字段位置、甚至字段标签都可能发生变化。传统RPA强依赖于精确的选择器。一旦UI元素的位置、ID或属性发生微小变动，选择器就会失效，导致机器人崩溃。
- 缺乏语义理解： 传统RPA不理解“发票号码”的含义，它只知道要在某个特定的文本框中输入数据。如果“发票号码”的标签变为“凭证编号”或位于一个不同的位置，机器人无法自行适应。
- 硬编码逻辑的复杂性： 为了应对非标表单，开发人员不得不编写大量的条件判断（If/Else If），为每一种可能的表单变体创建一套独立的自动化逻辑。这导致了：
- 开发成本高昂： 每增加一种表单变体，都需要重新开发和测试。
- 维护困难： 任何微小的UI改动都可能触发连锁反应，需要大规模的调试和修改。
- 可扩展性差： 无法轻松应对未知或新的表单格式。
- OCR的局限性： 虽然UiPath集成了OCR功能，但它通常需要精确的区域定义。对于动态变化的非标表单，预定义OCR区域变得不切实际。即使能够提取文本，如何将这些文本映射到正确的业务字段，仍需大量硬编码逻辑。
- 缺乏推理和决策能力： 传统RPA本质上是“指令执行者”，它无法根据上下文进行推理、规划下一步行动或从错误中学习。
1.3. 传统RPA处理非标表单的伪代码示例（概念性）
假设我们要从一个简单的发票中提取发票号和总金额。
// UiPath Studio中的伪代码流程
// 步骤1: 打开发票应用程序或文件
OpenApplication("InvoiceViewer.exe")
AttachWindow("InvoiceViewer - Invoice 12345")
// 步骤2: 尝试通过选择器提取发票号
Try
invoiceNumber = GetText("Selector='<webctrl id="invoiceNumField" tag="INPUT" />'")
Catch SelectorNotFoundException
// 如果第一个选择器失效，尝试另一个选择器
Try
invoiceNumber = GetText("Selector='<webctrl aaname="Invoice Number" tag="SPAN" />/following-sibling::<webctrl tag="INPUT" />'")
Catch SelectorNotFoundException
// 尝试通过OCR在特定区域查找
invoiceNumber = OCR_ExtractText(Region(100, 150, 200, 30), "Invoice Number:")
End Try
End Try
// 步骤3: 尝试通过选择器提取总金额
Try
totalAmount = GetText("Selector='<webctrl id="totalAmountField" tag="SPAN" />'")
Catch SelectorNotFoundException
// 如果第一个选择器失效，尝试另一个选择器
Try
totalAmount = GetText("Selector='<webctrl aaname="Total Due" tag="SPAN" />/following-sibling::<webctrl tag="SPAN" />'")
Catch SelectorNotFoundException
// 尝试通过OCR在特定区域查找
totalAmount = OCR_ExtractText(Region(300, 400, 100, 50), "Total:")
End Try
End Try
// 步骤4: 记录或处理提取的数据
LogMessage("Invoice Number: " + invoiceNumber)
LogMessage("Total Amount: " + totalAmount)
// 步骤5: 关闭应用程序
CloseApplication("InvoiceViewer.exe")这段伪代码清晰地展示了传统RPA在面对非标表单时，需要大量的Try-Catch块和备用逻辑，这使得流程变得极其复杂且难以维护。
2. Agentic RPA：范式转变与核心理念
面对传统RPA在非结构化和半结构化数据处理上的瓶颈，Agentic RPA应运而生，它代表着自动化领域的一次深刻范式转变。
2.1. 什么是Agentic RPA？
Agentic RPA超越了简单的指令执行，它引入了“智能体（Agent）”的概念。一个Agentic RPA机器人不再仅仅是脚本的执行者，而是一个能够：
- 理解目标（Goal Understanding）： 明确任务的最终目的。
- 感知环境（Perception）： 能够“看”和“理解”屏幕上的内容（不仅仅是UI元素的属性）。
- 规划行动（Planning）： 根据目标和当前环境，动态地制定一系列操作步骤。
- 执行操作（Action Execution）： 模拟人类与GUI交互。
- 自我修正与学习（Self-Correction & Learning）： 在执行过程中遇到障碍时，能够进行推理，调整计划，甚至从错误中学习。
- 拥有记忆（Memory）： 记住之前的操作和提取的信息，以供后续决策使用。
其核心驱动力在于大型语言模型（LLMs）的强大推理能力和视觉模型（Vision Models）的图像理解能力。LLMs充当了机器人的“大脑”，赋予其语义理解、逻辑推理和生成自然语言指令的能力。
2.2. Agentic RPA的关键组成部分
-
感知模块 (Perception Module):
- 屏幕截图/视频流： 获取当前屏幕的视觉信息。
- OCR (Optical Character Recognition): 提取屏幕上的所有可见文本。
- UI元素识别 (UI Element Recognition): 识别按钮、文本框、链接等UI元素，并获取其位置、大小、可访问性属性。这可以结合传统的UI自动化技术（如Playwright/Selenium的DOM解析）和基于视觉的元素识别（如LayoutLMv3, Donut等模型）。
- 上下文理解 (Context Understanding): 不仅仅是识别元素，还要理解它们之间的关系和语义。
-
推理与规划模块 (Reasoning & Planning Module):
- 大型语言模型 (LLM): 作为核心，接收感知模块的输出和任务目标。
- 任务分解 (Task Decomposition): 将复杂任务分解为一系列更小的、可管理子任务。
- 行动选择 (Action Selection): 根据当前状态和目标，选择最合适的下一步操作（例如，点击一个按钮，输入文本，滚动页面，查找特定信息）。
- 意图理解 (Intent Understanding): 即使表单布局变化，也能理解用户或系统期望的意图。
-
行动执行模块 (Action Execution Module):
- UI交互工具： 例如Playwright、Selenium，用于模拟鼠标点击、键盘输入、页面导航等操作。
- API调用： 与后端系统或Web服务进行交互。
-
记忆模块 (Memory Module):
- 短期记忆： 存储当前会话的上下文，如已提取的数据、已执行的步骤、当前屏幕状态等。
- 长期记忆（可选）： 存储以往的学习经验、常用模式等，用于提高未来任务的效率。
-
反馈与学习模块 (Feedback & Learning Module):
- 结果验证： 检查执行结果是否符合预期。
- 错误处理： 当遇到异常时，LLM可以尝试诊断问题并规划恢复策略。
3. 基于LangGraph的“视觉驱动RPA”：对抗非标表单的利器
现在，让我们聚焦到如何利用LangGraph来构建一个强大的“视觉驱动RPA”代理，专门解决非标表单问题。
3.1. LangGraph简介
LangGraph是一个用于构建有状态、多actor（多代理）应用程序的Python库，它基于LangChain框架。它的核心思想是将复杂的LLM应用建模为一张有向无环图（DAG）或更一般的图（Graph）。图中的每个节点（Node）代表一个处理步骤（例如，调用LLM、执行工具、处理数据），而边（Edge）定义了数据流和控制流。
LangGraph的优势在于：
- 状态管理： 允许在节点之间传递和修改状态，这对于构建具有记忆和决策能力的代理至关重要。
- 条件路由： 能够根据节点的输出动态地决定下一个要执行的节点，实现复杂的逻辑分支。
- 循环与迭代： 支持在图中创建循环，使得代理能够进行迭代式的问题解决或错误恢复。
- 可观测性： 易于调试和理解代理的执行路径。
这些特性使得LangGraph成为构建Agentic RPA的理想选择，因为它能够模拟人类在处理复杂任务时的决策、规划和适应能力。
3.2. 视觉驱动RPA的架构与LangGraph映射
我们将构建一个Agentic RPA代理，它通过“看”（视觉感知）和“思考”（LLM推理）来处理非标表单。
| 模块类别 | 核心功能 | LangGraph中的对应 |
|---|---|---|
| 感知 (Perception) | 捕获屏幕，OCR文本，识别UI元素，理解布局 | perceive_screen_node |
| 推理 (Reasoning) | 分析感知结果，理解任务目标，生成下一步行动计划 | llm_decision_node |
| 行动 (Action) | 执行UI操作（点击、输入），调用API | execute_action_node |
| 记忆 (Memory) | 存储已提取数据，当前状态，历史操作 | AgentState (图的状态对象) |
| 验证 (Validation) | 检查提取数据是否符合规范，任务是否完成 | validate_extraction_node |
| 流程控制 | 根据LLM决策或验证结果，动态切换节点 | conditional_edges |
3.3. LangGraph实现非标表单处理的详细步骤
场景： 自动化处理不同格式的在线发票，提取“发票号码”、“总金额”和“开票日期”。
核心思想： 代理不会被硬编码去查找特定位置的元素，而是会“看”整个屏幕，理解屏幕上的文本和布局，然后“思考”在哪里可以找到所需的信息，并规划如何提取。
步骤概述：
-
定义工具 (Tools): 代理需要能够与外部世界交互的工具。
take_screenshot(): 捕获当前屏幕。perform_ocr(image_path): 对图像进行OCR，返回文本及其坐标。find_ui_element(text_label): 尝试通过文本标签在屏幕上定位UI元素。click_element(x, y): 在指定坐标点击。type_text(x, y, text): 在指定坐标输入文本。scroll_page(direction): 滚动页面。
-
定义代理状态 (Agent State): LangGraph通过一个状态对象在节点之间传递信息。
from typing import TypedDict, List, Dict, Any class AgentState(TypedDict): # 任务目标 goal: str # 当前屏幕的文本内容（OCR结果） screen_text: str # 当前屏幕的截图路径（可选，用于视觉模型） screenshot_path: str # 识别到的UI元素列表 (text, bounding_box, type) ui_elements: List[Dict[str, Any]] # 代理的决策（下一个动作） llm_action: Dict[str, Any] # 已提取的数据 extracted_data: Dict[str, str] # 历史记录/对话 chat_history: List[str] # 错误信息 error_message: str # 是否完成 is_finished: bool -
定义节点 (Nodes): 每个节点都是一个Python函数，接收当前状态并返回一个更新后的状态。
-
perceive_screen_node(state):- 捕获屏幕截图。
- 运行OCR获取屏幕文本及位置信息。
- （可选）使用视觉模型识别关键UI元素及其类型。
- 更新
screen_text和ui_elements。
-
llm_decision_node(state):- 将
goal、screen_text、ui_elements和extracted_data以及chat_history作为上下文，发送给LLM。 - LLM的职责是：
- 理解当前状态： “我在哪里？屏幕上有什么？我需要什么？”
- 规划下一步动作： “为了实现目标，我应该做什么？是查找某个字段，还是点击某个按钮？”
- 输出结构化的动作指令： 例如，
{"action": "find_and_extract", "field": "发票号码"}或{"action": "click", "element_text": "提交"}。
- 更新
llm_action。
- 将
-
execute_action_node(state):- 根据
llm_action中定义的动作，调用相应的工具函数。 - 如果动作是“查找并提取”，则尝试定位字段并提取数据，更新
extracted_data。 - 如果动作是“点击”，则执行点击。
- 如果动作失败，更新
error_message。
- 根据
-
validate_extraction_node(state):- 检查
extracted_data是否包含了所有必需的字段。 - 检查提取的数据格式是否正确（例如，日期格式）。
- 如果所有字段都已提取且有效，设置
is_finished = True。 - 如果发现问题，更新
error_message，并可能引导LLM重新规划。
- 检查
-
-
构建图 (Graph Construction): 使用LangGraph将节点和边连接起来。
from langgraph.graph import StateGraph, END # 假设我们已经定义了上述的 AgentState 和节点函数 builder = StateGraph(AgentState) # 添加节点 builder.add_node("perceive_screen", perceive_screen_node) builder.add_node("llm_decision", llm_decision_node) builder.add_node("execute_action", execute_action_node) builder.add_node("validate_extraction", validate_extraction_node) # 定义入口点 builder.set_entry_point("perceive_screen") # 定义边和条件逻辑 # 1. 感知屏幕后，总是进入LLM决策阶段 builder.add_edge("perceive_screen", "llm_decision") # 2. LLM决策后，执行动作 builder.add_edge("llm_decision", "execute_action") # 3. 执行动作后，根据结果进行条件路由 builder.add_conditional_edges( "execute_action", # 这是一个条件函数，它将根据execute_action_node的输出决定下一个节点 # 例如，如果LLM决定是“提取数据”，并且数据已提取，则进入验证 # 如果LLM决定是“导航/滚动”，则重新感知屏幕 # 如果LLM决定是“完成”，则直接结束 lambda state: "validate" if state.get("llm_action", {}).get("action") == "extract_data" else "perceive_screen" if state.get("llm_action", {}).get("action") in ["navigate", "scroll"] else "finish" if state.get("llm_action", {}).get("action") == "complete" else "llm_decision", # 如果有错误或需要重新思考 { "validate": "validate_extraction", "perceive_screen": "perceive_screen", "llm_decision": "llm_decision", # 重新思考 "finish": END } ) # 4. 验证后，如果任务完成，则结束；否则重新感知或让LLM重新规划 builder.add_conditional_edges( "validate_extraction", lambda state: "finish" if state.get("is_finished") else "llm_decision", { "finish": END, "llm_decision": "llm_decision" # 验证失败，让LLM重新规划 } ) # 编译图 app = builder.compile()
3.4. 视觉驱动RPA的代码示例（Python概念性实现）
这个示例将专注于展示LangGraph的结构和LLM在其中的作用，省略具体的UI自动化工具（如Playwright）和OCR库（如Tesseract）的详细实现，而是用模拟函数代替。
import os
from typing import TypedDict, List, Dict, Any, Union
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain_community.llms import OpenAI, FakeListLLM # 假设使用OpenAI，或用FakeListLLM模拟
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
# --- 1. 定义AgentState ---
class AgentState(TypedDict):
goal: str
screen_text: str
screenshot_path: str
ui_elements: List[Dict[str, Any]]
llm_action: Dict[str, Any] # LLM的决策，包含action_type和action_details
extracted_data: Dict[str, str]
chat_history: List[BaseMessage]
error_message: str
is_finished: bool
# --- 2. 定义工具 (模拟实现) ---
# 实际项目中，这些会调用Playwright, OCR库等
@tool
def take_screenshot() -> str:
"""Captures the current screen and returns the path to the screenshot image."""
print("[TOOL] Taking screenshot...")
# Simulate different invoice layouts for demonstration
invoice_layout = os.getenv("INVOICE_LAYOUT", "layout_A")
if invoice_layout == "layout_A":
# Simulate a screenshot path
return "screenshots/invoice_A.png"
elif invoice_layout == "layout_B":
return "screenshots/invoice_B.png"
else: # layout_C
return "screenshots/invoice_C.png"
@tool
def perform_ocr(image_path: str) -> List[Dict[str, Any]]:
"""Performs OCR on the given image path and returns a list of detected text with bounding boxes."""
print(f"[TOOL] Performing OCR on {image_path}...")
# Simulate OCR results for different layouts
if "invoice_A" in image_path:
return [
{"text": "Invoice Number:", "box": [50, 100, 150, 120]},
{"text": "INV-2023-001", "box": [160, 100, 250, 120]},
{"text": "Total Due:", "box": [50, 300, 120, 320]},
{"text": "$150.75", "box": [130, 300, 200, 320]},
{"text": "Date:", "box": [50, 150, 90, 170]},
{"text": "2023-10-26", "box": [100, 150, 180, 170]},
{"text": "Submit", "box": [700, 500, 750, 520]},
]
elif "invoice_B" in image_path:
return [
{"text": "Bill ID:", "box": [80, 120, 130, 140]},
{"text": "BL-XYZ-987", "box": [140, 120, 220, 140]},
{"text": "Amount:", "box": [80, 320, 140, 340]},
{"text": "234.99 USD", "box": [150, 320, 250, 340]},
{"text": "Billing Date:", "box": [80, 170, 150, 190]},
{"text": "10/25/2023", "box": [160, 170, 250, 190]},
{"text": "Process", "box": [650, 550, 720, 570]},
]
else: # invoice_C
return [
{"text": "Reference #:", "box": [60, 90, 140, 110]},
{"text": "REF-ABC-456", "box": [150, 90, 230, 110]},
{"text": "Grand Total:", "box": [60, 280, 150, 300]},
{"text": "€ 89.50", "box": [160, 280, 230, 300]},
{"text": "Doc Date:", "box": [60, 140, 120, 160]},
{"text": "26-Oct-2023", "box": [130, 140, 200, 160]},
{"text": "Confirm", "box": [720, 480, 780, 500]},
]
@tool
def find_ui_element_by_text(text_label: str) -> Dict[str, Any]:
"""
Finds a UI element on the screen based on its text label.
Returns its bounding box and assumed type (e.g., 'input', 'button').
"""
print(f"[TOOL] Finding UI element by text: '{text_label}'")
# This would involve complex visual parsing or DOM inspection in a real scenario
# For now, we simulate finding a button for 'Submit', 'Process', 'Confirm'
screen_text_data = perform_ocr(take_screenshot()) # Re-perceive for current text
for item in screen_text_data:
if text_label.lower() in item['text'].lower() and len(item['text']) < len(text_label) + 5: # fuzzy match
return {"text": item['text'], "box": item['box'], "type": "button"}
return {}
@tool
def click_element(x: int, y: int) -> str:
"""Clicks on the screen at the given coordinates (x, y)."""
print(f"[TOOL] Clicking at ({x}, {y})")
return "Click successful."
@tool
def type_text(x: int, y: int, text: str) -> str:
"""Types the given text at the given coordinates (x, y)."""
print(f"[TOOL] Typing '{text}' at ({x}, {y})")
return "Text typed successfully."
# --- 3. 定义LLM (使用FakeListLLM模拟，实际应为OpenAI/Anthropic等) ---
# LLM will decide the next action based on prompt and screen context.
# We need to simulate its output to be structured JSON.
# This is a VERY simplified simulation of LLM responses for the given task.
# In a real scenario, the LLM would be prompted to output a JSON object.
mock_llm_responses = [
"""
{
"action_type": "extract_data",
"field_name": "发票号码",
"keywords": ["Invoice Number:", "Bill ID:", "Reference #:"],
"extraction_strategy": "find_nearest_text_after_keyword"
}
""",
"""
{
"action_type": "extract_data",
"field_name": "总金额",
"keywords": ["Total Due:", "Amount:", "Grand Total:"],
"extraction_strategy": "find_nearest_text_after_keyword"
}
""",
"""
{
"action_type": "extract_data",
"field_name": "开票日期",
"keywords": ["Date:", "Billing Date:", "Doc Date:"],
"extraction_strategy": "find_nearest_text_after_keyword"
}
""",
"""
{
"action_type": "complete",
"message": "All required data extracted."
}
""",
"""
{
"action_type": "click_button",
"button_text": "Submit"
}
""",
"""
{
"action_type": "click_button",
"button_text": "Process"
}
""",
"""
{
"action_type": "click_button",
"button_text": "Confirm"
}
"""
]
# Using FakeListLLM to simulate LLM responses for demonstration
llm = FakeListLLM(responses=mock_llm_responses * 5) # Repeat responses to cover multiple steps
# We'll use a prompt template to guide the LLM to output JSON
llm_prompt = ChatPromptTemplate.from_messages([
("system",
"""You are an intelligent RPA agent designed to extract specific information from non-standard forms.
Your goal is: {goal}.
Current screen text (OCR results):
{screen_text}
Identified UI elements (text, box, type):
{ui_elements}
Already extracted data:
{extracted_data}
Based on the above, decide the best next action. Your response MUST be a JSON object with 'action_type' and 'action_details'.
Possible action_types:
- "extract_data": {"field_name": "...", "keywords": ["...", "..."], "extraction_strategy": "find_nearest_text_after_keyword"}
- "click_button": {"button_text": "..."}
- "type_text": {"target_field_text": "...", "value": "..."}
- "scroll": {"direction": "up" | "down"}
- "complete": {"message": "..."} (Use this when the goal is fully achieved)
- "replan": {"reason": "..."} (Use this if you are stuck or need to rethink)
"""),
("human", "{chat_history}")
])
# For structured output, we might use a dedicated parser
json_parser = JsonOutputParser()
# --- 4. 定义节点函数 ---
def perceive_screen_node(state: AgentState) -> AgentState:
print("n--- Node: Perceive Screen ---")
screenshot_path = take_screenshot.func()
ocr_results = perform_ocr.func(screenshot_path)
# Combine OCR results into a single string for LLM context
screen_text = "n".join([item['text'] for item in ocr_results])
# Simulate identifying UI elements (here, just passing OCR results as potential elements)
ui_elements = [{"text": item['text'], "box": item['box'], "type": "text_label"} for item in ocr_results]
# Add some common buttons if they exist
for btn_text in ["Submit", "Process", "Confirm"]:
btn_info = find_ui_element_by_text.func(btn_text)
if btn_info:
ui_elements.append({**btn_info, "type": "button"})
state["screenshot_path"] = screenshot_path
state["screen_text"] = screen_text
state["ui_elements"] = ui_elements
state["chat_history"].append(AIMessage(content="Screen perceived. Ready for decision."))
print(f"Screen text: {screen_text[:100]}...") # Print first 100 chars
return state
def llm_decision_node(state: AgentState) -> AgentState:
print("n--- Node: LLM Decision ---")
current_goal = state["goal"]
current_screen_text = state["screen_text"]
current_ui_elements = state["ui_elements"]
current_extracted_data = state["extracted_data"]
# Prepare prompt for LLM
prompt_value = llm_prompt.format_prompt(
goal=current_goal,
screen_text=current_screen_text,
ui_elements=current_ui_elements,
extracted_data=current_extracted_data,
chat_history=state["chat_history"]
)
# Get LLM response (simulated as JSON)
try:
raw_llm_response = llm.invoke(prompt_value)
# Assuming FakeListLLM gives string, we parse it
llm_action = json_parser.parse(raw_llm_response)
state["llm_action"] = llm_action
state["chat_history"].append(AIMessage(content=f"Decided action: {llm_action['action_type']}"))
print(f"LLM decided: {llm_action}")
except Exception as e:
state["error_message"] = f"LLM decision failed: {e}"
state["llm_action"] = {"action_type": "replan", "reason": f"LLM parsing failed: {e}"}
state["chat_history"].append(AIMessage(content=f"LLM decision failed: {e}"))
print(f"LLM decision failed: {e}")
return state
def execute_action_node(state: AgentState) -> AgentState:
print("n--- Node: Execute Action ---")
action = state["llm_action"]
extracted_data = state["extracted_data"]
screen_text_data = perform_ocr.func(state["screenshot_path"]) # Use full OCR data for extraction
if action["action_type"] == "extract_data":
field_name = action["field_name"]
keywords = action["keywords"]
extracted_value = None
for keyword in keywords:
for i, item in enumerate(screen_text_data):
if keyword.lower() in item['text'].lower():
# Simplified extraction: find the next text item after the keyword
if i + 1 < len(screen_text_data):
extracted_value = screen_text_data[i+1]['text']
print(f"Extracted '{extracted_value}' for '{field_name}' using keyword '{keyword}'")
break
if extracted_value:
break
if extracted_value:
extracted_data[field_name] = extracted_value
state["extracted_data"] = extracted_data
state["chat_history"].append(AIMessage(content=f"Extracted {field_name}: {extracted_value}"))
else:
state["error_message"] = f"Could not extract {field_name}."
state["chat_history"].append(AIMessage(content=f"Failed to extract {field_name}."))
state["llm_action"] = {"action_type": "replan", "reason": f"Failed to extract {field_name}"} # Force replan
print(f"Failed to extract {field_name}")
elif action["action_type"] == "click_button":
button_text = action["button_text"]
btn_info = find_ui_element_by_text.func(button_text)
if btn_info:
box = btn_info['box']
click_element.func(box[0] + (box[2]-box[0])//2, box[1] + (box[3]-box[1])//2)
state["chat_history"].append(AIMessage(content=f"Clicked button: {button_text}"))
else:
state["error_message"] = f"Button '{button_text}' not found."
state["chat_history"].append(AIMessage(content=f"Failed to find button: {button_text}"))
state["llm_action"] = {"action_type": "replan", "reason": f"Button '{button_text}' not found"}
print(f"Failed to find button: {button_text}")
elif action["action_type"] == "complete":
state["is_finished"] = True
state["chat_history"].append(AIMessage(content="Task completed as per LLM instruction."))
print("Task completed.")
elif action["action_type"] == "replan":
state["chat_history"].append(AIMessage(content=f"Re-planning due to: {action.get('reason', 'unspecified reason')}"))
print(f"Re-planning due to: {action.get('reason', 'unspecified reason')}")
# For other actions like 'type_text', 'scroll', implement similarly
return state
def validate_extraction_node(state: AgentState) -> AgentState:
print("n--- Node: Validate Extraction ---")
required_fields = ["发票号码", "总金额", "开票日期"]
extracted_data = state["extracted_data"]
all_extracted = True
for field in required_fields:
if field not in extracted_data or not extracted_data[field]:
all_extracted = False
state["error_message"] = f"Missing required field: {field}"
state["chat_history"].append(AIMessage(content=f"Validation failed: Missing {field}"))
print(f"Validation failed: Missing {field}")
break
if all_extracted:
state["is_finished"] = True
state["chat_history"].append(AIMessage(content="All required data validated successfully."))
print("All required data validated successfully.")
else:
# If validation fails, force the LLM to replan for missing fields
state["llm_action"] = {"action_type": "replan", "reason": state["error_message"]}
return state
# --- 5. 构建图 ---
workflow = StateGraph(AgentState)
workflow.add_node("perceive_screen", perceive_screen_node)
workflow.add_node("llm_decision", llm_decision_node)
workflow.add_node("execute_action", execute_action_node)
workflow.add_node("validate_extraction", validate_extraction_node)
workflow.set_entry_point("perceive_screen")
workflow.add_edge("perceive_screen", "llm_decision")
workflow.add_edge("llm_decision", "execute_action")
# Conditional routing after action execution
def route_after_action(state: AgentState) -> str:
action_type = state["llm_action"].get("action_type")
error_msg = state.get("error_message")
if state["is_finished"]:
return "end"
if error_msg and action_type == "replan": # If action failed and LLM decided to replan
return "llm_decision" # Go back to LLM to rethink based on error
if action_type == "extract_data":
return "validate_extraction"
if action_type == "click_button":
return "perceive_screen" # After clicking, screen state might change, so re-perceive
if action_type == "complete":
return "end"
return "llm_decision" # Default: go back to LLM to decide
workflow.add_conditional_edges(
"execute_action",
route_after_action,
{
"validate_extraction": "validate_extraction",
"perceive_screen": "perceive_screen",
"llm_decision": "llm_decision",
"end": END
}
)
# Conditional routing after validation
def route_after_validation(state: AgentState) -> str:
if state["is_finished"]:
return "end"
return "llm_decision" # If validation failed, go back to LLM to find missing fields
workflow.add_conditional_edges(
"validate_extraction",
route_after_validation,
{
"llm_decision": "llm_decision",
"end": END
}
)
app = workflow.compile()
# --- 6. 运行代理 ---
if __name__ == "__main__":
initial_state: AgentState = {
"goal": "Extract '发票号码', '总金额', '开票日期' from the invoice on screen.",
"screen_text": "",
"screenshot_path": "",
"ui_elements": [],
"llm_action": {},
"extracted_data": {},
"chat_history": [HumanMessage(content="Start processing invoice.")],
"error_message": "",
"is_finished": False
}
print("--- Running Agent for Layout A ---")
os.environ["INVOICE_LAYOUT"] = "layout_A"
for s in app.stream(initial_state.copy()):
print(s)
if "__end__" in s:
final_state = s["__end__"]
print(f"nFinal Extracted Data (Layout A): {final_state['extracted_data']}")
break
print("nn--- Running Agent for Layout B ---")
os.environ["INVOICE_LAYOUT"] = "layout_B"
# Reset initial state for a new run
initial_state["extracted_data"] = {}
initial_state["is_finished"] = False
initial_state["chat_history"] = [HumanMessage(content="Start processing invoice.")]
for s in app.stream(initial_state.copy()):
print(s)
if "__end__" in s:
final_state = s["__end__"]
print(f"nFinal Extracted Data (Layout B): {final_state['extracted_data']}")
break
print("nn--- Running Agent for Layout C ---")
os.environ["INVOICE_LAYOUT"] = "layout_C"
# Reset initial state for a new run
initial_state["extracted_data"] = {}
initial_state["is_finished"] = False
initial_state["chat_history"] = [HumanMessage(content="Start processing invoice.")]
for s in app.stream(initial_state.copy()):
print(s)
if "__end__" in s:
final_state = s["__end__"]
print(f"nFinal Extracted Data (Layout C): {final_state['extracted_data']}")
break
代码解释：
AgentState： 定义了代理在整个流程中需要维护的所有信息，包括目标、屏幕内容、LLM决策、已提取数据等。- 工具函数：
take_screenshot、perform_ocr、find_ui_element_by_text、click_element、type_text是模拟的工具，它们代表了代理与外部环境（屏幕、操作系统）交互的能力。在实际应用中，这些函数将封装对Playwright、Selenium、Tesseract、Google Vision API等库的调用。 llm和llm_prompt：FakeListLLM用于模拟LLM的响应，llm_prompt定义了如何构建发送给LLM的指令。关键在于，LLM被明确指示输出结构化的JSON，其中包含代理的下一步行动类型和详细信息。- 节点函数：
perceive_screen_node：负责感知当前屏幕状态，获取截图和OCR文本。llm_decision_node：接收感知结果，利用LLM进行推理，决定下一步行动（例如，“查找发票号码”或“点击提交按钮”），并将决策结果存储在llm_action中。execute_action_node：根据llm_action执行具体操作，如调用工具提取数据或模拟点击。validate_extraction_node：检查是否所有目标字段都已提取并符合预期。
- 图构建 (
workflow)： 使用StateGraph定义节点和它们之间的连接。add_node()：添加各个功能节点。set_entry_point()：定义流程的起始节点。add_edge()：定义无条件跳转。add_conditional_edges()：这是LangGraph的核心威力所在。 它允许我们根据state（特别是llm_action或is_finished）的当前值，动态地决定下一个执行的节点。例如，如果LLM决定提取数据，就进入验证节点；如果决定点击按钮，就重新感知屏幕；如果验证失败，则让LLM重新规划。
- 运行代理：
app.stream()方法允许我们逐步观察代理的状态变化，直到任务完成。通过设置INVOICE_LAYOUT环境变量，我们模拟了代理处理不同布局的非标发票的能力，而无需修改核心逻辑。
这个示例展示了Agentic RPA如何利用LLM的推理和LangGraph的流程控制，实现对非标表单的语义理解和动态适应。代理不再被硬编码的坐标或选择器所束缚，而是根据视觉感知和LLM的决策，自主地探索和提取所需信息。
4. 对比：传统RPA与Agentic RPA的优劣
让我们通过一个表格来直观地对比传统RPA和Agentic RPA在处理非标表单时的关键差异。
| 特性 | 传统RPA (如UiPath) | Agentic RPA (基于LangGraph的视觉驱动) |
|---|---|---|
| 核心机制 | 基于规则、硬编码选择器、固定工作流 | 基于LLM推理、视觉感知、动态规划、自主决策 |
| 适应性 | 对UI变化和非标布局极其脆弱，需大量重构 | 对UI变化和非标布局更具韧性，能通过推理适应新变体 |
| 语义理解 | 缺乏，仅识别UI元素属性或文本 | 具备，通过LLM理解“发票号码”的含义，而非仅仅其表面文本 |
| 维护成本 | 高，任何UI变动或新表单类型都需手动更新大量规则 | 相对较低，通过LLM的泛化能力处理更多变体，减少硬编码规则 |
| 开发复杂性 | 针对每种变体开发大量If/Else分支和备用选择器 | 需要设计合理的AgentState、工具和LLM提示词，但核心逻辑更通用 |
| 错误处理 | 基于预设异常处理，通常是重试或通知人类 | LLM可尝试理解错误原因并规划恢复策略，进行更智能的自我修正 |
| 可扩展性 | 差，难以应对未知或大量新表单类型 | 强，LLM能泛化到未曾见过的表单布局，自动化潜力更高 |
| 性能 | 高效执行固定流程 | LLM推理和视觉处理可能引入延迟，但鲁棒性优势显著 |
| 成本 | 部署和许可成本，开发/维护人力成本 | 除了RPA平台成本，还需考虑LLM API调用成本、视觉模型推理成本 |
| 技能要求 | RPA开发人员（熟悉特定平台） | 编程专家、LLM工程师、数据科学家（熟悉LangChain/LangGraph、LLM） |
5. Agentic RPA的优势与挑战
5.1. 优势
- 真正的鲁棒性： 不再被像素或选择器绑定，而是理解屏幕上的内容和上下文，大大减少了因UI变化导致的自动化中断。
- 高度的适应性与灵活性： 能够处理前所未见的非标表单布局，无需为每一种变体编写特定规则。这极大地扩展了RPA的应用范围。
- 智能决策与问题解决： LLM赋予代理在复杂场景下进行推理、规划和动态调整策略的能力，使其能够自主地应对意外情况。
- 降低维护成本： 随着代理变得更加智能和自适应，未来对自动化流程的维护工作量将显著减少。
- 更高的自动化率： 能够自动化那些传统RPA因其复杂性、多变性而无法触及的业务流程。
5.2. 挑战与考量
- LLM的成本与延迟： 大规模使用高级LLM API可能会带来较高的运行成本和一定的推理延迟，这在对实时性要求极高的场景中需要仔细权衡。
- LLM的幻觉与准确性： LLM可能生成不准确或不真实的响应（幻觉）。因此，需要设计强大的验证机制和人类干预点来确保数据准确性。
- 数据安全与隐私： 将敏感的屏幕信息或文档内容发送给外部LLM API进行处理，需要严格遵守数据隐私和安全规范。可以考虑部署私有化LLM或采用更安全的API网关。
- 代理设计的复杂性： 构建高效、可靠的Agentic RPA需要深厚的编程和LLM应用开发经验，以及对业务流程的深刻理解。
- 计算资源： 视觉模型和LLM在本地运行需要强大的计算资源（GPU），如果依赖云API则需要考虑网络带宽。
- 可解释性与调试： LLM的“黑盒”特性使得理解代理为何做出某个决策变得困难，调试也更具挑战性。
6. 展望未来
Agentic RPA，特别是基于LangGraph和视觉驱动的方法，正在将自动化带入一个全新的维度。它不仅仅是效率的提升，更是智能化的飞跃。未来，我们可能会看到：
- 混合型代理： 结合传统RPA的效率和Agentic RPA的智能，处理高度结构化和非结构化任务。
- 更小的、专业化的LLM： 针对特定领域和任务进行优化，降低成本和延迟。
- 更强大的多模态模型： 能够更深入地理解图像、文本、语音等多源信息。
- 更智能的自主学习： 代理不仅能执行任务，还能从每次执行中学习和优化自身的策略。
结束语
Agentic RPA，尤其是通过LangGraph构建的视觉驱动代理，为处理非标表单这一长期困扰传统自动化的难题，提供了颠覆性的解决方案。它将我们的自动化机器人从单纯的“操作员”提升为具备“感知、思考、决策”能力的智能“助理”。虽然挑战犹存，但技术进步的浪潮不可阻挡，我们正站在一个新时代的门槛上，见证自动化领域一次激动人心的变革。
感谢大家的时间！