让 AI 成为个人助理：JiuwenClaw 在日程规划中的应用实践

让 AI 成为个人助理：JiuwenClaw 在日程规划中的应用实践

一、背景

在日常开发中，任务管理常常让人疲于应付。待办事项散落在飞书任务、本地 Todo 列表、会议纪要甚至同事的一句“回头处理一下”里。等到周会复盘时，才惊觉好几个关键事项早已被遗漏。问题不在于我们不做事，而在于信息太分散、优先级太模糊、时间安排又缺乏全局视角——结果就是忙而无效。

真正能提升效率的任务系统，得能自动聚合多源任务，用可量化的逻辑判断轻重缓急，并在排期冲突或截止临近时主动提醒。该文基于JiuwenClaw，从实际工程出发，拆解如何构建这样一个智能调度系统：包括统一采集架构、基于依赖与紧急度的排序引擎、飞书 OAuth 集成与 Token 自动刷新机制，以及通过心跳驱动的定时调度能力。

二、JiuwenClaw 相关介绍

2.1 JiuwenClaw 核心介绍

**JiuwenClaw是一款基于Python开发的智能AI Agent，**正如其名——“Claw” 象征着精准的抓取与连接。它能够将大语言模型的强大能力，通过你日常使用的各类通讯应用，直接延伸至你的指尖。**JiuwenClaw 并非传统的聊天机器人，而是一款本地优先、云端适配的AI自动化代理——它以大语言模型 为“大脑”，以Skills插件生态为“手脚”，能理解自然语言指令，自主完成网页操作、文档处理、多平台协同等具象化任务，无需编写复杂的自动化脚本，**其技能系统非常适合构建任务调度工具。

2.2 JiuwenClaw 目录结构

workspace/agent/skills/smart-scheduler/ ├── SKILL.md # 技能定义文件 ├── run_scheduler.py # 统一入口脚本 ├── models/ │ ├── task.py # 任务模型 │ ├── schedule.py # 日程模型 │ └── time_slot.py # 时间槽模型 ├── collectors/ │ ├── task_collector.py # 任务采集器 │ ├── dependency_analyzer.py # 依赖分析器 │ └── history_analyzer.py # 历史分析器 ├── analyzers/ │ ├── priority_scorer.py # 优先级评分 │ ├── time_estimator.py # 时间预估 │ └── conflict_detector.py # 冲突检测器 ├── generators/ │ ├── schedule_generator.py # 日程生成器 │ └── schedule_formatter.py # 格式化输出 ├── integrations/ │ └── feishu_client.py # 飞书客户端 └── services/ ├── reminder_service.py # 提醒服务 ├── reschedule_service.py # 重调度服务 └── feedback_service.py # 学习反馈服务 

三、智能任务调度器技术方案

3.1 分层架构设计

智能任务调度器采用五层架构，实现多源任务采集、依赖分析、优先级评分与冲突检测，最终生成可执行日程并推送至飞书与Web端，提升任务管理效率。

3.2 任务处理流程

智能任务调度器流程通过多源采集、优先级分析与日程生成，自动整合待办事项并推送到飞书，支持用户预览确认，实现高效、智能的日程规划与提醒。

四、Skills 技能系统工程化设计

智能任务调度器以“技能”（Skills）为核心构建单元，将任务采集、优先级计算、冲突检测、日程生成等能力模块化封装。每个 Skill 独立可插拔、可组合，既保证逻辑清晰，又支持灵活扩展，为智能调度提供工程化、可维护的架构基础。

4.1 Skills 目录结构

workspace/agent/skills/smart-scheduler/ ├── SKILL.md ├── README.md ├── run_scheduler.py ├── config.yaml ├── models/ │ ├── __init__.py │ ├── task.py │ ├── schedule.py │ └── time_slot.py ├── collectors/ │ ├── __init__.py │ ├── task_collector.py │ └── dependency_analyzer.py ├── analyzers/ │ ├── __init__.py │ ├── priority_scorer.py │ └── conflict_detector.py ├── generators/ │ ├── __init__.py │ └── schedule_generator.py ├── integrations/ │ ├── __init__.py │ └── feishu_client.py └── services/ ├── __init__.py └── reminder_service.py 

4.2 SKILL.md 技能定义

