这次尝试的宗旨是

AI可替换，不和特定AI绑定

不考虑Token消耗

设计出通用的 AI 驱动框架，并不局限于 RPA 这个方面

1. 浅尝

最开始的想法是实现两个Agent分别是·、任务规划Agent和任务执行Agent

任务规划Agent根据用户需求列出步骤清单，任务执行Agent负责循环步骤清单执行步骤

1.1 任务规划Agent

我在实现完任务规划Agent后就发现，任务规划和执行分离不是一个好设计，规划的任务准确性会越来越低，所以这个方案我就没有继续实现执行Agent

rask_planner_agent.py

import base64
from langchain_deepseek import ChatDeepSeek
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from .output_schema import TaskSchema


class TaskPlannerAgent:
    def __init__(self):
        self.llm = ChatOpenAI(
            openai_api_key="sk-or-v1-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx,
            openai_api_base="https://openrouter.ai/api/v1",
            model_name="google/gemini-3-flash-preview"
        )

        # System prompt
        self.system_prompt = """
        你是一个专业的任务规划助手。用户将提供任务描述或图片，你需要结合这些信息将其拆解为结构化步骤。
        """

        self.agent = create_agent(
            model=self.llm,
            system_prompt=self.system_prompt,
            response_format=TaskSchema
        )

    def plan(self, instruction: str, image_path: str = None) -> TaskSchema:
        content = [{"type": "text", "text": instruction}]

        if image_path:
            image_data = self._encode_image(image_path)
            content.append({
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{image_data}"
                }
            })
        result = self.agent.invoke(
            {"messages": [{"role": "user", "content": content}]}
        )
        return result["structured_response"]

    def _encode_image(self, image_path: str):
        """将本地图片转为 base64编码"""
        with open(image_path, 'rb') as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')


def main(args):
    agent = TaskPlannerAgent()
    for step in agent.plan("搜索熊猫，并进入搜索结果的第一项内容", r"C:\Users\Levit\Desktop\1.jpg").steps:
        print(step)

output_schema.py 规范Agent输出格式

from typing import List, Optional, Literal, Union
from pydantic import BaseModel, Field


class Point(BaseModel):
    x: int = Field(..., description="目标终点横坐标")
    y: int = Field(..., description="目标终点纵坐标")


class Click(BaseModel):
    type: Literal['click']
    point: Point = Field(..., description="鼠标点击的位置")
    button: Literal["left", "middle", "right"] = Field(
        ...,
        description="鼠标按键，可选值为 left（左键）、middle（中键）、right（右键）"
    )
    clicks: int = Field(..., description="表示按下次数")


class Input(BaseModel):
    point: Point = Field(..., description="输入框所在位置")
    value: str = Field(..., description="需要输入的文本内容")


class Step(BaseModel):
    id: int = Field(..., description="步骤的唯一序号")
    description: str = Field(..., description="描述该步骤的具体意图")
    action: Union[Click, Input] = Field(..., description="具体操作类型")


class TaskSchema(BaseModel):
    """任务执行计划的根结构"""
    goal: str = Field(..., description="任务的最终目标描述")
    steps: List[Step] = Field(..., description="拆解后的执行步骤列表")

1.2 结论

这个方案有个问题，规划和后续遇到的问题可能会不一样。

目前的逻辑是规划 → 执行

实际优秀的逻辑是：感知 → 决策 → 行动 → 再感知

2. 改进模型

这个方案，每轮 Agent 都会感知当前界面并做出下一步的决策，同时 Agent 会记住之前自己做过的决策。

2.1 感知决策Agent

这个 Agent 负责感知并做出决策，同时Agent包含了简单的消息管理，Agent能知道之前自己做过什么操作。

agent.py

import base64
from langchain_deepseek import ChatDeepSeek
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from .output_schema import StepSchema


