Python自動化AI工作流:AI Agent平臺框架入門與實戰指南

用Python打造你的AI自動化流水線:AI Agent平臺框架入門指南
每天重復執行AI任務是不是很煩?比如定時讓AI總結新聞,或者收到郵件后自動分類回復。手動操作太累,寫腳本又得從頭對接各種API。AI Agent平臺就是為解決這類問題設計的開源框架,用幾行Python代碼就能搭建可編程的AI工作流。
為什么選擇AI Agent平臺?
AI Agent平臺的核心優勢是開源可自部署。數據完全留在自己的服務器上,隱私有保障。它支持50多種服務集成(數據庫、郵件、Slack等),而且用Python寫工作流,學習成本很低。MIT許可證意味著可以自由用于商業項目。
安裝與基礎配置
首先確保環境有Python 3.8+和pip。安裝步驟很簡單:
# 創建虛擬環境(推薦)
python -m venv ai-agent-env
source ai-agent-env/bin/activate # Linux/Mac
# ai-agent-env\Scripts\activate # Windows
# 安裝AI Agent平臺
pip install ai-agent-framework
# 驗證安裝
python -c "import ai-agent; print(ai-agent.__version__)"為什么用虛擬環境? 避免不同項目的依賴包沖突,保持系統Python環境干凈。
定義你的第一個工作流
AI Agent平臺的工作流由觸發器和動作組成。先創建一個簡單的HTTP觸發工作流:收到HTTP請求時,調用AI處理數據。
創建文件 my_first_workflow.py:
from ai-agent import Workflow, HttpTrigger, AiAction
import json
# 定義工作流
workflow = Workflow(
name="郵件智能分類器",
description="收到郵件內容后自動判斷類別并回復"
)
# 添加HTTP觸發器
http_trigger = HttpTrigger(
path="/classify-email", # 訪問路徑
method="POST" # 只接受POST請求
)
workflow.add_trigger(http_trigger)
# 定義AI動作
ai_action = AiAction(
model="gpt-3.5-turbo", # 可以換成其他模型
prompt_template="""
請分析這封郵件的類別(工作/個人/廣告/緊急)并生成簡短回復:
郵件內容:{email_content}
返回JSON格式:
{{"category": "類別", "reply": "回復內容"}}
""",
input_mapping={"email_content": "$trigger.body.content"} # 從請求體獲取內容
)
workflow.add_action(ai_action)
# 添加響應動作(返回AI結果)
workflow.set_response("$action.ai_action.output")
# 導出工作流配置
if __name__ == "__main__":
workflow.export("email_classifier.json")
print("工作流配置已導出!")為什么這樣設計? HTTP觸發器讓你可以通過API調用工作流,適合集成到現有系統。input_mapping讓數據流動更靈活,$trigger.body.content表示從觸發請求的body中提取content字段。
啟動HTTP服務
現在啟動工作流服務:
# 啟動工作流引擎(開發模式)
ai-agent serve --workflow my_first_workflow.py --port 8080
# 或者直接運行Python文件
python my_first_workflow.py
ai-agent serve --config email_classifier.json --port 8080測試一下:
# 用curl測試
curl -X POST http://localhost:8080/classify-email \
-H "Content-Type: application/json" \
-d '{"content": "明天下午3點開會討論項目進度,請準備相關材料。"}'你應該會看到類似這樣的響應:
{
"category": "工作",
"reply": "已記錄會議安排,我會準時參加并準備好項目材料。"
}定時執行模式:每日新聞摘要
HTTP觸發適合實時任務,但有些任務需要定期執行。比如每天早上8點自動生成新聞摘要。AI Agent平臺同樣支持定時觸發器。
創建 daily_news_summary.py:
from ai-agent import Workflow, ScheduleTrigger, AiAction, EmailAction
workflow = Workflow(name="每日AI新聞摘要")
# 定時觸發:每天早上8點
schedule_trigger = ScheduleTrigger(
cron="0 8 * * *", # cron表達式:每天8:00
timezone="Asia/Shanghai"
)
workflow.add_trigger(schedule_trigger)

# 第一步:抓取新聞(這里用模擬數據)
fetch_action = {
"type": "http_request",
"url": "https://newsapi.org/v2/top-headlines?country=cn&apiKey=YOUR_KEY",
"method": "GET"
}
workflow.add_action(fetch_action)
# 第二步:AI總結新聞
summary_action = AiAction(
model="gpt-4",
prompt_template="""
請用中文總結今天的5條最重要科技新聞,格式:
1. [標題] - 一句話摘要
2. ...
新聞數據:{news_data}
""",
input_mapping={"news_data": "$action.http_request.output"}
)
workflow.add_action(summary_action)
# 第三步:發送郵件
email_action = EmailAction(
to="your-email@example.com",
subject="每日科技新聞摘要 - {date}",
body="$action.ai_action.output",
smtp_config={
"host": "smtp.gmail.com",
"port": 587,
"username": "your-email@gmail.com",
"password": "your-app-password" # 使用應用專用密碼
}
)
workflow.add_action(email_action)
# 導出配置
workflow.export("daily_news.json")為什么用cron表達式? 這是定時任務的標準語法,0 8 * * *表示每天第8小時0分鐘執行。你可以用 crontab.guru 在線驗證表達式。
部署到生產環境
開發測試沒問題后,可以部署到服務器。AI Agent平臺支持多種部署方式:
# 方式1:直接運行(適合測試)
ai-agent serve --config daily_news.json --daemon
# 方式2:Docker部署(推薦生產環境)
docker run -d \
--name ai-agent-news \
-v $(pwd)/workflows:/app/workflows \
-e OPENCLAW_CONFIG=/app/workflows/daily_news.json \
ai-agent/ai-agent:latest
# 方式3:systemd服務(Linux服務器)
sudo nano /etc/systemd/system/ai-agent-news.servicesystemd服務文件示例:
[Unit]
Description=AI Agent平臺 News Summary Service
After=network.target
[Service]
Type=simple
User=your-user
WorkingDirectory=/path/to/workflows
ExecStart=/path/to/ai-agent-env/bin/ai-agent serve --config daily_news.json
Restart=always
[Install]
WantedBy=multi-user.target然后啟動服務:
sudo systemctl daemon-reload
sudo systemctl enable ai-agent-news
sudo systemctl start ai-agent-news
sudo systemctl status ai-agent-news驗證與調試
工作流運行后,怎么確認它正常工作?AI Agent平臺提供了日志和監控:
# 查看實時日志
ai-agent logs --follow --name email_classifier
# 查看執行歷史
ai-agent history --workflow daily_news --limit 10
# 手動觸發定時任務(測試用)
ai-agent trigger --workflow daily_news --manual如果遇到問題,常見原因:
- API密鑰錯誤:檢查模型API密鑰是否有效
- 網絡問題:確保服務器能訪問外部API
- 權限問題:檢查文件權限和SMTP密碼
下一步學習
現在你已經掌握了AI Agent平臺的基礎用法。接下來可以探索:
- 條件分支:根據AI返回結果執行不同動作
- 錯誤處理:添加重試機制和失敗通知
- 數據庫集成:將工作流結果存儲到PostgreSQL/MySQL
- 多模型對比:同時調用GPT-4和Claude,對比結果
推薦閱讀:
最好的學習方式是動手嘗試。從解決自己的一個小需求開始,比如自動備份聊天記錄,或者監控網站狀態。AI Agent平臺的開源特性意味著你可以隨時查看源碼、提交改進建議,甚至開發自己的插件。
遇到問題? 龍蝦社區有很多開發者分享經驗,歡迎在 www.nhjb.com.cn/community 提問交流。