Agent工具调用新手入门：从零构建你的第一个自动化工作流

1次阅读

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

传统脚本像按菜谱做菜的厨师，只能线性执行预设步骤。而 Agent 更像是智能管家：

自主性 ：根据环境动态调整行为（如发现磁盘满时自动清理日志）
反应式 ：实时响应事件（如监控到新文件立即触发处理）
目标驱动 ：持续朝着既定目标运作（如保持服务器 CPU 利用率 <70%）

举个实际例子：传统备份脚本每天固定时间全量备份，而 Agent 可以根据文件变化频率智能调整备份策略。

Agent 的核心是 OODA 循环（Observe-Orient-Decide-Act）：

stateDiagram-v2
    [*] --> Observe: 启动
    Observe --> Orient: 获取环境状态
    Orient --> Decide: 评估可行动作
    Decide --> Act: 执行最优动作
    Act --> Observe: 循环

关键术语解释：

动作空间 (Action Space)：Agent 能执行的所有操作（如移动文件、发送告警）
环境状态 (State)：Agent 感知到的系统状况（如目录文件列表、CPU 温度）

安装依赖：

pip install watchdog

核心代码框架：

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class FileHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"检测到文件变化: {event.src_path}")

observer = Observer()
event_handler = FileHandler()
observer.schedule(event_handler, path="./monitor_dir", recursive=True)
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

事件过滤 （只处理.csv 文件）

def on_modified(self, event):
    if event.src_path.endswith('.csv'):
        self.process_csv(event.src_path)

优先级队列 （使用 heapq 模块）

import heapq

task_queue = []

def add_task(filepath, priority=0):
    heapq.heappush(task_queue, (-priority, filepath))  # 使用负数实现最大堆

异常处理 （网络请求重试）

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def upload_file(filepath):
    # 模拟可能失败的 API 调用
    requests.post("http://api.example.com/upload", files={"file": open(filepath)})

针对高频文件事件场景：

批处理 ：累计 5 个事件后统一处理

from collections import deque

batch = deque(maxlen=5)

def on_modified(self, event):
    batch.append(event.src_path)
    if len(batch) >= 5:
        process_batch(list(batch))
        batch.clear()

异步 IO（使用 asyncio）

import asyncio

async def process_file(filepath):
    await asyncio.sleep(0.1)  # 模拟 IO 操作
    print(f"处理完成: {filepath}")

资源竞争 ：多个 Agent 同时写日志文件
方案：使用文件锁（fcntl.flock）
循环触发 ：处理文件时又生成新文件导致死循环
方案：设置忽略列表或特殊后缀（如.processing）
内存泄漏 ：长期运行后占用内存持续增长
方案：定期调用 gc.collect()，避免全局变量累积

让 Agent 理解自然语言指令：

from openai import OpenAI

client = OpenAI()

def interpret_command(user_input):
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": f"将用户指令转为 JSON 命令: {user_input}"}]
    )
    return json.loads(response.choices[0].message.content)

示例指令：” 当有新销售数据时，先备份到 S3 再通知数据分析团队 ”