class AgentMemory:
    def __init__(self):
        self.steps = []
    
    def add_step(self, step: StepSchema):
        self.steps.append(step.model_dump_json())
    
    def summary(self) -> str:
        if not self.steps:
            return "尚未执行任何操作"
        return '\n'.join(self.steps)


class DecisionAgent:
    def __init__(self, purpose):
        self.llm = ChatOpenAI(
            openai_api_key="sk-or-v1-xxxxxxxxxxxxxxxxxxxxxxxxxxx",
            openai_api_base="https://openrouter.ai/api/v1",
            model_name="google/gemini-3-flash-preview"
        )

        # System prompt
        self.system_prompt = f"""
        你是一个桌面RPA的决策Agent，你需要根据客户目的与当前电脑页面分析下一步需要执行的动作并以结构化的方式输出下一步动作。
        客户的目标是：{purpose}
        请参考之前的操作历史，不要重复已经失败的动作。
        """

        self.memory = AgentMemory()

        self.agent = create_agent(
            model=self.llm,
            system_prompt=self.system_prompt,
            response_format=StepSchema
        )

    def next_step(self, image_path) -> StepSchema:
        content = []
        content.append({
            "type": "text",
            "text": f"你之前已经执行过的步骤如下:\n{self.memory.summary()}"
        })

        image_data = self._encode_image(image_path)
        content.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{image_data}"
            }
        })
        result = self.agent.invoke(
            {"messages": [{"role": "user", "content": content}]}
        )
        self.memory.add_step(result["structured_response"])
        return result["structured_response"]

    def _encode_image(self, image_path: str):
        """将本地图片转为 base64编码"""
        with open(image_path, 'rb') as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')

output_schema.py

from typing import List, Optional, Literal, Union, Any
from pydantic import BaseModel, Field


class Point(BaseModel):
    x: int = Field(..., description="目标终点横坐标")
    y: int = Field(..., description="目标终点纵坐标")


class Click(BaseModel):
    type: Literal["click"]
    point: Point = Field(..., description="鼠标点击的位置")
    button: Literal["left", "middle", "right"] = Field(
        ...,
        description="鼠标按键，可选值为 left（左键）、middle（中键）、right（右键）"
    )
    clicks: int = Field(..., description="表示按下次数")


class Input(BaseModel):
    type: Literal["input"]
    point: Point = Field(..., description="输入框所在位置")
    value: str = Field(..., description="需要输入的文本内容")


class Finish(BaseModel):
    type: Literal["finish"]
    output: dict[str, Any] | None = Field(..., description="当任务需要返回内容时返回结果放这里")


class StepSchema(BaseModel):
    thought: str = Field(..., description="表示当前需要做的事情")
    action: Union[Click, Input, Finish] = Field(..., description="具体操作类型")

2.2 执行器

指令负责执行的模块已经不带Agent了，因为执行模块不用带脑子，只需要根据决策Agent的结果执行对应动作。

executor.py

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv

from .output_schema import StepSchema

class Executor:
    def __init__(self):
        # 截图的宽高
        self.width = 1920
        self.height = 1080
    
    def execute(self, step: StepSchema):
        action = step.action
        if action.type == "finish":
            return
        
        x, y = self.get_pixel_coords(action.point.x, action.point.y)
        if action.type == "click":
            xbot.win32.mouse_move(x, y)
            xbot.win32.mouse_click()
        elif action.type == "input":
            point = action.point
            xbot.win32.mouse_move(x, y)
            xbot.win32.mouse_click()
            xbot.win32.send_keys("^{a}{BACKSPACE}")
            xbot.win32.send_keys(action.value)
    
    def get_pixel_coords(self, normalized_x, normalized_y):
        pixel_x = (normalized_x / 1000) * self.width
        pixel_y = (normalized_y / 1000) * self.height
        return int(pixel_x), int(pixel_y)

2.3 效果

我这里是把代码丢在影刀里面运行的，所以会有一些影刀的Python包

main.py

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv

from agent import DecisionAgent
from excutor import Excutor