--- name: smart-scheduler version: "1.0.0" description: 智能任务调度器 - 采集、分析、生成日程安排 allowed_tools: - read_file - write_file - list_directory - web_search --- # 智能任务调度器 根据用户指令生成日程安排。 ## 功能 1. **任务采集**：从 TodoToolkit、飞书任务、记忆系统采集任务 2. **依赖分析**：分析任务依赖关系，生成执行顺序 3. **优先级评分**：综合优先级、紧急程度计算评分 4. **冲突检测**：检测时间重叠、资源冲突 5. **日程生成**：生成可执行的时间安排 6. **飞书推送**：推送日程到飞书 ## 使用方式 - "生成今日日程" - "规划本周任务" - "检查任务冲突" ## 输出格式 生成 Markdown 格式的日程报告。 

五、JiuwenClaw 本地快速部署

5.1 环境要求

JiuwenClaw 本地部署需要满足：

5.2 安装方式

1. 源码安装

# 克隆仓库git clone https://atomgit.com/openJiuwen/jiuwenclaw.git cd jiuwenclaw # 安装依赖（推荐使用 uv） uv sync# 或使用 pip pip install-e.

2. PyPI 安装

pip install jiuwenclaw # 首次使用需初始化工作空间 jiuwenclaw-init 

5.3 配置说明

1. 环境变量（**<font>.env</font>**）

在项目根目录创建 <font>.env</font> 文件，填入模型与飞书相关配置：

# 必填：大模型服务MODEL_PROVIDER=OpenAI MODEL_NAME=deepseek-chat API_BASE=https://api.modelarts-maas.com/openai/v1 API_KEY=sk-xxxxxxxxx # 可选：向量嵌入服务EMBED_API_BASE=EMBED_API_KEY=EMBED_MODEL=# 可选：飞书应用凭证FEISHU_APP_ID=FEISHU_APP_SECRET=

注意点：所有环境变量均可在 <font>config/config.yaml</font> 中通过 <font>${VAR_NAME}</font> 引用。

2. 系统配置（**<font>config/config.yaml</font>**）

关键配置项示例：

react:model_name: ${MODEL_NAME:-deepseek-chat}model_client_config:api_base: ${API_BASE}api_key: ${API_KEY}heartbeat:every:3600# 心跳调度间隔（秒）target: web # 日程推送目标：web 或 feishuactive_hours:start:"08:00"end:"22:00"# 仅在此时间段内触发调度

5.4 启动服务

首次启动 JiuwenClaw 需要执行一下 jiuwenclaw-init

若使用源码模式，可通过 <font>uv</font> 直接运行：

uv run jiuwenclaw-start all 

服务启动成功后，打开浏览器访问 <font>http://localhost:5173</font>，即可与 JiuwenClaw Agent 交互，开始智能任务调度体验。

六、任务采集与飞书集成

6.1 任务采集器实现

任务采集器负责从多个数据源采集任务，位于 collectors/task_collector.py：

# collectors/task_collector.pyfrom dataclasses import dataclass from datetime import date, datetime from pathlib import Path from typing import Any import json import sys from models.task import Task, TaskPriority, TaskStatus @dataclassclassCollectorConfig:"""采集器配置""" workspace_dir: Path |None=None session_id:str|None=NoneclassTaskCollector:"""任务数据采集器"""def__init__(self, config: CollectorConfig |None=None): self.config = config or CollectorConfig() self.workspace_dir = self.config.workspace_dir or Path.home()/".jiuwenclaw"defcollect_all_tasks(self, include_feishu:bool=True)->list[Task]:"""采集所有任务""" tasks:list[Task]=[]# 1. 从 TodoToolkit 采集 todo_tasks = self._collect_from_todokit() tasks.extend(todo_tasks)# 2. 从飞书采集if include_feishu: feishu_tasks = self._collect_from_feishu() tasks.extend(feishu_tasks)# 去重（基于 title） seen_titles =set() unique_tasks =[]for task in tasks:if task.title notin seen_titles: seen_titles.add(task.title) unique_tasks.append(task)return unique_tasks def_collect_from_todokit(self)->list[Task]:"""从 TodoToolkit 的 todo.json 采集任务""" tasks:list[Task]=[] todo_file = self.workspace_dir /"todo.json"ifnot todo_file.exists():return tasks try:withopen(todo_file,"r", encoding="utf-8")as f: data = json.load(f)for item in data.get("todos",[]): task = self._parse_todo_item(item)if task: tasks.append(task)except Exception:passreturn tasks def_parse_todo_item(self, item:dict[str, Any])-> Task |None:"""解析 todo.json 中的条目"""try: priority_map ={"CRITICAL": TaskPriority.CRITICAL,"HIGH": TaskPriority.HIGH,"MEDIUM": TaskPriority.MEDIUM,"LOW": TaskPriority.LOW,} priority = priority_map.get( item.get("priority","MEDIUM").upper(), TaskPriority.MEDIUM ) due_date =Noneif item.get("due_date"):try: due_date = date.fromisoformat(item["due_date"])except ValueError:passreturn Task(id=item.get("id",""), title=item.get("title",""), description=item.get("description",""), priority=priority, status=TaskStatus.PENDING, due_date=due_date, estimated_minutes=item.get("estimated_minutes",60), dependencies=item.get("dependencies",[]), tags=item.get("tags",[]), source="local",)except Exception:returnNone

