Agent实战入门：从零构建自动操作软件的核心架构

1次阅读

共计 2188 个字符，预计需要花费 6 分钟才能阅读完成。

最近在做一个自动处理 Excel 报表的项目时，我深刻体会到了手动操作的痛苦：每天要重复点击几十次相同的按钮，还要处理各种突发弹窗。这种重复性工作不仅效率低下，还容易出错。于是我开始研究自动化操作软件的实现方案，发现 Agent 架构能很好地解决这类问题。

传统自动化脚本的痛点很明显：

缺乏任务状态管理，脚本中断后无法恢复
异常处理机制薄弱，遇到弹窗就崩溃
难以扩展复杂业务流程

调研了几种主流方案后，我发现：

AutoGPT：适合复杂决策场景，但资源消耗大
LangChain：擅长处理自然语言交互
自定义 Agent：轻量灵活，适合基础自动化场景

对于操作 Excel 这种确定性任务，我最终选择自建轻量级 Agent，核心需求是：

可靠的任务队列
完善的错误恢复
可观测的运行状态

用 Python 实现的基础框架如下：

import asyncio
from typing import List, Callable

class AutomationAgent:
    def __init__(self, max_retries=3):
        """:param max_retries: 最大重试次数"""
        self.task_queue = asyncio.Queue()
        self.max_retries = max_retries
        self._is_running = False

    async def add_task(self, task_func: Callable, *args):
        """添加任务到队列"""
        await self.task_queue.put((task_func, args))

    async def _execute_task(self, task_func, args):
        """带重试机制的原子任务执行"""
        for attempt in range(self.max_retries):
            try:
                return await task_func(*args)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)  # 指数退避

    async def run(self):
        """启动 Agent 主循环"""
        self._is_running = True
        while self._is_running:
            task_func, args = await self.task_queue.get()
            try:
                await self._execute_task(task_func, args)
            except Exception as e:
                print(f"Task failed: {e}")
            finally:
                self.task_queue.task_done()

自动化操作必须考虑幂等性（idempotent）。比如点击按钮操作，可能会因为网络延迟被重复执行。我们的解决方案：

def make_idempotent(func):
    """装饰器确保函数只执行一次"""
    executed = False

    async def wrapper(*args, **kwargs):
        nonlocal executed
        if not executed:
            executed = True
            return await func(*args, **kwargs)
        return None

    return wrapper

使用 asyncio 实现高效任务调度：

async def batch_run(agent, tasks, concurrency=5):
    """并发执行多个任务"""
    semaphore = asyncio.Semaphore(concurrency)

    async def limited_task(task):
        async with semaphore:
            return await agent.add_task(task)

    await asyncio.gather(*[limited_task(t) for t in tasks])

长期运行的 Agent 需要监控内存使用：

import tracemalloc

def setup_memory_monitor():
    tracemalloc.start()

    def report():
        snapshot = tracemalloc.take_snapshot()
        top_stats = snapshot.statistics('lineno')
        for stat in top_stats[:5]:
            print(stat)

    return report