一个基于阿里云百炼大模型的智能 AI Agent,支持技能扩展、飞书机器人集成、多轮工具调用等功能。
| 功能 | 状态 | 说明 |
|---|---|---|
| 🧠 大模型集成 | ✅ | 阿里云百炼 Qwen 系列模型 |
| 🛠️ 技能系统 | ✅ | 支持扩展各种技能 |
| 🔄 多轮工具调用 | ✅ | 自动规划执行多个工具 |
| 💬 飞书机器人 | ✅ | 企业聊天工具集成 |
| 📝 对话记忆 | ✅ | 上下文管理 |
| 📊 日志系统 | ✅ | 文件轮转日志 |
| 🔒 安全控制 | ✅ | 敏感操作可配置开关 |
| 🌐 网页抓取 | ✅ | 内置网页内容抓取技能 |
| 💻 系统监控 | ✅ | 内置系统资源监控技能 |
| ⚙️ 命令执行 | ✅ | 内置系统命令执行技能 |
┌─────────────────────────────────────────────────────────────┐
│ MiniClaw AI Agent │
├─────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ 飞书机器人 │ │ 命令行 CLI │ │ Webhook Server │ │
│ └──────┬──────┘ └──────┬──────┘ └──────────┬──────────┘ │
│ │ │ │ │
│ └────────────────┼──────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Agent Core │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │ │
│ │ │ Memory │ │ LLM │ │ Scheduler │ │ │
│ │ │ (记忆) │ │ (阿里云百炼)│ │ (工具调度) │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌────────────────┼────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ web_scraper │ │system_monitor│ │ cmd_executor │ │
│ │ (网页抓取) │ │ (系统监控) │ │ (命令执行) │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
└─────────────────────────────────────────────────────────────┘
- Python 3.11+
- 阿里云百炼 API Key
- (可选)飞书应用配置
# 克隆项目
git clone <your-repo-url>
cd miniclaw
# 创建虚拟环境(推荐)
python3.11 -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt# 复制配置模板
cp .env.example .env
# 编辑配置
vim .env最小配置(仅需阿里云 API Key):
# 阿里云百炼配置
DASHSCOPE_API_KEY=sk-your-api-key-here
LLM_MODEL=qwen-max
# 系统配置
LOG_LEVEL=INFO
LOG_FILE=logs/agent.logmkdir -p logs# 命令行对话
python3.11 main.py -m "你好,请介绍一下你自己"
# 查看系统状态
python3.11 main.py -m "查看系统 CPU 和内存使用情况"| 变量名 | 必填 | 默认值 | 说明 |
|---|---|---|---|
DASHSCOPE_API_KEY |
✅ | - | 阿里云百炼 API Key |
LLM_MODEL |
❌ | qwen-max |
模型名称 |
FEISHU_VERIFICATION_TOKEN |
❌ | - | 飞书验证 Token |
FEISHU_APP_ID |
❌ | - | 飞书 App ID |
FEISHU_APP_SECRET |
❌ | - | 飞书 App Secret |
LOG_LEVEL |
❌ | INFO |
日志级别 |
LOG_FILE |
❌ | logs/agent.log |
日志文件路径 |
ENABLE_CMD_EXECUTION |
❌ | false |
是否允许命令执行 |
miniclaw/
├── .env # 环境变量配置
├── .env.example # 配置模板
├── config/
│ └── settings.py # 配置加载模块
└── logs/
└── agent.log # 运行日志
# 查看帮助
python3.11 main.py --help
# 简单对话
python3.11 main.py -m "你好"
# 系统监控
python3.11 main.py -m "查看系统 CPU 使用率"
python3.11 main.py -m "查看系统可用内存"
# 网页抓取
python3.11 main.py -m "帮我抓取 https://www.python.org 的标题"
# 命令执行(需开启 ENABLE_CMD_EXECUTION=true)
python3.11 main.py -m "列出当前目录下的文件"# 启动 Webhook 服务
python3.11 main.py --serve
# 指定端口
python3.11 main.py --serve --port 8080
# 后台运行
nohup python3.11 main.py --serve > logs/feishu.log 2>&1 &
# 查看日志
tail -f logs/feishu.log# 注意:CLI 模式每次是新实例,不支持跨进程多轮对话
# 如需多轮对话,请使用飞书机器人模式或修改代码支持 session
# 第一轮
python3.11 main.py -m "系统内存多少?"
# 第二轮(新实例)
python3.11 main.py -m "那磁盘呢?"skills/
├── __init__.py
├── base.py # 技能基类
├── registry.py # 技能注册中心
├── web_scraper.py # 网页抓取技能
├── system_monitor.py # 系统监控技能
└── cmd_executor.py # 命令执行技能
1. 继承 BaseSkill 类:
# skills/my_skill.py
from skills.base import BaseSkill
from skills.registry import skill_registry
class MySkill(BaseSkill):
name = "my_skill"
description = "我的技能描述"
parameters = {
"type": "object",
"properties": {
"param1": {"type": "string", "description": "参数 1 描述"}
},
"required": ["param1"]
}
def execute(self, param1: str) -> str:
# 技能实现逻辑
return f"执行结果:{param1}"
# 注册技能
skill_registry.register(MySkill())2. 在 core/agent.py 中导入:
# core/agent.py
from skills import web_scraper, system_monitor, cmd_executor, my_skill # 新增3. 测试技能:
python3.11 -c "
from skills.registry import skill_registry
from skills import my_skill
print('已注册技能:', list(skill_registry._skills.keys()))
"
python3.11 main.py -m "使用 my_skill 处理 xxx"| 规范 | 说明 |
|---|---|
| 命名 | 使用小写字母和下划线,如 my_skill |
| 描述 | 清晰描述技能用途和使用场景 |
| 参数 | 使用 JSON Schema 格式定义参数 |
| 返回值 | 返回字符串,便于 LLM 理解 |
| 错误处理 | 捕获异常并返回友好错误信息 |
| 安全性 | 敏感操作需配置开关或白名单 |
- 登录 飞书开发者后台
- 创建企业自建应用
- 获取 App ID 和 App Secret
FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/xxx
FEISHU_VERIFICATION_TOKEN=your-verification-token
FEISHU_APP_ID=cli_xxxxxxxxxxxxxxxx
FEISHU_APP_SECRET=xxxxxxxxxxxxxxxxxxxxxxxx- 事件订阅 → 添加事件
- 订阅地址:
http://你的服务器 IP:5000/feishu/webhook - 验证 Token:与
.env中一致 - 订阅事件:
im.message.receive_v1
im:message 发送消息
im:message.p2p 私聊消息
im:message.group 群聊消息
- 版本管理与发布 → 发布应用
- 添加到飞书
- 在飞书中搜索机器人并添加
# 开放 5000 端口
firewall-cmd --zone=public --add-port=5000/tcp --permanent
firewall-cmd --reload
# 验证
firewall-cmd --zone=public --list-ports | grep 5000阿里云 ECS:
- 控制台 → ECS → 安全组
- 入方向 → 添加规则
- 端口:
5000/5000,协议:TCP,授权对象:0.0.0.0/0
腾讯云 CVM:
- 控制台 → 安全组
- 入站规则 → 添加规则
- 端口:
5000,来源:0.0.0.0/0
from core.agent import Agent
agent = Agent()
# 运行对话
response = agent.run("查看系统内存")
# 重置对话
agent.reset()from skills.registry import skill_registry
# 获取所有工具定义
tools = skill_registry.get_all_tools()
# 获取技能实例
skill = skill_registry.get_skill("system_monitor")
# 执行技能
result = skill_registry.execute("system_monitor")from core.memory import ConversationMemory
memory = ConversationMemory(max_history=20)
# 添加消息
memory.add_message("user", "你好")
memory.add_message("assistant", "你好!有什么可以帮助你的?")
# 获取历史
history = memory.get_history()
# 清空
memory.clear()| 问题 | 原因 | 解决方案 |
|---|---|---|
ModuleNotFoundError |
依赖未安装 | pip install -r requirements.txt |
DASHSCOPE_API_KEY 错误 |
API Key 无效 | 检查阿里云控制台 |
| 技能未注册 | 导入缺失 | 在 agent.py 中显式导入技能模块 |
| 飞书验证失败 | JSON 格式错误 | 确保返回 {"challenge": "xxx"} |
| 飞书连接超时 | 防火墙/安全组 | 开放 5000 端口 |
| 命令执行被拒绝 | 配置未开启 | 设置 ENABLE_CMD_EXECUTION=true |
| 工具调用失败 | 格式不匹配 | 检查 memory.py 中 tool_calls 转换 |
# 实时查看日志
tail -f logs/agent.log
# 查看错误日志
grep ERROR logs/agent.log
# 查看最近 100 行
tail -100 logs/agent.log# 开启 DEBUG 日志
LOG_LEVEL=DEBUG python3.11 main.py -m "测试"
# 查看完整 API 响应
# 在 core/llm.py 中添加 logger.debug(response)# 创建 diagnose.sh
cat > diagnose.sh << 'EOF'
#!/bin/bash
echo "=== MiniClaw 诊断报告 ==="
echo ""
echo "1. Python 版本"
python3.11 --version
echo ""
echo "2. 依赖检查"
python3.11 -c "import openai, requests, flask; print('✅ 核心依赖正常')"
echo ""
echo "3. 技能注册"
python3.11 -c "from skills.registry import skill_registry; from skills import web_scraper, system_monitor, cmd_executor; print('已注册:', list(skill_registry._skills.keys()))"
echo ""
echo "4. 配置检查"
grep -E "^(DASHSCOPE|LOG)" .env
echo ""
echo "5. 端口监听"
netstat -tlnp | grep 5000 || echo "⚠️ 5000 端口未监听"
echo ""
echo "6. 防火墙"
firewall-cmd --list-ports | grep 5000 || echo "⚠️ 5000 端口未开放"
echo ""
echo "=== 诊断完成 ==="
EOF
chmod +x diagnose.sh
./diagnose.sh# 创建服务文件
cat > /etc/systemd/system/miniclaw.service << 'EOF'
[Unit]
Description=MiniClaw AI Agent
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/home/allen/miniclaw
Environment="PATH=/home/allen/miniclaw/venv/bin"
ExecStart=/home/allen/miniclaw/venv/bin/python3.11 main.py --serve
Restart=always
[Install]
WantedBy=multi-user.target
EOF
# 启动服务
systemctl daemon-reload
systemctl enable miniclaw
systemctl start miniclaw
systemctl status miniclaw# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["python3.11", "main.py", "--serve"]# 构建和运行
docker build -t miniclaw .
docker run -d -p 5000:5000 --env-file .env miniclaw# /etc/nginx/conf.d/miniclaw.conf
server {
listen 80;
server_name your-domain.com;
location /feishu/webhook {
proxy_pass http://127.0.0.1:5000/feishu/webhook;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}miniclaw/
├── config/
│ ├── __init__.py
│ └── settings.py # 配置管理
├── core/
│ ├── __init__.py
│ ├── agent.py # Agent 核心
│ ├── llm.py # LLM 接口
│ └── memory.py # 记忆系统
├── skills/
│ ├── __init__.py
│ ├── base.py # 技能基类
│ ├── registry.py # 技能注册
│ ├── web_scraper.py # 网页抓取
│ ├── system_monitor.py # 系统监控
│ └── cmd_executor.py # 命令执行
├── integrations/
│ ├── __init__.py
│ └── feishu.py # 飞书集成
├── utils/
│ ├── __init__.py
│ └── logger.py # 日志系统
├── .env # 环境变量
├── .env.example # 配置模板
├── main.py # 入口文件
├── requirements.txt # 依赖列表
├── README.md # 本文档
├── wsgi.py # WSGI 入口(生产)
├── Dockerfile # Docker 配置
└── logs/
└── agent.log # 运行日志
# .env
ENABLE_CMD_EXECUTION=false # 生产环境建议关闭如需开启,建议添加白名单:
# skills/cmd_executor.py
ALLOWED_COMMANDS = ['ls', 'pwd', 'whoami', 'free', 'df', 'top']
def execute(self, command: str) -> str:
cmd_base = command.split()[0] if command else ""
if cmd_base not in ALLOWED_COMMANDS:
return f"Error: Command '{cmd_base}' not in whitelist"
# ...- 不要将
.env提交到 Git - 使用环境变量或密钥管理服务
- 定期轮换 API Key
# core/agent.py
from datetime import datetime, timedelta
class Agent:
def __init__(self):
self.last_call_time = None
self.min_interval = timedelta(seconds=1)
def run(self, user_input: str):
now = datetime.now()
if self.last_call_time and now - self.last_call_time < self.min_interval:
return "请求过于频繁,请稍后再试"
self.last_call_time = now
# ...# 验证用户输入长度
if len(user_input) > 1000:
return "输入内容过长,请缩短后重试"
# 过滤敏感词
SENSITIVE_WORDS = ['password', 'secret', 'token']
if any(word in user_input.lower() for word in SENSITIVE_WORDS):
return "输入包含敏感内容,无法处理"# integrations/feishu.py
class FeishuIntegration:
def __init__(self):
self._tenant_access_token = None
self._token_expire_time = 0
def _get_tenant_access_token(self):
current_time = time.time()
if self._tenant_access_token and current_time < self._token_expire_time:
return self._tenant_access_token
# ... 获取新 token# core/memory.py
class ConversationMemory:
def __init__(self, max_history: int = 20):
self.max_history = max_history # 限制历史记录数量
def _trim_history(self):
if len(self.history) > self.max_history:
self.history = self.history[-self.max_history:]# 使用 asyncio 处理并发请求
import asyncio
async def process_message(user_input: str):
agent = Agent()
return await asyncio.to_thread(agent.run, user_input)- Fork 项目
- 创建功能分支 (
git checkout -b feature/AmazingFeature) - 提交更改 (
git commit -m 'Add some AmazingFeature') - 推送到分支 (
git push origin feature/AmazingFeature) - 提交 Pull Request
- 遵循 PEP 8 风格指南
- 函数添加类型提示
- 关键逻辑添加注释
- 新功能添加测试
MIT License
- 阿里云百炼 - 大模型服务
- OpenAI Python - API 客户端
- Flask - Web 框架
- 飞书开放平台 - 企业集成
Happy Coding! 🚀