6.2 飞书 OAuth 认证实现

这是整个项目踩坑最多的部分。飞书用户任务属于个人隐私数据，必须通过 OAuth 授权获取 user_access_token。

授权脚本 (feishu_oauth.py)

# feishu_oauth.py"""飞书用户授权脚本 用法: python feishu_oauth.py # 交互模式 python feishu_oauth.py --yes # 跳过确认直接执行 """import os import sys import json import webbrowser import time from urllib.parse import urlencode if sys.platform =="win32": sys.stdout.reconfigure(encoding='utf-8')from dotenv import load_dotenv from pathlib import Path import httpx from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse, parse_qs import threading # 加载环境变量 load_dotenv(Path(__file__).parent /".env") APP_ID = os.getenv("FEISHU_APP_ID","") APP_SECRET = os.getenv("FEISHU_APP_SECRET","") REDIRECT_URI ="http://localhost:8080/callback"# 全局变量存储结果 result_data ={"code":None,"done":False}classCallbackHandler(BaseHTTPRequestHandler):"""OAuth 回调处理器"""defdo_GET(self): parsed = urlparse(self.path)if parsed.path =="/callback": params = parse_qs(parsed.query) code = params.get("code",[None])[0] result_data["code"]= code result_data["done"]=True self.send_response(200) self.send_header("Content-type","text/html; charset=utf-8") self.end_headers()if code: msg ="<html><body><h1>✓ 授权成功!</h1><p>请返回终端查看结果。</p></body></html>"else: msg ="<html><body><h1>✗ 授权失败</h1></body></html>" self.wfile.write(msg.encode("utf-8"))deflog_message(self,*args):passdefmain():print("="*50)print("飞书用户授权")print("="*50)# 构建授权 URL - 必须包含 scope 参数 auth_params ={"app_id": APP_ID,"redirect_uri": REDIRECT_URI,"state":"feishu_oauth_"+str(int(time.time())),# 关键：必须请求任务权限"scope":"task:task:readonly contact:user.base:readonly"} auth_url =f"https://open.feishu.cn/open-apis/authen/v1/authorize?{urlencode(auth_params)}"print(f"\n授权 URL: {auth_url}")# 启动本地服务器 server = HTTPServer(("localhost",8080), CallbackHandler) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start()# 打开浏览器 webbrowser.open(auth_url)print("\n请在浏览器中完成授权...")# 等待回调 timeout =120 start = time.time()whilenot result_data["done"]and(time.time()- start)< timeout: time.sleep(0.5) server.shutdown()ifnot result_data["done"]:print("\n超时! 请重试。")return code = result_data["code"]ifnot code:print("\n授权失败!")returnprint(f"\n✓ 获取到授权码: {code[:20]}...")# 换取 user_access_token# 1. 先获取 app_access_token app_token_resp = httpx.post("https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal", json={"app_id": APP_ID,"app_secret": APP_SECRET}, timeout=30) app_access_token = app_token_resp.json().get("app_access_token")# 2. 换取 user_access_token token_resp = httpx.post("https://open.feishu.cn/open-apis/authen/v1/oidc/access_token", headers={"Authorization":f"Bearer {app_access_token}"}, json={"grant_type":"authorization_code","code": code,"redirect_uri": REDIRECT_URI,}, timeout=30) token_data = token_resp.json()if token_data.get("code")!=0:print(f"错误: {token_data}")return user_token = token_data["data"]["access_token"] refresh_token = token_data["data"]["refresh_token"]print(f"\n✓ 用户令牌获取成功!")print(f" Access Token: {user_token[:30]}...")print(f" Refresh Token: {refresh_token[:30]}...")# 保存到文件 result ={"access_token": user_token,"refresh_token": refresh_token,"created_at": time.strftime("%Y-%m-%d %H:%M:%S")}withopen("feishu_user_token.json","w", encoding="utf-8")as f: json.dump(result, f, ensure_ascii=False, indent=2)print(f"\n✓ 令牌已保存到: feishu_user_token.json")if __name__ =="__main__": main()

