uni-agent-sdk 0.2.2

极简的uni-im平台智能体开发框架

uni-agent-sdk

智能体开发SDK - 极简的uni-im平台智能体开发框架

让开发者只需3行代码就能创建强大的智能体，自动处理消息接收、LLM推理、响应发送等复杂流程。

🚀 特性

• 极简API: 只需3行代码创建智能体
• 无状态架构: 基于RabbitMQ的无状态消费者模式
• 开箱即用: 内置LLM集成、平台通信、文件处理
• 类型安全: 完整的类型注解和Pydantic验证
• 高性能: 异步处理，支持并发消息处理
• 易于扩展: 支持函数调用、上下文记忆等高级功能

📦 安装

pip install uni-agent-sdk

⚡ 快速开始

最简单的智能体

import asyncio
from uni_agent_sdk import Agent, Message, Response

class SimpleAgent(Agent):
    async def handle_message(self, message: Message, context: dict) -> Response:
        return Response.text(f"你好！我收到了：{message.content}")

# 启动智能体
agent = SimpleAgent(
    api_key="your_api_key",
    api_secret="your_api_secret",
    openrouter_api_key="your_openrouter_key"
)
asyncio.run(agent.start())

AI智能体（使用LLM）

class AIAgent(Agent):
    async def handle_message(self, message: Message, context: dict) -> Response:
        # 直接使用内置的LLM服务
        reply = await self.llm.chat(
            message.content,
            system_prompt="你是一个友善的AI助手"
        )
        return Response.text(reply)

上下文感知智能体

class ContextualAgent(Agent):
    async def handle_message(self, message: Message, context: dict) -> Response:
        # 获取历史消息
        history = context.get('messages', [])

        # 构建对话历史
        conversation = []
        for hist_msg in history[-5:]:  # 最近5条消息
            conversation.append({
                "role": "user" if hist_msg['from_uid'] == message.from_uid else "assistant",
                "content": hist_msg['body']
            })

        # 添加当前消息
        conversation.append({"role": "user", "content": message.content})

        # 生成回复
        reply = await self.llm.chat(conversation)
        return Response.text(reply)

🔧 配置

环境变量配置

# 必需配置
export OPENROUTER_API_KEY="sk-or-v1-xxx"
export RABBITMQ_HOST="your-rabbitmq-host"
export RABBITMQ_USER="your-username"
export RABBITMQ_PASSWORD="your-password"

# 可选配置
export PLATFORM_BASE_URL="https://uni-im.dcloud.net.cn"
export DEFAULT_MODEL="openai/gpt-3.5-turbo"
export LOG_LEVEL="INFO"

代码配置

agent = MyAgent(
    api_key="robot_xxx",
    api_secret="xxx",

    # OpenRouter配置
    openrouter_api_key="sk-or-v1-xxx",

    # RabbitMQ配置
    rabbitmq_host="localhost",
    rabbitmq_port=5673,
    rabbitmq_user="guest",
    rabbitmq_password="guest",

    # LLM配置
    default_model="openai/gpt-3.5-turbo",
    default_temperature=0.7,

    # 其他配置
    log_level="INFO"
)

📚 示例

查看 examples/ 目录获取更多示例：

• simple_agent.py: 最简单的回声智能体
• ai_agent.py: 使用LLM的AI智能体
• contextual_agent.py: 支持上下文记忆的智能体
• function_agent.py: 支持函数调用的智能体

🏗️ 架构

用户消息 → uni-im平台 → RabbitMQ → SDK消费者 → 开发者智能体
                                                    ↓
用户收到回复 ← uni-im平台 ← HTTP API ← SDK响应 ← LLM推理

核心组件

• Agent: 智能体基类，处理消息接收和响应
• LLMService: LLM服务，支持多种模型
• PlatformAPI: 平台通信服务
• FileService: 文件上传下载服务
• Config: 配置管理

🧪 测试

# 运行所有测试
python run_tests.py

# 运行特定测试
python -m unittest tests.test_crypto

# 运行集成测试
python -m pytest tests/test_integration.py -v

📖 API文档

Agent类

class Agent(ABC):
    """智能体抽象基类"""

    @abstractmethod
    async def handle_message(self, message: Message, context: dict) -> Response:
        """处理消息的抽象方法"""
        pass

    async def start(self):
        """启动智能体"""
        pass

    async def stop(self):
        """停止智能体"""
        pass

消息类型

class Message(BaseModel):
    """接收消息模型"""
    id: str
    from_uid: str
    to_uid: str
    conversation_id: str
    content: str
    message_type: str = "text"
    create_time: int

class Response(BaseModel):
    """响应消息模型"""
    content: str
    response_type: str = "text"
    metadata: Dict[str, Any] = {}

    @classmethod
    def text(cls, content: str) -> "Response":
        """创建文本响应"""

    @classmethod
    def image(cls, image_url: str, caption: str = "") -> "Response":
        """创建图片响应"""

    @classmethod
    def file(cls, file_url: str, filename: str) -> "Response":
        """创建文件响应"""

LLM服务

class LLMService:
    """LLM推理服务"""

    async def chat(self, messages, model=None, temperature=None) -> str:
        """聊天接口"""

    async def chat_stream(self, messages, model=None) -> AsyncIterator[str]:
        """流式聊天接口"""

    async def function_call(self, messages, functions) -> Dict:
        """函数调用接口"""

🔒 安全

• 使用HMAC-SHA256签名验证API请求
• 支持签名过期时间设置
• 敏感信息自动脱敏显示
• 配置验证和类型检查

🚀 部署

单进程部署

# main.py
import asyncio
from my_agent import MyAgent

async def main():
    agent = MyAgent(
        api_key="xxx",
        api_secret="xxx",
        openrouter_api_key="xxx"
    )
    await agent.start()

if __name__ == "__main__":
    asyncio.run(main())

Docker部署

FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
CMD ["python", "main.py"]

多实例负载均衡

同一智能体可以运行多个进程实例，RabbitMQ会自动进行负载均衡。

🤝 贡献

欢迎提交Issue和Pull Request！

1. Fork 项目
2. 创建特性分支 (git checkout -b feature/AmazingFeature)
3. 提交更改 (git commit -m 'Add AmazingFeature')
4. 推送到分支 (git push origin feature/AmazingFeature)
5. 开启Pull Request

📄 许可证

本项目采用 MIT 许可证 - 查看 LICENSE 文件了解详情。

🆘 支持

• 📧 邮箱: [e-mail]
• 💬 QQ群: 123456789
• 📖 文档: https://uni-im.dcloud.net.cn/docs
• 🐛 问题反馈: https://github.com/uni-im/uni-agent-sdk/issues

🏆 致谢

感谢以下开源项目：

• litellm - 统一LLM API
• pika - RabbitMQ客户端
• pydantic - 数据验证
• aiohttp - 异步HTTP客户端