def main(args):
    agent = DecisionAgent("打开浏览器，下载当前页面的订单数据，需要将文件的名字重命名为“测试.xlsx”")
    excutor = Executor()
    while True:
        xbot.win32.screenshot.save_screen_to_file(r"C:\Users\Levit\Desktop\测试.jpg", "jpg")
        step = agent.next_step(r"C:\Users\Levit\Desktop\测试.jpg")
        print(step)
        if step.action.type == "finish":
            break
        excutor.execute(step)
        sleep(2)

2.4 结论

这个方案存在一下几个问题

Decision Agent 同时承担了「感知 + 决策」：短期 OK，长期会出三个问题：prompt 越来越长、agent 开始“看不清重点”、无法复用 perception 能力
没有「结果校验 / 失败感知」：需要识别当前操作是否执行正确并且在 memory 中记录出每次操作是否正确

将感知和决策分离的优势是，可以换任意的感知，换上不同的感知配合上不同的执行器就是不同的Agent

3. 进一步改进模型

根据 2 中的结论，进一步优化的 Agent 模型如下

PerceptionAgent：负责感知世界，相当于人眼，使用有多模态能力的AI模型

DecisionAgent：负责做决策，相当于人脑，使用推理类的大语言模型

Executor：负责执行操作，不需要接入AI模型，纯代码实现

ResultObserverAgent：负责校验实际运行结果与预期运行结果是否相符，可以对操作分层，一部分操作可机械判断结果，一部分操作可通过多模态的AI模型判断结果

3.1. 感知 Agent

perception_agent.py

import base64
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from .perception_output_schema import ViewSchema


class PerceptionAgent:
    def __init__(self):
        self.llm = ChatOpenAI(
            openai_api_key="sk-or-v1-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
            openai_api_base="https://openrouter.ai/api/v1",
            model_name="google/gemini-3-flash-preview"
        )

        self.system_prompt = """
        你是一个 Perception Agent，负责感知当前电脑屏幕的状态，
        返回图片中所有可以操作的元素

        你的任务是：
        - 描述当前屏幕的可见内容
        - 如果提供了“期望结果”，判断当前画面是否满足该期望

        你智能基于当前的截图判断

        如果无法判断是否满足期望，请返回“不确定”
        """

        self.agent = create_agent(
            model=self.llm,
            system_prompt=self.system_prompt,
            response_format=ViewSchema
        )

    def view(self, image_path: str, expectation: str=None) -> ViewSchema:
        content = []
        if expectation:
            content.append({
                "type": "text",
                "text": expectation
            })

        image_data = self._encode_image(image_path)
        content.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{image_data}"
            }
        })
        result = self.agent.invoke(
            {"messages": [{"role": "user", "content": content}]}
        )
        return result["structured_response"]

    def _encode_image(self, image_path: str):
        """将本地图片转为 base64编码"""
        with open(image_path, 'rb') as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')

perception_agent_schema.py

from typing import List, Optional, Literal, Union, Any
from pydantic import BaseModel, Field


class Position(BaseModel):
    x: int = Field(..., description="横坐标，范围 0-1000，左上角为原点")
    y: int = Field(..., description="纵坐标，范围 0-1000，左上角为原点")


class Element(BaseModel):
    label: str = Field(..., description="元素名称")
    focus: Optional[bool] = Field(..., description="元素是否获取到输入焦点；无法判断则为 null")
    value:Optional[str] = Field(..., description="输入框或文本元素当前显示的内容，如无法识别则为 null")
    elem_type: Literal["button", "input", "text", "icon", "link"] = Field(..., description="元素的类型")
    position: Position = Field(..., description="元素的坐标")


class ViewSchema(BaseModel):
    summary: str = Field(
        ..., description="一句话描述当前界面状态"
    )
    elements: List[Element] = Field(
        ..., description="当前界面中可以操作的元素"
    )
    notes: str = Field(
        ..., description="任何你认为对决策有帮助的客观信息"
    )