6.3 Token 自动刷新机制

飞书 user_access_token 只有 2 小时有效期，需要实现自动刷新：

# integrations/feishu_client.py（部分代码）from datetime import datetime from pathlib import Path import json import httpx classFeishuClient:"""飞书客户端 - 支持 Token 自动刷新"""asyncdef_get_user_access_token(self)->str|None:"""获取用户访问令牌，如果快过期则自动刷新""" token = self.config.user_access_token ifnot token:returnNone# 检查 token 是否快过期 token_file = Path("feishu_user_token.json")if token_file.exists():withopen(token_file,"r", encoding="utf-8")as f: token_data = json.load(f) saved_time = token_data.get("created_at")if saved_time: created = datetime.strptime(saved_time,"%Y-%m-%d %H:%M:%S") elapsed =(datetime.now()- created).total_seconds()# 提前 4 分钟刷新if elapsed >7000:# 约 1 小时 56 分钟 new_token =await self._refresh_user_token()if new_token:return new_token return token asyncdef_refresh_user_token(self)->str|None:"""使用 refresh_token 刷新访问令牌"""ifnot self.config.user_refresh_token:returnNoneprint("正在刷新飞书用户令牌...")# 1. 获取 app_access_token app_token_resp =await self._client.post("https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal", json={"app_id": self.config.app_id,"app_secret": self.config.app_secret,}) app_access_token = app_token_resp.json().get("app_access_token")# 2. 刷新 token refresh_resp =await self._client.post("https://open.feishu.cn/open-apis/authen/v1/oidc/refresh_access_token", headers={"Authorization":f"Bearer {app_access_token}"}, json={"grant_type":"refresh_token","refresh_token": self.config.user_refresh_token,}) data = refresh_resp.json()if data.get("code")!=0:print(f"刷新失败: {data.get('msg')}")returnNone new_access_token = data["data"]["access_token"] new_refresh_token = data["data"].get("refresh_token")# 3. 保存新 token token_file = Path("feishu_user_token.json")withopen(token_file,"r", encoding="utf-8")as f: old_data = json.load(f) old_data["access_token"]= new_access_token if new_refresh_token: old_data["refresh_token"]= new_refresh_token old_data["created_at"]= datetime.now().strftime("%Y-%m-%d %H:%M:%S")withopen(token_file,"w", encoding="utf-8")as f: json.dump(old_data, f, ensure_ascii=False, indent=2)print("✓ 飞书用户令牌已刷新")return new_access_token 

Token 有效期说明

七、依赖分析与优先级排序

7.1 依赖分析器实现

# collectors/dependency_analyzer.pyfrom dataclasses import dataclass from models.task import Task @dataclassclassDependencyEdge:"""依赖边""" predecessor_id:str# 前置任务 successor_id:str# 后继任务classDependencyAnalyzer:"""依赖分析器 - 使用 Kahn 算法实现拓扑排序"""def__init__(self): self.nodes:list[Task]=[] self.edges:list[DependencyEdge]=[] self.adjacency:dict[str,list[str]]={}defbuild_graph(self, tasks:list[Task])->None:"""构建依赖图""" self.nodes = tasks self.edges =[] self.adjacency ={task.id:[]for task in tasks} task_ids ={task.idfor task in tasks}for task in tasks:for dep_id in task.dependencies:if dep_id in task_ids: self.edges.append(DependencyEdge(dep_id, task.id)) self.adjacency[dep_id].append(task.id)defget_execution_order(self, tasks:list[Task])->list[str]:"""获取执行顺序（拓扑排序）""" self.build_graph(tasks)# 计算入度 in_degree:dict[str,int]={task.id:0for task in self.nodes}for edge in self.edges: in_degree[edge.successor_id]= in_degree.get(edge.successor_id,0)+1# 入度为 0 的节点入队 queue =[task_id for task_id, degree in in_degree.items()if degree ==0] result:list[str]=[] task_map ={t.id: t for t in self.nodes}while queue:# 同等条件下优先执行高优先级任务 queue.sort( key=lambda x: task_map.get(x, Task(id=x, title="")).priority.value, reverse=True) current = queue.pop(0) result.append(current)for successor_id in self.adjacency.get(current,[]): in_degree[successor_id]-=1if in_degree[successor_id]==0: queue.append(successor_id)return result defdetect_cycles(self)->list[list[str]]:"""检测循环依赖""" cycles:list[list[str]]=[] visited:set[str]=set() rec_stack:set[str]=set()defdfs(node_id:str, path:list[str]): visited.add(node_id) rec_stack.add(node_id) path.append(node_id)for successor_id in self.adjacency.get(node_id,[]):if successor_id notin visited: dfs(successor_id, path)elif successor_id in rec_stack: cycle_start = path.index(successor_id) cycles.append(path[cycle_start:]+[successor_id]) path.pop() rec_stack.remove(node_id)for task in self.nodes:if task.idnotin visited: dfs(task.id,[])return cycles 

