1. 从零复刻 OpenClaw:为什么用 Flask + DeepSeek 搭 SkillHub 骨架
OpenClaw 这类 AI Agent 最近被讨论得很多,但真正拆开看,它的核心并不神秘:一个能理解自然语言的大模型,加上一套可注册、可调度、可扩展的技能系统,再配一个把两者串起来的服务层。你给它一句“帮我查一下今天的新闻”,它先判断这句话该调用哪个技能,再把技能执行结果回灌给模型做总结。这套“意图解析 + 技能调度”的循环,就是 OpenClaw 式 Agent 的运行骨架。
我这次要做的,是一个叫 iClaw 的最小可运行版本。它不追求功能大而全,而是把 SkillHub 这个技能注册与调度中心讲透:技能怎么用 JSON 描述、Flask 怎么暴露注册和调用接口、DeepSeek 怎么负责把用户意图翻译成技能调用、新增一个自定义技能后怎么验证它真的被调起来了。适合已经会一点 Python、想搞懂 AI Agent 背后调度逻辑的开发者,也适合想给自己项目加一个“能动手”的 Agent 层的同学。
整条链路是这样的:前端或命令行发来一句自然语言,Flask 后端把它连同技能清单一起交给 DeepSeek,DeepSeek 返回一个结构化的技能调用指令,后端解析后执行对应技能,再把结果返回。技能本身可以是查新闻、读文件、建 Word 文档,甚至执行系统命令。关键在于,技能是注册进来的,不是硬编码在提示词里的,这样才能像 SkillHub 一样持续扩展。
下面我会给出可复制的目录结构、Skill 注册 JSON、Flask 接口代码和本地启动命令,并演示新增一个自定义技能后的完整调用验证。你跟着做,能跑出一个属于自己的“龙虾”骨架。
2. TaoToken 前置:给 Agent 一个稳定的模型接入层
在动手写代码之前,先把模型接入这层理清楚。iClaw 的意图解析完全依赖大模型,如果模型调用不稳定,整个 Agent 的调度就会时好时坏。我自己的做法是,把模型接入统一走一个兼容 OpenAI 协议的中转层,这样换模型、换 Key、加限流都不用改业务代码。
TaoToken 就是这样一个接入层,官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 入口是 https://taotoken.net/api 。它对外暴露的是标准的/v1/chat/completions接口,也就是说,你原来调 DeepSeek 官方接口的代码,只需要把 Base URL 和 Key 换掉就能用,模型名照填deepseek-chat即可。
为什么 Agent 项目特别需要这一层?因为 Agent 的调用模式和普通聊天不一样。普通聊天一次请求就结束了,而 Agent 一次任务可能触发多轮:先解析意图、再执行技能、再把结果交给模型总结。请求密度高、对延迟敏感、还经常需要切换不同模型做对比。如果每次都直连各家官方接口,Key 管理、额度监控、失败重试都得自己写一遍。走统一接入层,这些都能收敛到一个地方。
具体操作上,你先在 TaoToken 控制台创建一个 API Key,地址是 https://taotoken.net/api-keys ,拿到形如sk-xxxx的密钥后,把它写进环境变量,不要硬编码进代码。然后在项目里配置两个变量:TAOTOKEN_BASE_URL=https://taotoken.net/api/v1和TAOTOKEN_API_KEY=你的密钥。模型 ID 用deepseek-chat,这是 DeepSeek 的对话模型标识,意图解析和总结都够用。
如果你后面要做长期编码类 Agent,或者想让 Agent 常驻跑任务,可以了解一下 Coding Plan,地址是 https://taotoken.net/coding-plan 。它更适合高频、长时间的调用场景。想先验证模型通不通,可以直接用模型对话页面 https://taotoken.net/chat 发一句话试试,确认 Key 和网络都没问题,再回到代码里。
这里有个细节要注意:Base URL 到底带不带/v1,取决于你的 HTTP 客户端怎么拼路径。如果你用的是 OpenAI SDK,通常 Base URL 填到/api/v1,SDK 会自动补/chat/completions;如果你用 requests 手写,就要自己拼完整的https://taotoken.net/api/v1/chat/completions。我下面给的代码用的是 requests 手写方式,路径写全,避免歧义。
把这一层配好之后,iClaw 的模型调用就稳定了。接下来进入项目搭建。
3. 可复制配置:目录结构、Skill 注册 JSON 与 Flask 接口
这一节是全文的核心,所有代码都可以直接复制。先把目录结构定下来,这是 SkillHub 架构的物理基础。
iClaw/ ├── app.py # Flask 后端主程序 ├── skillhub.py # SkillHub 技能注册与调度中心 ├── skills/ │ └── skills.json # 技能注册配置 ├── Agent.md # 系统提示词 ├── templates/ │ └── index.html # 前端页面 ├── requirements.txt └── README.md先写技能注册配置skills/skills.json。每个技能用一段 JSON 描述,包含名称、描述、参数定义和执行类型。执行类型这里用command表示走系统命令,用python表示走内置函数。
{ "skills": [ { "name": "news_search", "description": "从百度新闻搜索指定关键词的新闻内容", "parameters": { "keyword": { "type": "string", "required": true, "desc": "搜索关键词" } }, "type": "command", "template": "curl.exe -L -A \"Mozilla/5.0\" \"https://news.baidu.com/ns?wd={keyword}&tn=news&cl=2&rn=10&ct=1\"" }, { "name": "file_read", "description": "读取指定文件的内容", "parameters": { "filepath": { "type": "string", "required": true, "desc": "文件路径" } }, "type": "python", "handler": "read_file" }, { "name": "word_create", "description": "创建一个 Word 文档并写入内容", "parameters": { "filename": { "type": "string", "required": true, "desc": "文档文件名" }, "content": { "type": "string", "required": true, "desc": "文档正文" }, "title": { "type": "string", "required": false, "desc": "文档标题" } }, "type": "python", "handler": "create_word" } ] }然后是skillhub.py,它负责加载 JSON、注册技能、解析调用、执行技能。核心逻辑是:把技能清单拼成一段文本塞进系统提示词,让模型知道有哪些技能可用;模型返回技能调用: name(args)格式后,这里负责解析并执行。
import json import os import re import subprocess class SkillHub: def __init__(self, config_path="skills/skills.json"): self.skills = {} self.config_path = config_path self.load_skills() def load_skills(self): if not os.path.exists(self.config_path): print(f"技能配置不存在: {self.config_path}") return with open(self.config_path, "r", encoding="utf-8") as f: data = json.load(f) for item in data.get("skills", []): self.skills[item["name"]] = item print(f"已注册技能: {item['name']}") def get_skills_prompt(self): lines = ["可用技能列表:"] for name, skill in self.skills.items(): params = ", ".join(skill.get("parameters", {}).keys()) lines.append(f"- {name}({params}): {skill['description']}") lines.append("") lines.append("调用格式:技能调用: 技能名(参数1, 参数2)") return "\n".join(lines) def parse_call(self, text): match = re.search(r"技能调用[::]\s*(\w+)\((.*?)\)", text) if not match: return None, None, None name = match.group(1) raw_args = match.group(2) args = [a.strip().strip('"').strip("'") for a in raw_args.split(",") if a.strip()] return name, args, raw_args def execute(self, name, args): if name not in self.skills: return f"技能不存在: {name}" skill = self.skills[name] if skill["type"] == "command": cmd = skill["template"] for key, val in zip(skill["parameters"].keys(), args): cmd = cmd.replace("{" + key + "}", val) return self._run_command(cmd) elif skill["type"] == "python": handler = globals().get(skill["handler"]) if handler: return handler(*args) return f"未找到处理函数: {skill['handler']}" return "未知技能类型" def _run_command(self, cmd): try: result = subprocess.run(cmd, shell=True, capture_output=True, text=True, encoding="utf-8", errors="ignore") if result.returncode == 0: return result.stdout.strip() or "命令执行成功" return f"命令执行失败: {result.stderr.strip()}" except Exception as e: return f"执行异常: {str(e)}" def read_file(filepath): try: with open(filepath, "r", encoding="utf-8") as f: return f.read() except Exception as e: return f"读取失败: {str(e)}" def create_word(filename, content, title="文档"): try: from docx import Document doc = Document() doc.add_heading(title, level=1) doc.add_paragraph(content) doc.save(filename) return f"Word 文档创建成功: {filename}" except Exception as e: return f"创建失败: {str(e)}" skillhub = SkillHub()接着是app.py,Flask 后端。它暴露三个接口:/api/chat处理对话,/api/skills列出技能,/api/skills/<name>查技能详情。模型调用走 TaoToken 的兼容接口。
import os import requests from flask import Flask, request, jsonify, render_template from skillhub import skillhub app = Flask(__name__) BASE_URL = os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api/v1") API_KEY = os.getenv("TAOTOKEN_API_KEY", "") MODEL = os.getenv("TAOTOKEN_MODEL", "deepseek-chat") def call_model(messages): headers = { "Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}" } data = {"model": MODEL, "messages": messages, "temperature": 0.3, "max_tokens": 1024} resp = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=data, timeout=60) if resp.status_code == 200: return resp.json()["choices"][0]["message"]["content"] return f"模型调用失败: {resp.status_code} - {resp.text}" def build_system_prompt(): base = "你是一个 AI Agent,负责理解用户意图并调用技能完成任务。" return base + "\n\n" + skillhub.get_skills_prompt() @app.route("/api/chat", methods=["POST"]) def chat(): user_msg = request.json.get("message", "") messages = [ {"role": "system", "content": build_system_prompt()}, {"role": "user", "content": user_msg} ] ai_reply = call_model(messages) name, args, _ = skillhub.parse_call(ai_reply) result = None if name: result = skillhub.execute(name, args) return jsonify({"reply": ai_reply, "skill": name, "result": result}) @app.route("/api/skills", methods=["GET"]) def list_skills(): return jsonify([{"name": k, "description": v["description"]} for k, v in skillhub.skills.items()]) @app.route("/api/skills/<name>", methods=["GET"]) def skill_detail(name): skill = skillhub.skills.get(name) if not skill: return jsonify({"error": "技能不存在"}), 404 return jsonify(skill) @app.route("/") def index(): return render_template("index.html") if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=True)requirements.txt里写清楚依赖:
flask>=3.0 requests>=2.31 python-docx>=0.8.11到这里,SkillHub 的注册、调度、接口三件套就齐了。技能是数据驱动的,加技能只改 JSON,不动主程序。
4. 验证请求:本地启动与新增技能调用实测
配置写完了,现在跑起来验证。先建虚拟环境、装依赖。
python -m venv venv venv\Scripts\activate pip install -r requirements.txt设置环境变量,Windows 下用set,PowerShell 用$env:。
set TAOTOKEN_BASE_URL=https://taotoken.net/api/v1 set TAOTOKEN_API_KEY=sk-你的密钥 set TAOTOKEN_MODEL=deepseek-chat启动 Flask:
python app.py看到Running on http://0.0.0.0:5000就说明起来了。先验证技能列表接口:
curl http://127.0.0.1:5000/api/skills正常返回应该是三个技能的名称和描述。如果这里返回空数组,说明skills/skills.json路径不对,检查启动目录。
接着验证对话接口,让它查新闻:
curl -X POST http://127.0.0.1:5000/api/chat -H "Content-Type: application/json" -d "{\"message\":\"帮我查一下今天的人工智能新闻\"}"预期返回里skill字段是news_search,result字段是百度新闻的搜索结果文本。如果skill是 null,说明模型没按格式返回技能调用,检查系统提示词有没有正确拼进去。
现在演示新增一个自定义技能。假设我要加一个“统计当前目录文件数量”的技能。在skills/skills.json的数组里追加一段:
{ "name": "count_files", "description": "统计当前目录下的文件数量", "parameters": {}, "type": "command", "template": "dir /b /a-d | find /c /v \"\"" }保存后重启 Flask,启动日志里会多一行已注册技能: count_files。然后发请求:
curl -X POST http://127.0.0.1:5000/api/chat -H "Content-Type: application/json" -d "{\"message\":\"当前目录有多少个文件\"}"预期模型返回类似技能调用: count_files(),后端执行dir命令统计文件数,result字段返回一个数字。这就证明新增技能被 SkillHub 正确注册并调度了,全程没改一行 Python 代码。
再验证一个带参数的技能,比如建 Word 文档:
curl -X POST http://127.0.0.1:5000/api/chat -H "Content-Type: application/json" -d "{\"message\":\"创建一个 Word 文档 iClaw说明.docx,内容是 iClaw 是一个基于 DeepSeek 的 AI Agent\"}"返回的skill应该是word_create,result显示创建成功,当前目录下会出现iClaw说明.docx。打开确认内容写入正确。
实测下来,这套流程最关键的验证点是:模型返回的技能名必须和 JSON 里注册的name完全一致,参数顺序也要对得上。只要这两点满足,调度就能跑通。
5. 本篇常见错排查:401、local proxy failed 与 choices 读取失败
跑这套 Agent 最容易卡在模型调用和技能解析两个环节。下面按真实报错逐个拆。
报错一:401 Unauthorized。返回体通常是{"error":{"message":"Invalid API key"}}。原因有三个:Key 没设置进环境变量、Key 复制时带了空格、或者 Base URL 和 Key 不匹配。排查顺序是先打印os.getenv("TAOTOKEN_API_KEY")确认非空,再确认TAOTOKEN_BASE_URL是https://taotoken.net/api/v1。注意 requests 手写方式下,完整路径是https://taotoken.net/api/v1/chat/completions,如果你把 Base URL 写成不带/v1,拼出来就是错的。修正后重启 Flask 再试。
报错二:local proxy failed 或连接超时。这类报错说明请求根本没到服务端,通常是本机网络环境或代理配置干扰。检查系统环境变量里有没有残留的HTTP_PROXY、HTTPS_PROXY,有的话在启动 Flask 前清掉。另外确认防火墙没拦 Python 的出站请求。如果是在公司内网,可能需要走内网出口,这个得找运维确认,不要自己乱配代理。
报错三:读取 choices 失败,KeyError: 'choices'。这个报错发生在resp.json()["choices"][0]这一行。原因是返回的 JSON 结构里没有choices字段,常见于两种情况:一是接口返回了错误信息但状态码是 200,二是模型名写错了。先打印完整的resp.text看返回体长什么样。如果返回的是{"error":...},那就是 Key 或模型名问题;如果返回的是流式格式,说明你误开了 stream。确认MODEL变量是deepseek-chat,且请求体里没有stream: true。
报错四:技能调用解析不到,skill 字段为 null。模型回复了内容,但parse_call没匹配上。原因是模型没严格按技能调用: name(args)格式输出。解决办法是在系统提示词里把格式要求写得更强硬,比如加一句“必须严格使用以下格式,不要添加任何其他文字”。另外正则技能调用[::]\s*(\w+)\((.*?)\)对中文冒号和英文冒号都兼容,但如果模型写成技能调用:news_search(关键词)用了中文括号,就会匹配失败。把正则里的括号也兼容一下,或者提示词里明确要求用英文括号。
报错五:技能执行了但结果为空。比如news_search返回空字符串。这通常是命令模板里的参数没替换成功。检查template里的占位符{keyword}和parameters里的键名是否完全一致,大小写敏感。另外 Windows 下curl.exe和curl行为不同,PowerShell 里curl是Invoke-WebRequest的别名,必须写curl.exe。如果命令里有中文参数,确保subprocess.run带了encoding="utf-8",否则中文会乱码导致命令失败。
报错六:Word 创建失败,ModuleNotFoundError: No module named 'docx'。这是依赖没装。注意包名是python-docx,但导入名是docx,两者不一样。执行pip install python-docx后重启 Flask。如果还报错,确认虚拟环境激活了,别装到全局 Python 里去了。
排查这类问题的通用思路是:先确认模型调用通不通(单独发一个不带技能的请求),再确认技能注册有没有成功(看启动日志),最后确认解析和执行环节。把这三段拆开定位,比一股脑看报错快得多。
6. 语义一致 CTA:把 iClaw 接到你的工作流里
到这里,一个带 SkillHub 架构的 AI Agent 骨架就跑通了。它能理解自然语言、注册技能、调度执行、回灌结果,新增技能只需要改 JSON。这套结构你可以直接拿去改造成自己的工具:把news_search换成内部 API 查询,把word_create换成报表生成,把count_files换成日志清理,Agent 的调度逻辑完全不用动。
模型接入这层,我建议你继续用 TaoToken 统一管理。创建 Key 在 https://taotoken.net/api-keys ,接入文档在 https://taotoken.net/doc ,里面有完整的接口说明和参数示例。想先验证模型效果,直接去 https://taotoken.net/chat 发几句话,确认意图解析的准确度再写进代码。如果你打算让 Agent 长期跑编码或自动化任务,Coding Plan 页面 https://taotoken.net/coding-plan 有更适合高频调用的方案。
最后留一个实用技巧:技能描述写得越具体,模型选技能的准确率越高。比如news_search的描述里加上“适用于查询实时新闻、热点事件”,模型就不会在用户问“今天有什么大事”时犹豫。技能参数的类型和必填标记也要写清楚,模型会据此决定传几个参数。这套提示词工程配合 SkillHub 的数据驱动注册,才是 OpenClaw 式 Agent 真正好用的地方。