OpenClaw大模型Agent工作流稳定化实践：新闻抓取与邮件管理程序封装

我们向AI描述任务时，其实是用自然语言写了一段充满歧义的“代码”。这段“代码”的执行环境极不稳定，导致同样一句话，在不同时间或环境下可能产生千差万别的结果。大模型常将大量Token浪费在“尝试-失败-再尝试”的循环里，而且我们还很难对过程进行调试。

在云栈社区的实践交流中，一个核心原则逐渐清晰：将确定性的、不稳定的任务交给稳定的程序，让大模型回归其擅长的决策与总结。

一、痛点分析：当自然语言指令遭遇不稳定的执行

想象以下几个常见场景：

• “获取今天新闻”：大模型可能会尝试直接爬取网站（被反爬机制屏蔽）、调用未提供密钥的API，或者干脆回应“无法浏览网页”。
• “帮我看看邮箱里有没有重要邮件”：大模型无法知晓你的邮箱服务器地址、账号密码，更无法安全地访问你的收件箱。
• “让Agent A发消息给Agent B”：每新增一个智能体，你都需要重新教它如何配置邮件、微信或飞书，这是巨大的重复劳动。

结果就是，大模型把宝贵的计算资源消耗在不稳定的执行尝试上，而我们却难以追溯问题根源。

二、解决方案：为模糊指令编写确定性的程序接口

我们的解决思路非常直接：

1. 程序化确定性任务：所有网络请求、文件读写、邮件收发、复杂计算等任务，全部封装成有日志、有错误处理、可重试的稳定程序。
2. 大模型专注决策：大模型只负责两件事：① 理解用户意图，决定调用哪个程序（类似“编译器”）；② 对程序返回的结构化数据进行筛选、总结与推荐。
3. 接口化与共享：将这些程序统一封装为API或MCP工具，所有Agent共享同一套能力库。Agent间的通讯也通过这套共享接口间接完成。

下面通过两个具体例子来阐释这一理念。

---

实例一：新闻获取 —— 将“抓取”动作程序化

传统方式（依赖大模型直接执行）
用户提问：“今天有什么值得看的科技新闻？”
大模型会尝试访问多个网站，可能遭遇反爬、解析出无关广告，即使成功也会返回大量原始HTML（消耗上千Token），最终勉强总结几条，还时常遗漏重要信源。

新方式：程序抓取 + 大模型筛选

1. 编写稳定的新闻获取程序 (news_fetcher.py)
   此程序调用固定的API（如NewsAPI）或RSS源，返回结构化的JSON数据。

   import requests
   
   def get_latest_news():
       try:
           # 使用稳定新闻源API
           response = requests.get('https://gnews.io/api/v4/top-headlines?token=YOUR_KEY&lang=zh')
           data = response.json()
           articles = [{'title': a['title'], 'source': a['source']['name'], 'url': a['url']} for a in data['articles']]
           return articles
       except Exception as e:
           log_error(f“新闻获取失败: {e}”)
           return []  # 失败时返回空列表，不干扰大模型流程

2. 封装为API
   例如 GET /api/news/latest，返回格式：

   {
     “status”: “success”,
     “news”: [
       {“title”: “…”, “source”: “…”, “url”: “…”}
     ]
   }

3. OpenClaw调用流程
   • 用户指令触发机器人调用 GET /api/news/latest（仅消耗少量Token获取新闻标题列表）。
   • 将标题列表发送给大模型：“请从中选出3条最值得阅读的，并说明理由。”
   • 大模型进行精炼推荐后输出给用户。

效果：新闻获取成功率接近100%，每次调用仅需几十Token（获取列表）加几百Token（进行总结）。一旦出错，直接查询新闻程序的日志即可快速定位问题。

---

实例二：邮件管理与Agent间通讯 —— 统一接口，能力共享

痛点场景
当你拥有多个Agent（如新闻助手、运维监控、客服机器人）时，它们可能都需要发送邮件通知或读取收件箱。如果每个Agent都单独配置SMTP/IMAP，不仅配置繁琐、密码管理混乱，一旦需要更换邮箱服务器，所有Agent都需要修改。

新方式：编写统一的邮件服务程序

1. 核心邮件服务类 (mail_service.py)
   这个类封装了稳定的邮件收发逻辑。

   import smtplib
   import imaplib
   import email
   
   class MailService:
       def __init__(self, config):
           self.smtp_server = config[‘SMTP_SERVER’]
           self.imap_server = config[‘IMAP_SERVER’]
           self.username = config[‘EMAIL’]
           self.password = config[‘PASSWORD’]
   
       def send_email(self, to, subject, body):
           # 包含重试和日志的稳定发送逻辑
           try:
               with smtplib.SMTP_SSL(self.smtp_server, 465) as server:
                   server.login(self.username, self.password)
                   msg = f”Subject: {subject}\n\n{body}”
                   server.sendmail(self.username, to, msg)
               return {“status”: “sent”, “to”: to}
           except Exception as e:
               log_error(f”发送失败: {e}”)
               return {“status”: “error”, “message”: str(e)}
   
       def read_unread(self, folder=“INBOX”):
           # 读取未读邮件，返回结构化列表
           # … 具体实现细节
           return [{“from”: x, “subject”: y, “body_preview”: z}]