7.2 冲突检测器实现

# analyzers/conflict_detector.pyfrom dataclasses import dataclass from datetime import date from models.task import Task from models.time_slot import TimeSlot @dataclassclassConflict:"""冲突信息""" conflict_type:str# time_overlap, deadline_conflict severity:str# error, warning task_ids:list[str] message:str suggestion:str=""classConflictDetector:"""冲突检测器"""defdetect_time_conflicts(self, slots:list[TimeSlot])->list[Conflict]:"""检测时间重叠""" conflicts:list[Conflict]=[] sorted_slots =sorted(slots, key=lambda s: s.start_time)for i inrange(len(sorted_slots)):for j inrange(i +1,len(sorted_slots)):if sorted_slots[i].overlaps(sorted_slots[j]): conflicts.append(Conflict( conflict_type="time_overlap", severity="warning", task_ids=[sorted_slots[i].task_id, sorted_slots[j].task_id], message=f"时间重叠: {sorted_slots[i].start_time:%H:%M} 和 {sorted_slots[j].start_time:%H:%M}",))return conflicts defdetect_deadline_conflicts( self, tasks:list[Task], available_minutes:int, target_date: date )->list[Conflict]:"""检测截止日期冲突""" urgent_tasks =[ t for t in tasks if t.due_date and t.due_date <= target_date and t.status.value in["pending","in_progress"]] total_required =sum(t.estimated_minutes for t in urgent_tasks)if total_required > available_minutes:return[Conflict( conflict_type="deadline_conflict", severity="error", task_ids=[t.idfor t in urgent_tasks], message=f"紧急任务需 {total_required} 分钟，可用仅 {available_minutes} 分钟", suggestion="建议延期低优先级任务")]return[]

八、功能实测演示

8.1 飞书 OAuth 授权

运行 <font>feishu_oauth.py</font> 后，程序会生成一个授权链接：

https://open.feishu.cn/open-apis/authen/v1/authorize?app_id=cli_xxx... 

8.2 功能展示

本地todo任务解析：

飞书任务获取测试：

# 测试飞书任务获取 $ python -c " import asyncio from integrations.feishu_client import FeishuClient, FeishuConfig import json withopen('feishu_user_token.json','r')as f: token_data = json.load(f) config = FeishuConfig( user_access_token=token_data['access_token'], user_refresh_token=token_data['refresh_token'],)asyncdeftest(): client = FeishuClient(config) tasks =await client.get_tasks()print(f'获取到 {len(tasks)} 个飞书任务:')for t in tasks:print(f' - {t.get(\"summary\", \"N/A\")}')await client.close() asyncio.run(test()) " 

获取到的任务记忆至系统：

web端演示：

总结

飞书与 JiuwenClaw 智能任务调度器结合，无需手动整理待办、反复协调日程，即可实现高效、自动化的个人与团队任务管理。通过 AI 理解任务指令、自动采集多源待办、分析依赖关系并生成合理日程，将原本碎片化、易遗漏的任务管理流程，转变为“一句话交代 → 自动规划 → 冲突预警 → 结果同步”的闭环体验，将“被动响应任务”升级为“主动掌控节奏”。让用户从琐碎协调中解放出来，把精力真正投入到执行与决策中，特别适合事务繁杂的职场人或追求执行力的团队。

如果你也想开发类似的智能体，可以了解了解 jiuwenclaw，非常强大！

openJiuwen 官网：https://openJiuwen.com

框架源码：https://atomgit.com/openJiuwen/jiuwenclaw