Agent开源框架新手入门指南：从零搭建你的第一个智能体

1次阅读

没有评论

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

传统任务调度系统通常采用集中式控制，比如用 Cron 或 Airflow 来管理定时任务。这种系统有两大局限：

决策僵化：所有执行逻辑需要预先定义，遇到未预料的情况就会失败
缺乏适应性：不同任务之间难以共享上下文（Context），每次执行都是独立过程

Agent 框架通过三个特性解决这些问题：

消息队列（Message Queue）：允许异步事件触发和传递
技能注册（Skill Registration）：动态扩展处理能力
状态保持（State Persistence）：跨任务维持上下文

相比 LangChain 等侧重语言模型的框架，Agent 更擅长需要长期运行和复杂决策的自动化场景（如 IT 运维自动化）。

推荐使用 Python 3.8+ 环境，以下是具体步骤：

创建虚拟环境

python -m venv agent_env
source agent_env/bin/activate  # Linux/Mac
agent_env\Scripts\activate    # Windows

安装基础包
```
pip install agent-framework==1.2.0
```

验证安装

import agent
print(agent.__version__)  # 应输出 1.2.0

示例 requirements.txt（注意版本锁定的重要性）：

agent-framework==1.2.0  # 核心框架
requests==2.28.1        # 常用 HTTP 库
python-dotenv==0.19.0   # 环境变量管理

graph LR
    A[Message Queue] --> B[Router]
    B --> C[Skill 1]
    B --> D[Skill 2]
    C --> E[Context Store]
    D --> E

消息队列：所有外部输入和内部通信都通过标准化消息传递
技能注册 ：通过@skill 装饰器将函数变为可调用的能力单元
上下文保持：每个会话（Session）有独立的数据存储空间

以下是一个天气查询智能体的完整实现：

# 行号 1 -30
from agent import skill, Context
import requests

@skill(name="weather_query", priority=1)
async def get_weather(ctx: Context, city: str):
    """
    输入城市名返回天气信息
    行号 6 - 8 是函数文档，会出现在技能描述中
    """
    try:
        # 从上下文中获取 API 密钥（避免硬编码）api_key = ctx.config["WEATHER_API_KEY"]

        # 演示多轮对话：如果未收到城市参数，主动询问
        if not city:
            await ctx.send("请告诉我您想查询哪个城市的天气？")
            city = await ctx.listen(timeout=30)  # 等待用户输入

        url = f"https://api.weatherapi.com/v1/current.json?key={api_key}&q={city}"
        resp = requests.get(url)
        resp.raise_for_status()  # 自动处理 HTTP 错误

        data = resp.json()
        return f"{city}当前气温：{data['current']['temp_c']}℃"

    except Exception as e:
        # 异常处理逻辑
        ctx.logger.error(f"天气查询失败: {str(e)}")
        return "抱歉，天气查询服务暂时不可用"

关键点说明：

行号 4：@skill装饰器注册技能，priority 决定调用顺序
行号 12：通过 ctx.config 安全获取敏感配置
行号 16：ctx.listen()实现多轮对话等待
行号 25：结构化日志记录

新手常遇到的三个陷阱：

环境变量遗漏：
错误表现：运行时提示 ”Missing API KEY”
解决方案：使用 .env 文件 +python-dotenv 加载
事件循环问题：
错误表现：程序结束后报错 ”Event loop is closed”
解决方案：确保用 asyncio.run() 启动主程序
技能优先级冲突：
错误表现：预期外的技能被触发
解决方案：明确设置 @skill 的 priority 参数

集成 Prometheus 的示例：

from prometheus_client import Counter

weather_queries = Counter('weather_queries_total', '天气查询次数')

@skill(name="weather")
async def get_weather(ctx):
    weather_queries.inc()
    # ... 原有逻辑...

使用 pytest 测试技能：

def test_weather_skill(mocker):
    """测试天气查询技能"""
    mock_ctx = mocker.MagicMock()
    mock_ctx.config = {"WEATHER_API_KEY": "test"}

    # 测试正常流程
    result = asyncio.run(get_weather(mock_ctx, "北京"))
    assert "气温" in result

    # 测试异常流程
    mock_ctx.config = {}
    result = asyncio.run(get_weather(mock_ctx, ""))
    assert "不可用" in result

当某个技能连续失败时，如何设计熔断机制？可以考虑：