AI Agent 框架选型实战：2026 年 5 大主流框架深度对比

# 安装命令
pip install langgraph>=0.3.0          # LangGraph
pip install autogen-agentchat>=0.4.0  # AutoGen AG2
pip install crewai>=0.80.0            # CrewAI
pip install openclaw-cli              # OpenClaw Skills
# Proactive Agent 需从 GitHub 安装
pip install git+https://github.com/openclaw/proactive-agent.git

# CrewAI (约 30 行)
from crewai import Agent, Task, Crew

researcher = Agent(role="Researcher", goal="Find info", backstory="Expert")
task = Task(description="Search {topic}", expected_output="Summary", agent=researcher)
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff(inputs={"topic": "AI Agents"})

# LangGraph (约 60 行)
from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List[str]

graph = StateGraph(AgentState)
graph.add_node("search", search_node)
graph.add_node("summarize", summarize_node)
graph.add_edge("search", "summarize")
app = graph.compile()
result = app.invoke({"messages": ["AI Agents"]})

# AutoGen (约 50 行)
from autogen import AssistantAgent, UserProxyAgent, GroupChat

assistant = AssistantAgent("assistant", llm_config={...})
user_proxy = UserProxyAgent("user", code_execution_config=False)
group_chat = GroupChat(agents=[assistant, user_proxy], messages=[])
result = user_proxy.initiate_chat(assistant, message="Search AI Agents")

用户消息 → 意图识别 → 条件路由 → 处理节点 → 回复
                    ↓
              [订单查询] [政策咨询] [产品规格] [投诉]
                    ↓
              检查点保存 → 人类介入（如需要）

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional

class ConversationState(TypedDict):
    messages: List[str]
    intent: Optional[str]
    order_id: Optional[str]
    resolved: bool

def intent_classifier(state: ConversationState):
    # 调用 LLM 分类意图
    intent = llm.classify(state["messages"][-1])
    return {"intent": intent}

def order_lookup(state: ConversationState):
    order_id = extract_order_id(state["messages"])
    order_info = db.query(order_id)
    return {"order_id": order_id, "messages": [format_response(order_info)]}

def human_handoff(state: ConversationState):
    # 发送到人工客服队列
    slack.notify(state["messages"])
    return {"messages": ["已转接人工客服，请稍候"]}

# 构建图
graph = StateGraph(ConversationState)
graph.add_node("classify", intent_classifier)
graph.add_node("order_lookup", order_lookup)
graph.add_node("handoff", human_handoff)

# 条件路由
graph.add_conditional_edges("classify", 
    lambda s: s["intent"],
    {
        "order": "order_lookup",
        "complaint": "handoff",
        "general": "order_lookup"  # 简化示例
    }
)

graph.add_edge("order_lookup", END)
graph.add_edge("handoff", END)

app = graph.compile(checkpointer=PostgresSaver())

from crewai import Agent, Task, Crew, Process

# 研究员：负责收集信息
researcher = Agent(
    role="市场研究员",
    goal="收集竞品最新动态和信息",
    backstory="你有 10 年市场研究经验，擅长从各种渠道获取信息",
    tools=[SearchTool(), ScrapeTool()],
    verbose=True
)

# 分析师：负责提炼洞察
analyst = Agent(
    role="数据分析师",
    goal="从收集的信息中提炼关键洞察",
    backstory="你是资深数据分析师，擅长发现数据背后的趋势",
    verbose=True
)

# 写作者：负责生成报告
writer = Agent(
    role="技术写作者",
    goal="撰写清晰、专业的市场分析报告",
    backstory="你是资深技术写作者，擅长把复杂信息变成易读的报告",
    verbose=True
)

# 任务 1：收集信息
research_task = Task(
    description="搜索以下竞品的最新动态：{competitors}",
    expected_output="包含 5-10 条关键动态的列表，每条注明来源和日期",
    agent=researcher
)

# 任务 2：分析趋势
analysis_task = Task(
    description="基于收集的信息，分析市场趋势和机会",
    expected_output="3-5 个关键趋势，每个趋势有数据支撑",
    agent=analyst,
    context=[research_task]
)

# 任务 3：撰写报告
writing_task = Task(
    description="撰写周报，包含动态汇总、趋势分析、建议行动",
    expected_output="1500-2000 字的专业报告，Markdown 格式",
    agent=writer,
    context=[research_task, analysis_task]
)

# 组建 Crew
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    verbose=True
)

# 执行
result = crew.kickoff(inputs={"competitors": "竞品 A, 竞品 B, 竞品 C"})

定时触发器 → WAL 日志记录 → 技能执行 → 结果保存
                                    ↓
                              失败？→ 重试机制
                                    ↓
                              通知（Slack/邮件）

---
name: daily-backup-check
description: 检查每日数据库备份是否完成
version: 1.0.0
triggers:
  - "检查备份"
  - "备份状态"
requires:
  bins: [pg_dump, curl]
  env: [DB_HOST, DB_USER, SLACK_WEBHOOK]
---

# Daily Backup Check Skill

## 职责
每天检查数据库备份是否按时完成，失败则通知运维团队。

## 执行步骤
1. 查询 `backup_logs` 表，获取最近 24 小时的备份记录
2. 检查是否有成功标记的记录
3. 如果没有，发送 Slack 告警
4. 记录检查结果到 WAL 日志

## 输出格式
```json
{
  "status": "success|failed",
  "last_backup": "2026-03-22 02:00:00",
  "message": "备份正常"
}

**定时任务配置**（`config/crons.yaml`）：

```yaml
tasks:
  - name: backup-check
    skill: daily-backup-check
    schedule: "0 9 * * *"  # 每天 9 点
    retry: 3
    timeout: 300
    notify_on_failure: true

# 简化的 WAL 日志
import json
from datetime import datetime

class WALProtocol:
    def __init__(self, log_path="wal.log"):
        self.log_path = log_path

    def log(self, action: str, data: dict):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "action": action,
            "data": data
        }
        with open(self.log_path, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def recover(self):
        # 从日志恢复状态
        entries = []
        with open(self.log_path) as f:
            for line in f:
                entries.append(json.loads(line))
        return entries