3.2 决策 Agent

decision_agent.py

from langchain_openai import ChatOpenAI
from langchain_deepseek import ChatDeepSeek
from langchain.agents import create_agent
from .perception_output_schema import ViewSchema
from .decision_output_schema import StepSchema
from .result_observer_output_schema import ObservationResultSchema


class AgentMemory:
    def __init__(self):
        self.steps = []
    
    def add_step(self, step: StepSchema):
        self.steps.append(step.model_dump_json())

    def summary(self) -> str:
        if not self.steps:
            return "尚未执行任何操作"
        return '\n'.join(self.steps)


class DecisionAgent:
    def __init__(self, purpose):
        self.llm = ChatDeepSeek(
            model='deepseek-chat',
            api_key='sk-xxxxxxxxxxxxxxxx',
            temperature=0.7
        )
        
        self.system_prompt = f"""
        你是一个桌面自动化系统中的 Decision Agent

        你的职责是：
        - 根据任务目标、与当前界面状态、历史执行记录以及上一步执行结果决定下一步要执行的动作。
        - 明确该动作执行后“期望出现的结果”和“结果观察策略”。

        注意：
        - 只输出 JSON 格式，不要包含任何前导或后续文本

        前任务目标是：{purpose}
        """

        self.memory = AgentMemory()

        self.agent = create_agent(
            model=self.llm,
            system_prompt=self.system_prompt,
            response_format=StepSchema
        )

    def step(self, view: ViewSchema, observer: ObservationResultSchema = None):
        content = []
        content.append({
            "type": "text",
            "text": f"当前页面感知结果：\n{view.model_dump_json()}"
        })
        content.append({
            "type": "text",
            "text": f"你之前已经执行过的步骤如下：\n{self.memory.summary()}"
        })
        if observer:
            content.append({
                "type": "text",
                "text": f"上一步执行后的验证结果：\n{observer.model_dump_json()}"
            })
        result = self.agent.invoke(
            {"messages": [{"role": "user", "content": content}]}
        )
        
        step = result["structured_response"]
        self.memory.add_step(step)
        return step

decision_agent.py

from typing import List, Optional, Literal, Union, Any
from pydantic import BaseModel, Field


class Position(BaseModel):
    x: int = Field(..., description="横坐标，范围 0-1000，左上角为原点")
    y: int = Field(..., description="纵坐标，范围 0-1000，左上角为原点")


class Element(BaseModel):
    label: str = Field(..., description="元素名称")
    focus: Optional[bool] = Field(..., description="元素是否获取到输入焦点；无法判断则为 null")
    value:Optional[str] = Field(..., description="输入框或文本元素当前显示的内容，如无法识别则为 null")
    elem_type: Literal["button", "input", "text", "icon", "link"] = Field(..., description="元素的类型")
    position: Position = Field(..., description="元素的坐标")


class ViewSchema(BaseModel):
    summary: str = Field(
        ..., description="一句话描述当前界面状态"
    )
    elements: List[Element] = Field(
        ..., description="当前界面中可以操作的元素"
    )
    notes: str = Field(
        ..., description="任何你认为对决策有帮助的客观信息"
    )

3.3 执行器

executor.py

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv

from .decision_output_schema import StepSchema


class Executor:
    def __init__(self):
        # 截图的宽高
        self.width = 1920
        self.height = 1080
    
    def execute(self, step: StepSchema):
        action = step.action
        if action.type == "finish":
            return
        
        x, y = self.get_pixel_coords(action.point.x, action.point.y)
        if action.type == "click":
            xbot.win32.mouse_move(x, y)
            xbot.win32.mouse_click()
        elif action.type == "input":
            point = action.point
            xbot.win32.mouse_move(x, y)
            xbot.win32.mouse_click()
            xbot.win32.send_keys("^{a}{BACKSPACE}")
            xbot.win32.send_keys(action.value)
    
    def get_pixel_coords(self, normalized_x, normalized_y):
        pixel_x = (normalized_x / 1000) * self.width
        pixel_y = (normalized_y / 1000) * self.height
        return int(pixel_x), int(pixel_y)