2. 封装为统一API
   • 发邮件：POST /api/mail/send，参数 to, subject, body
   • 读邮件：GET /api/mail/unread，返回未读邮件列表
3. 所有Agent共享调用
   • 新闻Agent：筛选出好文章后，调用 POST /api/mail/send 将摘要发送到你的邮箱。
   • 运维Agent：检测到服务器异常，调用同一个API发送报警邮件。
   • 客服Agent：定时调用 GET /api/mail/unread 检查客户问题，经大模型生成回复后，再通过API发送回复邮件。
4. 实现Agent间解耦通讯
   Agent之间可以通过这套共享的邮件服务进行异步通讯，无需知晓对方的具体配置。
   • 监控Agent需要通知日志分析Agent时，调用 POST /api/mail/send，收件人填写为 log-analyzer@yourdomain（一个由日志Agent负责轮询的邮箱）。
   • 日志Agent调用 GET /api/mail/unread 获取该指令邮件，解析并执行任务，完成后同样通过邮件API回复结果。
   • 双方仅需约定邮件主题格式，共享的邮件程序便充当了可靠、解耦的消息队列。

效果对比

维度	每个Agent单独配置邮件	统一邮件程序接口
配置成本	N个Agent需配置N次	只配置一次，所有Agent共享
安全性	密码散落在多个配置中	密码仅存在于邮件程序内部
Agent间通讯	需互相知道地址和协议	通过共享服务解耦，只需约定格式
更换邮箱服务器	需修改所有Agent配置	仅修改邮件程序一处配置
调试	需查看每个Agent的日志	所有邮件操作日志集中管理

---

三、快速落地：使用FastAPI封装邮件服务API

以下是一个使用 FastAPI 快速将邮件服务暴露为HTTP API的示例：

# mail_api.py - 使用FastAPI
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
from mail_service import MailService

app = FastAPI()
mail = MailService(load_config())  # 创建单例，所有请求共享

class SendRequest(BaseModel):
    to: str
    subject: str
    body: str

@app.post(“/api/v1/mail/send”)
async def send_email(req: SendRequest, x_api_key: str = Header(…)):
    if not verify_key(x_api_key):
        raise HTTPException(401)
    result = mail.send_email(req.to, req.subject, req.body)
    return result

@app.get(“/api/v1/mail/unread”)
async def get_unread(x_api_key: str = Header(…)):
    result = mail.read_unread()
    return {“status”: “ok”, “emails”: result}

在OpenClaw中，任何Agent都可以通过其 http_request 动作轻松调用这些接口。

---

四、进阶：实现Agent间高效内部通讯

如果觉得邮件通讯延迟较高，可以沿用相同模式，封装更高效的消息中间件。

• 例如，封装 Redis Pub/Sub 或 RabbitMQ，暴露为统一API：
  • POST /api/mq/publish → 发送消息到指定频道
  • GET /api/mq/subscribe?channel=xxx → （长轮询或WebSocket）接收消息
• 这样，Agent之间可以实现毫秒级的实时通讯，并且依然是“一次编写，所有Agent共用”。

五、实战流程示例

在实际操作中，你可以指挥大模型完成从环境搭建到程序封装的完整流程。例如：

1. 让OpenClaw在虚拟机上部署一个安全的Email服务，并创建一批测试邮箱，最后生成包含所有连接信息的文档。
2. 将这份文档交给另一个大模型（如Trae），指令它：“根据提供的邮箱信息，编写一个邮件收发插件。需支持收发邮件和附件，能读取邮件内容、标题、发件人、主题。请完成测试（邮箱互发、附件互发），并提供接口地址、使用方式及完整的API文档。”
3. 大模型基于指令和文档，编写出可工作的插件代码。

（邮件插件的实现与测试文档）

六、总结：让大模型成为可靠的“指挥官”

通过上述新闻获取和邮件管理的例子，我们可以总结出构建稳定Agent工作流的核心：

• 每个模糊的自然语言指令背后，都应映射到一个确定的程序。程序负责处理所有不稳定、需权限、重复性的脏活累活。
• 大模型无需学习具体的技术细节（如配置SMTP、应对反爬），只需学会在正确时机调用对应API。这极大节省了Token，并提升了任务成功率。
• 所有Agent共享同一套工具集，新加入的Agent无需重复配置，直接调用现有API即可获得能力。Agent间通过统一的消息接口解耦通讯。

最终，你的OpenClaw机器人集群将呈现出清晰的微服务架构形态——每个Agent是轻量级的“决策大脑”，背后由一系列稳定的“能力API”提供支持。任何问题，通过查询程序日志即可快速定位；任何新需求，编写一个新API后，所有Agent便能立即获得这项新能力。

这正是将“充满歧义的自然语言提示词，编译成确定性的程序调用”所释放的真正威力。一个真正能投入工作的智能体系统，必须是结果稳定、流程可控的。将日常琐事、定时任务交给它，我们才能从重复劳动中解放出来，专注于更需要创造力的工作。