⚛️ AtomAgent

AtomAgent 是一个极简的 Python AI Agent 框架，基于异步架构设计，提供简洁的工具注册机制，让您轻松构建具有函数调用能力的智能代理。

✨ 核心特性

• 🔌 轻量级的python库依赖：只依赖几个纯python库，无rust库依赖，方便部署在各种资源受限的端侧设备
• 🎯 openai api兼容：支持所有兼容openai api的大模型调用，例如qwen、豆包、deepseek、glm、ollama等大模型api调用
• 🔧 灵活的工具系统：通过装饰器轻松注册工具函数，支持自动参数解析和类型推断
• 🚀 异步支持：完整的异步实现，支持高并发场景和流式响应
• 📡 多工具调用：支持 OpenAI 最新 tools 协议，一次响应可调用多个工具
• 🔄 流式输出：支持实时流式响应，提供更好的用户体验
• 📝 详细日志：内置分级日志系统，便于调试和监控

📦 安装

通过 pip 安装（推荐）

# 安装基础版本（已包含异步支持）
pip install atomagent

从源码安装

# 克隆仓库
git clone https://github.com/liuyunrui123/AtomAgent.git
cd AtomAgent

# 安装开发模式
pip install -e .

# 或者构建并安装
python setup.py install

依赖项

• aiohttp>=3.8.0 - 异步 HTTP 客户端（用于异步 API 调用）
• requests>=2.25.0 - 同步 HTTP 客户端（用于同步 API 调用）
• python-dotenv>=0.19.0 - 环境变量管理

🚀 快速开始

1. 配置环境变量

创建 .env 文件：

BASE_URL=https://api.openai.com/v1/
MODEL_NAME=gpt-4.1
API_KEY=your_api_key_here

2. 创建简单的 Agent

import asyncio
from atomagent import AsyncAgent

async def main():
    # 创建 Agent
    agent = AsyncAgent(
        name="小助手",
        system_prompt="你是一个友好的助手",
        verbose=True
    )
    
    # 简单对话
    response = await agent.chat("你好，今天天气怎么样？")
    print(response)

# 运行
asyncio.run(main())

3. 注册工具函数

import asyncio
from datetime import datetime
from atomagent import AsyncAgent

async def main():
    agent = AsyncAgent(name="小助手", verbose=True)
    
    # 使用装饰器注册工具
    @agent.tool()
    def get_current_time() -> str:
        """获取当前时间"""
        return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")
    
    @agent.tool()
    def get_weather(city: str) -> dict:
        """获取指定城市的天气"""
        return {
            "city": city,
            "temperature": "22°C",
            "weather": "晴朗"
        }
    
    # 使用工具
    response = await agent.chat("现在几点了？北京天气如何？")
    print(response)

asyncio.run(main())

🔥 高级功能

异步 Agent

from atomagent import AsyncAgent
import asyncio

async def main():
    # 创建异步 Agent
    agent = AsyncAgent(
        name="异步助手",
        verbose=True
    )
    
    # 注册异步工具
    @agent.tool()
    async def fetch_data(url: str) -> dict:
        """异步获取数据"""
        # 异步操作
        await asyncio.sleep(1)
        return {"url": url, "status": "success"}
    
    # 异步对话
    response = await agent.chat("请获取 example.com 的数据")
    print(response)

# 运行
asyncio.run(main())

流式响应

import asyncio
from atomagent import AsyncAgent

async def stream_example():
    agent = AsyncAgent(name="助手", verbose=True)
    
    # 异步流式响应
    async for chunk in agent.chat_stream("讲一个故事"):
        print(chunk, end="", flush=True)

# 运行
asyncio.run(stream_example())

多工具并发调用

import asyncio
from atomagent import AsyncAgent

async def concurrent_tools_example():
    agent = AsyncAgent(name="助手", verbose=True)
    
    # 注册多个工具
    @agent.tool()
    def get_weather(city: str) -> dict:
        """获取城市天气"""
        return {"city": city, "temp_c": 25}
    
    @agent.tool()
    def celsius_to_fahrenheit(celsius: float) -> float:
        """摄氏度转华氏度"""
        return celsius * 9/5 + 32
    
    # AI 会智能地同时调用多个工具
    response = await agent.chat("北京、上海、广州的天气如何？都转换成华氏度")
    print(response)
    # AI 会并发调用 get_weather 三次，然后调用温度转换工具

asyncio.run(concurrent_tools_example())

📚 核心组件

AsyncAgent 类

异步智能代理，支持：

• 异步工具调用
• 并发处理多个工具
• 流式响应输出
• 高并发用户请求

适用场景：

• Web 应用
• 高并发场景
• 需要处理多个用户请求
• IO 密集型操作

工具系统

• 自动参数解析：从函数签名自动生成 OpenAI 工具 schema
• 类型推断：支持基本类型、List、Dict、Optional 等
• 参数描述：通过 Annotated 类型添加参数描述
• 灵活注册：支持装饰器和手动注册两种方式

🛠️ 工具开发指南

基础工具

from typing import Annotated
from atomagent import AsyncAgent

agent = AsyncAgent(name="计算助手")

@agent.tool()
def calculate(
    expression: Annotated[str, "数学表达式"],
    precision: Annotated[int, "小数位数"] = 2
) -> float:
    """执行计算"""
    result = eval(expression)
    return round(result, precision)

异步工具

import aiohttp
from typing import Annotated
from atomagent import AsyncAgent

async_agent = AsyncAgent(name="网络助手")

@async_agent.tool()
async def http_request(
    url: Annotated[str, "请求URL"],
    method: Annotated[str, "请求方法"] = "GET"
) -> dict:
    """发送HTTP请求"""
    async with aiohttp.ClientSession() as session:
        async with session.request(method, url) as response:
            return {
                "status": response.status,
                "data": await response.text()
            }

📊 性能优化

异步特性

特性	描述	优势
异步工具调用	支持异步函数作为工具	⚡ 非阻塞执行
并发工具执行	多个工具可同时执行	✅ 提高效率
流式响应	实时输出生成内容	✅ 更好的用户体验
高并发处理	支持多用户同时请求	✅ 适合生产环境

最佳实践

1. 工具设计

   • 保持工具函数简单、独立
   • 避免在工具中维护状态
   • 合理设置超时时间

2. 错误处理

   • 工具函数应返回明确的错误信息
   • 使用 try-except 捕获异常
   • 提供有意义的错误描述

3. 性能优化

   • 使用异步 Agent 处理 IO 密集型任务
   • 合理设置并发限制
   • 避免在工具中执行长时间阻塞操作

🧪 测试

运行测试示例：

# 异步功能测试
python async_agent_demo.py

# 流式响应测试
python test/test_async_streaming.py

📝 示例项目

查看 async_agent_demo.py 了解完整的使用示例，包括：

• 基础对话
• 工具注册和调用
• 多轮对话
• 流式响应
• 错误处理

🤝 贡献

欢迎提交 Issue 和 Pull Request！

📄 许可证

MIT License

🔗 相关资源

• OpenAI API 文档
• Function Calling 指南
• 异步编程最佳实践

---

Made with ❤️ by AtomAgent Team