3.4 结果检测Agent

result_observer_agent.py

import base64
from langchain_openai import ChatOpenAI
from langchain_deepseek import ChatDeepSeek
from langchain.agents import create_agent
from .perception_output_schema import ViewSchema
from .decision_output_schema import StepSchema
from .result_observer_output_schema import ObservationResultSchema


class ResultObserverAgent:
    def __init__(self):
        self.llm = ChatDeepSeek(
            model='deepseek-chat',
            api_key='sk-xxxxxxxxxxxxxxxxxxxxxxxxxx',
            temperature=0.7
        )
        
        self.system_prompt = """
        你是一个左面自动化系统中的 ResultObserver Agent。

        你的职责是：
        - 对比 “动作执行后的期望” 与 “执行动作后的界面感知结果”
        - 判断期望是否被满足
        - 提供一个清晰、客观、的观察结论
        """

        self.agent = create_agent(
            model=self.llm,
            system_prompt=self.system_prompt,
            response_format=ObservationResultSchema
        )
    
    def observe(self, view: ViewSchema, step: StepSchema) -> ObservationResultSchema:
        content = []
        content.append({
            "type": "text",
            "text": f"当前界面感知结果：\n{view.model_dump_json(indent=2)}"
        })
        content.append({
            "type": "text",
            "text": f"动作执行后的预期：{step.expectation}"
        })
        result = self.agent.invoke(
            {"messages": [{"role": "user", "content": content}]}
        )
        return result["structured_response"]

result_observer_output_schema.py

from typing import List, Optional, Literal, Union, Any
from pydantic import BaseModel, Field


class ObservationResultSchema(BaseModel):
    matches_expectation: Literal["yes", "no", "uncertain"] = Field(
        ..., description="当前界面是否满足预期"
    )
    reason: str = Field(
        ..., description="解释为何满足或不满足预期"
    )

3.5 效果

main.py

from .perception_agent import PerceptionAgent
from .decision_agent import DecisionAgent
from .result_observer_agent import ResultObserverAgent
from .executor import Executor


def main(args):
    sleep(5)
    job = "使用浏览器，下载当前页面的订单数据，并将文件命名为“测试.xlsx"

    perception_agent = PerceptionAgent()
    decision_agent = DecisionAgent(job)
    observer_agent = ResultObserverAgent()
    executor = Executor()

    count = 1
    pre_step = None
    while True:
        xbot.win32.screenshot.save_screen_to_file(r"C:\Users\Levit\Desktop\测试.jpg", "jpg")
        view = perception_agent.view(r"C:\Users\Levit\Desktop\测试.jpg")
        print(f"世界：{view.model_dump_json()}")
        if count != 1:
            observer = observer_agent.observe(view, pre_step)
            print(f"验证：{observer.model_dump_json()}")
            step = decision_agent.step(view, observer)
        else:
            step = decision_agent.step(view)
        print(f"决策：{step.model_dump_json()}")

        if step.action.type == "finish":
            break

        executor.execute(step)
        pre_step = step
        sleep(3)
        count += 1

4. 更进一步优化模型

3中的模型运行已经满足我的要求，这个模型我就没有实际实现，理论上肯定这个模型更优秀。

增加 Tracker Agent Perception Agent
增加一个反馈，让Perception Agent在下一次观察世界时候应该着重注意什么

1. 浅尝#

1.1 任务规划Agent#

1.2 结论#

2. 改进模型#

2.1 感知决策Agent#

2.2 执行器#

2.3 效果#

2.4 结论#

3. 进一步改进模型#

3.1. 感知 Agent#

3.2 决策 Agent#

3.3 执行器#

3.4 结果检测Agent#

3.5 效果#

4. 更进一步优化模型#