uni-agent-sdk 0.2.4
极简的uni-im平台智能体开发框架
License: MIT
Requires Python: >=3.8
Keywords: chatbot, agent, ai, uni-im, messaging

# uni-agent-sdk

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

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

## 🚀 特性

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

## 📦 安装

```bash
pip install uni-agent-sdk
```

## ⚡ 快速开始

### 最简单的智能体

```python
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）

```python
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)
```

### 上下文感知智能体

```python
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)
```

## 🔧 配置

### 环境变量配置

```bash
# 必需配置
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"
```

### 代码配置

```python
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**: 配置管理

## 🧪 测试

```bash
# 运行所有测试
python run_tests.py

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

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

## 📖 API文档

### Agent类

```python
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
```

### 消息类型

```python
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服务

```python
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请求
- 支持签名过期时间设置
- 敏感信息自动脱敏显示
- 配置验证和类型检查

## 🚀 部署

### 单进程部署

```python
# 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部署

```dockerfile
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](LICENSE) 文件了解详情。

## 🆘 支持

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

## 🏆 致谢

感谢以下开源项目：

- [litellm](https://github.com/BerriAI/litellm) - 统一LLM API
- [pika](https://github.com/pika/pika) - RabbitMQ客户端
- [pydantic](https://github.com/pydantic/pydantic) - 数据验证
- [aiohttp](https://github.com/aio-libs/aiohttp) - 异步HTTP客户端