Merge pull request #913 from datawhalechina/codex/recover-pr-875-squashed

[毕业设计] LifePlanAgent - 日程规划助手

Co-creation-projects/learner0102-LifePlanAgent/readme.md

@@ -0,0 +1,116 @@
+# Multi‑Agent 日程规划助手
+
+> 基于 HelloAgents 多智能体架构的本地日程规划系统，通过总控 Agent 调度解析、规划、校验子 Agent 完成用户日程生成与多轮修改。
+
+## 📝 项目简介
+
+问题：  
+--  传统大模型日程规划容易出现时间冲突、将任务排到已过去时刻、多轮对话丢失上下文、用户记忆无法复用等问题；  
+解决方案：  
+-- 本项目通过分工式多智能体，把输入解析、日程规划、合法性校验进行职责拆分，提升日程输出稳定性。  
+特色功能：  
+- 总控 Agent 调度多子 Agent 流水线处理；  
+- 支持多轮对话记忆，理解 “再加任务”“修改刚才任务” 这类指代式指令；  
+- 内置时间合法性校验，禁止将今日任务排到过去时刻；  
+- 支持用户记忆画像，保存用户作息、偏好、休息日；支持会话 JSON 持久化保存与加载。
+
+## ✨ 核心功能
+
+-  功能1:总控调度 Agent：负责流转调度 nl_parser、planner、validator 三个子 Agent，控制执行循环，异常自动降级兜底
+-  功能2:自然语言解析子 Agent：解析用户自然语言输入，输出标准化 Task 任务结构体，识别新建 / 修改 / 长期规划三种意图
+-  功能3:日程规划推理子 Agent：基于任务列表、用户记忆、当前真实时间生成 ScheduleItem 时间条目，禁止将今日任务排至过去时刻
+-  功能4:计划校验子 Agent：校验时间重叠、违反作息、历史时间任务等违规项，输出错误标记与提示信息
+-  功能5:本轮会话内存记忆：保存 user/assistant/tool 对话历史，多轮交互支持上下文指代
+-  功能6:用户记忆画像：独立维护用户作息、休息日、偏好配置，与聊天会话解耦
+-  功能7:会话持久化：支持将会话历史、用户记忆、最后生成计划保存 / 加载 JSON 文件
+-  功能8:异常降级机制：Agent 链路解析失败时自动调用备用 bak 逻辑，保障可用性
+
+## 🛠️ 技术栈
+
+- HelloAgents框架 : SimpleAgent、ToolRegistry 工具注册机制
+- 智能体范式: 分工流水线多 Agent（Plan‑and‑Solve 衍生，任务拆解→生成→校验闭环）
+- API: 大模型 LLM API（兼容 OpenAI 兼容接口）
+- 其他依赖库: pydantic (数据模型校验)、python‑json
+
+## 🚀 快速开始
+
+### 环境要求
+
+- Python 3.12+
+- 可用的大模型 API Key（OpenAI 兼容接口）
+
+### 安装依赖
+
+
+pip install -r requirements.txt
+
+
+### 配置API密钥
+
+
+# 创建.env文件
+cp .env.example .env
+
+# 编辑.env文件，填入你的API密钥
+
+
+### 运行项目
+
+# 打开main.py并运行
+
+
+## 📖 使用示例
+
+运行后在对话框输入测试指令，如：  
+
+- [ ] 请输入你的需求：帮我规划明天，复习90分钟  
+- 【助手回复】：已为您规划明天（9月6日，星期日）的日程。为您安排了90分钟的复习任务（09:00-10:30）。明天是您的休息日，完成任务后您可以自由安排慢跑或看书等休闲活动，注意劳逸结合，祝您周末愉快！   
+🚀 ---日程列表---  
+2026-09-06 09:00 ~ 2026-09-06 10:30 | 复习
+
+- [ ] 请输入你的需求：将复习任务修改为下午，并设置为120分钟  
+- 【助手回复】：已为您更新明天（9月6日）的计划。复习任务已调整至下午 13:30-15:30，时长更新为120分钟。完成复习后，您还有充足的下午时间进行慢跑或看书等休闲活动，祝您周末愉快！  
+🚀 ---日程列表---  
+2026-09-06 13:30 ~ 2026-09-06 15:30 | 复习
+
+- [ ] 请输入你的需求：不进行复习了，明天下午进行3小时锻炼  
+- 【助手回复】：  已为您更新明天（9月6日）的计划。取消了复习任务，为您安排了3小时的锻炼（13:30-16:30）。明天是休息日，下午锻炼后您还可以安排看书等放松活动，注意补充水分，祝您周末愉快！
+🚀 ---日程列表---  
+2026-09-06 13:30 ~ 2026-09-06 16:30 | 锻炼
+
+- [ ] 请输入你的需求：明天下午还有一个两小时的会议需要安排  
+- 【助手回复】：明天下午（13:30-18:00）的可用时间共4.5小时，无法同时安排3小时的锻炼和2小时的会议。请问您希望缩短锻炼时长、将锻炼调整至其他时间段，还是取消锻炼任务？  
+⚠️未生成有效计划，请根据提示补充信息或者重新输入。
+
+## 🎯 项目亮点
+
+- 亮点1:职责分离多 Agent 架构：解析、规划、校验各司其职，单一模块 prompt 改动不会全局影响，便于迭代调优
+- 亮点2:双重时间保护：规划阶段禁止排过去时间 + 校验 Agent 二次兜底检查，规避 “任务落在已过去时刻” 问题
+- 亮点3:完整记忆体系：内存会话记忆支持多轮指代；会话、用户画像、计划可 JSON 落盘，重启可恢复
+- 亮点4:强结构化输出：全部输出使用 Pydantic 模型校验，拒绝自由文本，降低大模型输出混乱带来的 bug
+- 亮点5:故障容错：链路解析异常自动降级 bak 备份逻辑，不会直接崩溃
+
+
+## 🔮 未来计划
+
+- [ ] 增加工具调用能力，接入日历导出、提醒生成
+- [ ] 优化长会话上下文裁剪，解决 token 无限膨胀
+- [ ] WebUI 前端界面，可视化查看 / 编辑日程
+- [ ] 用户记忆自动更新，从对话中自动提取用户作息偏好 
+
+## 🤝 贡献指南
+
+欢迎提出Issue和Pull Request！
+
+## 📄 许可证
+
+MIT License
+
+## 👤 作者
+
+- GitHub: [learner0102](https://github.com/learner0102)
+
+
+## 🙏 致谢
+
+感谢Datawhale社区和Hello-Agents项目！

Co-creation-projects/learner0102-LifePlanAgent/src/.env.example

@@ -0,0 +1,2 @@
+API_KEY = "这里填写你的LLM_API_KEY"
+BASE_URL = "这里填写你的大模型的BASE_URL"

Co-creation-projects/learner0102-LifePlanAgent/src/agent.py

@@ -0,0 +1,498 @@
+import json
+import uuid
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from hello_agents import HelloAgentsLLM, SimpleAgent, ToolRegistry
+
+from config import *
+from model import UserMemory, Task, Plan, ScheduleItem
+from prompt import *
+from tools import get_Tools
+
+
+class MainAgent:
+    def __init__(self):
+        self.llm = HelloAgentsLLM(model=LLM_MODEL, api_key=LLM_API_KEY, base_url=LLM_BASE_URL)
+        tool = get_Tools()
+        self.respostry = ToolRegistry()
+        self.respostry.register_tool(tool)
+        self.main_agent = SimpleAgent(
+            name="总控MAIN_AGENT",
+            llm=self.llm,
+            system_prompt=MAIN_AGENT_PROMPT,
+            tool_registry=self.respostry
+        )
+        self.np_parser_agent = SimpleAgent(
+            name="子Agent：自然语言解析，用户输入 → Task列表 + 意图",
+            llm=self.llm,
+            system_prompt=NL_PARSE_PROMPT,
+            tool_registry=self.respostry
+        )
+        # self.np_parser_agent.add_tool(tool)
+        self.plan_Agent = SimpleAgent(
+            name="子Agent：你是日程规划Agent，用于规划用户日程",
+            llm=self.llm,
+            system_prompt=PLAN_REASON_PROMPT,
+            tool_registry=self.respostry
+        )
+        # self.plan_Agent.add_tool(tool)
+        self.validator_agent = SimpleAgent(
+            name="子Agent：你是计划校验Agent，检查生成的日程计划是否违反用户记忆、是否存在时间冲突。",
+            llm=self.llm,
+            system_prompt=VALIDATE_PROMPT,
+            tool_registry=self.respostry
+        )
+        # self.validator_agent.add_tool(tool)
+        self.dialog_history: List[dict] = []
+
+    def _build_context_str(self, user_input: str, memory: UserMemory, current_plan: Optional[Plan], mem: Optional[UserMemory], plan: Optional[Plan]) -> str:
+        memory_json = json.dumps(memory.model_dump(), ensure_ascii=False, indent=2)
+        cp_json = json.dumps(current_plan.model_dump(), ensure_ascii=False, indent=2) if current_plan else 'null'
+        history_text = []
+        history_text.append(f"====当前系统时间====\n{datetime.now().strftime('%Y‑%m‑%d %H:%M:%S')}")
+        history_text.append("====对话历史====")
+        for msg in self.dialog_history:
+            role = msg['role']
+            cnt = msg["content"]
+            history_text.append(f"{role}: {cnt}")
+        history_text.append("====用户记忆====")
+        history_text.append(memory_json)
+        history_text.append(mem)
+        history_text.append("====当前已存在计划current_plan====")
+        history_text.append(cp_json)
+        history_text.append(plan)
+        history_text.append("====用户最新输入====")
+        history_text.append(user_input)
+        return "\n".join(history_text)
+
+    def _build_subagent_payload(self, pass_param: dict, memory: UserMemory, current_plan: Optional[Plan]) -> dict:
+        """构造子Agent输入：强制注入公共上下文，不依赖总控pass_to_sub"""
+        import datetime
+        payload = {**pass_param}
+        # 强制注入
+        payload["current_system_time"] = datetime.datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
+        payload["user_memory"] = memory.model_dump()
+        payload["current_plan"] = current_plan.model_dump() if current_plan else None
+        payload["dialog_history"] = self.dialog_history.copy()
+        return payload
+
+    def run(self, user_input: str, memory: UserMemory, current_plan: Optional[Plan], mem: Optional[UserMemory] = 'null', plan: Optional[Plan] = 'null') -> Tuple[Optional[Plan], str]:
+        try:
+            self.dialog_history.append({'role':'user','content':user_input})
+            max_loop = 10
+            loop_count = 0
+            final_plan: Optional[Plan] = None
+            final_output: str=""
+            while loop_count < max_loop:
+                loop_count += 1
+                context = self._build_context_str(user_input, memory, current_plan, mem, plan)
+                main_raw = self.main_agent.run(context)
+                main_raw = main_raw.strip().removeprefix("```json").removesuffix("```").strip()
+                try:
+                    main_desicion = json.loads(main_raw)
+                except:
+                    err_msg = f'总控Agent决策解析失败，原始输出：{main_raw[:300]}'
+                    final_output = err_msg
+                    break
+
+                call_sub = main_desicion.get("call_sub_agent", 'finish')
+                pass_param = main_desicion.get("pass_to_sub", {})
+                if call_sub == 'finish':
+                    # 总控通知结束
+                    fp_json = pass_param.get("final_plan_json")
+                    final_output = pass_param.get("output_text", "处理完成")
+                    if fp_json and fp_json != "null":
+                        try:
+                            final_plan = Plan.model_validate_json(fp_json)
+                        except Exception:
+                            final_plan = None
+                    break
+                if call_sub == 'nl_parser':
+                    sub_payload = self._build_subagent_payload(pass_param, memory, current_plan)
+                    sub_query = json.dumps(sub_payload, ensure_ascii=False, indent=2)
+                    sub_out = self.np_parser_agent.run(sub_query)
+                    self.dialog_history.append({'role': 'tool:nl_parser', "content": sub_out})
+
+                elif call_sub == 'planner':
+                    sub_payload = self._build_subagent_payload(pass_param, memory, current_plan)
+                    plan_query = json.dumps(sub_payload, ensure_ascii=False, indent=2)
+                    sub_out = self.plan_Agent.run(plan_query)
+                    self.dialog_history.append({'role': 'tool:planner', "content": sub_out})
+
+                elif call_sub == 'validator':
+                    sub_payload = self._build_subagent_payload(pass_param, memory, current_plan)
+                    sub_query = json.dumps(sub_payload, ensure_ascii=False, indent=2)
+                    sub_out = self.validator_agent.run(sub_query)
+                    self.dialog_history.append({'role': 'tool:validator', "content": sub_out})
+
+                else:
+                    final_output = f"总控调用未知子Agent:{call_sub}"
+                    break
+            else:
+                final_output = "循环已满，处理过程终止"
+            self.dialog_history.append({'role':'assistant', "content":final_output})
+            return final_plan, final_output
+
+        except Exception as e:
+            print(f"总控智能体出现错误：{str(e)}，即将使用默认模式")
+            return self.run_bak(user_input, memory, current_plan)
+
+    def run_bak(self, user_input: str, memory: UserMemory, current_plan: Optional[Plan]) -> Tuple[Optional[Plan], str]:
+        """
+        :param user_input: 用户原始输入
+        :param memory: 用户记忆对象
+        :param current_plan: 当前已存在计划（修改场景使用）
+        :return: (Plan对象 or None, 对外输出文本/需要澄清的问题)
+        """
+        # self.dialog_history.append(AgentMessage(role="user", content=user_input))
+        memory_json = json.dumps(memory.model_dump(), ensure_ascii=False, indent=2)
+        # 把用户输入 + 用户记忆拼为子Agent的user query
+        nl_user_query = f"""
+        用户记忆：
+        {memory_json}
+        用户输入：
+        {user_input}
+        
+        请输出JSON格式结果，包含
+        intent_type, target_date(YYYY‑MM‑DD，解析不出返回null), tasks, modify_operation。
+        tasks 为任务数组。
+        """
+        ### Agent 第一次调用，解析用户问题，然后生成任务列表
+        nl_result_raw = self.np_parser_agent.run(nl_user_query)
+        nl_result_raw = nl_result_raw.strip().removeprefix("```json").removesuffix("```").strip()
+        try:
+            nl_out = json.loads(nl_result_raw)
+        except json.JSONDecodeError:
+            return None, "解析用户需求失败，大模型返回格式异常，请重新描述你的需求。"
+
+        intent_type = nl_out.get("intent_type", "")
+        task_dicts = nl_out.get("tasks", [])
+        target_date: str | None = nl_out.get("target_date")
+        plan_name: str | None = nl_out.get("name_summary")
+        modify_operation = nl_out.get("modify_operation", {})
+        long_term_meta = nl_out.get("long_term_meta", {})
+        if not target_date:
+            # 兜底：取系统当前日期
+            target_date = datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
+        try:
+            task_list: List[Task] = [Task.model_validate(d) for d in task_dicts]
+        except Exception:
+            return None, "任务数据解析失败，请简化你的描述。"
+        ### 识别意图，然后决定怎么做这个任务
+        if intent_type == "new_plan":
+            task_list_json = json.dumps([t.model_dump() for t in task_list], ensure_ascii=False, indent=2)
+            plan_user_query = f"""
+            用户记忆：
+            {memory_json}
+            需要规划的目标日期：
+            {target_date}
+            待规划任务列表：
+            {task_list_json}
+
+            请输出JSON，包含 schedule_items(日程条目数组)、plan_summary。
+            日程的时间需要基于目标日期生成，时间格式 YYYY‑MM‑DD HH:MM。
+            """
+
+            plan_result_raw = self.plan_Agent.run(plan_user_query)
+            plan_result_raw = plan_result_raw.strip().removeprefix("```json").removesuffix("```").strip()
+            try:
+                plan_out = json.loads(plan_result_raw)
+            except json.JSONDecodeError:
+                return None, "生成日程失败，格式错误。"
+            schedule_dicts = plan_out.get("schedule_items", [])
+            try:
+                schedule_items: List[ScheduleItem] = [ScheduleItem.model_validate(s) for s in schedule_dicts]
+            except Exception:
+                return None, "日程条目数据解析失败。"
+            new_plan = Plan(
+                plan_id=str(uuid.uuid4()),
+                plan_name=plan_name,
+                create_time=target_date if target_date else datetime.now().strftime("%Y‑%m‑%d %H:%M"),
+                source_user_input=user_input,
+                tasks=task_list,
+                schedule_items=schedule_items,
+                plan_summary=plan_out.get("plan_summary", ""),
+                conflict_check_result="pending"
+            )
+
+            plan_json = json.dumps(new_plan.model_dump(), ensure_ascii=False, indent=2)
+            validate_user_query = f"""
+            用户记忆：
+            {memory_json}
+
+            待校验计划：
+            {plan_json}
+
+            请输出JSON：
+            {{
+              "has_error": bool,
+              "need_user_clarify": bool,
+              "clarify_question": str,
+              "error_msg": str
+            }}
+            """
+            validate_raw = self.validator_agent.run(validate_user_query)
+            validate_raw = validate_raw.strip().removeprefix("```json").removesuffix("```").strip()
+            try:
+                validate_out = json.loads(validate_raw)
+            except json.JSONDecodeError:
+                new_plan.conflict_check_result = 'unknown'
+                return new_plan, new_plan.plan_summary + "\n 校验新建环节解析异常"
+            has_error = validate_out.get("has_error", False)
+            need_user_clarify = validate_out.get("need_user_clarify", False)
+            clarify_question = validate_out.get("clarify_question", "")
+            if need_user_clarify:
+                return None, clarify_question
+            if has_error:
+                new_plan.conflict_check_result = "has_error"
+            else:
+                new_plan.conflict_check_result = "ok"
+            return new_plan, new_plan.plan_summary
+
+        elif intent_type == "modify_plan":
+            if current_plan is None:
+                return None, "当前没有已生成的计划，请先生成一份日程计划之后再执行修改。"
+            old_plan_json = json.dumps(current_plan.model_dump(), ensure_ascii=False, indent=2)
+            op_json = json.dumps(modify_operation, ensure_ascii=False, indent=2)
+            plan_user_query = f"""
+            用户记忆：
+            {memory_json}
+            目标日期：{target_date}
+            原始旧计划：
+            {old_plan_json}
+            用户修改操作：
+            {op_json}
+
+            请基于旧计划，按照修改操作重新生成完整的schedule_items。
+            输出JSON：{{"schedule_items":[...],"plan_summary":"描述本次修改后的计划"}}
+            时间格式 YYYY‑MM‑DD HH:MM。
+            """
+            plan_result_raw = self.plan_Agent.run(plan_user_query)
+            plan_result_raw = plan_result_raw.strip().removeprefix("```json").removesuffix("```").strip()
+            try:
+                plan_out = json.loads(plan_result_raw)
+            except json.JSONDecodeError:
+                return None, "修改计划失败，返回格式异常。"
+
+            schedule_dicts = plan_out.get("schedule_items", [])
+            try:
+                schedule_items: List[ScheduleItem] = [ScheduleItem.model_validate(s) for s in schedule_dicts]
+            except Exception:
+                return None, "修改后日程数据解析失败。"
+
+            merged_tasks = current_plan.tasks.copy()
+            merged_tasks.extend(task_list)
+            modified_plan = Plan(
+                plan_id=str(uuid.uuid4()),
+                plan_name=plan_name,
+                create_time=target_date if target_date else datetime.now().strftime("%Y‑%m‑%d %H:%M"),
+                source_user_input=f"[修改] {user_input}",
+                tasks=task_list,
+                schedule_items=schedule_items,
+                plan_summary=plan_out.get("plan_summary", ""),
+                conflict_check_result="pending"
+            )
+
+            plan_json = json.dumps(modified_plan.model_dump(), ensure_ascii=False, indent=2)
+            validate_user_query = f"""
+            用户记忆：
+            {memory_json}
+
+            待校验计划：
+            {plan_json}
+
+            请输出JSON：
+            {{
+              "has_error": bool,
+              "need_user_clarify": bool,
+              "clarify_question": str,
+              "error_msg": str
+            }}
+            """
+            validate_raw = self.validator_agent.run(validate_user_query)
+            validate_raw = validate_raw.strip().removeprefix("```json").removesuffix("```").strip()
+            try:
+                validate_out = json.loads(validate_raw)
+            except json.JSONDecodeError:
+                modified_plan.conflict_check_result = 'unknown'
+                return modified_plan, modified_plan.plan_summary + "\n 校验修改环节解析异常"
+            has_error = validate_out.get("has_error", False)
+            need_user_clarify = validate_out.get("need_user_clarify", False)
+            clarify_question = validate_out.get("clarify_question", "")
+            if need_user_clarify:
+                return None, clarify_question
+            if has_error:
+                modified_plan.conflict_check_result = "has_error"
+            else:
+                modified_plan.conflict_check_result = "ok"
+            return modified_plan, modified_plan.plan_summary
+
+
+        elif intent_type == "long_term_goal":
+            """
+            1.读取long_term_meta，调用 subtask_split_helper_tool 拆出多个子任务描述
+            2.把拆分得到的子任务组装成Task列表
+            3.复用new_plan整套规划+校验逻辑
+            """
+            try:
+                total_days = int(long_term_meta.get("total_days", 0))
+                daily_duration = int(long_term_meta.get("daily_duration_min", 0))
+                goal_desc = long_term_meta.get("goal_desc", "")
+            except (ValueError, TypeError):
+                return None, "长期目标参数解析失败，请明确说明总天数、每日耗时。"
+
+            if total_days <= 0 or daily_duration < 0:
+                return None, "总天数或者每日时长必须大于0"
+            try:
+                tool = self.respostry.get_tool("subtask_split_helper_tool")
+                sub_task_info = tool.run({"total_days": total_days, "goal_desc": goal_desc})
+            except:
+                print("仓库工具调用失败，进行本地调用")
+                sub_task_info = get_Tools.subtask_split_helper_tool(total_days, main_desc=goal_desc)
+            # 把拆分结果转为Task模型
+            long_tasks: List[Task] = []
+            for idx, info in enumerate(sub_task_info):
+                t = Task(
+                    task_id=f"lt_{uuid.uuid4()}",
+                    task_name=info["sub_desc"],
+                    description=f"长期目标：{goal_desc}",
+                    estimated_duration_min=daily_duration,
+                    priority="medium",
+                    deadline=None,
+                    fixed_time=None,
+                    allowed_time=None
+                )
+                long_tasks.append(t)
+
+            # 合并用户输入附带的任务 + 拆分出来长期子任务
+            full_task_list = task_list + long_tasks
+            task_list_json = json.dumps([t.model_dump() for t in full_task_list], ensure_ascii=False, indent=2)
+            plan_user_query = f"""
+            用户记忆：
+            {memory_json}
+            规划起始目标日期：{target_date}
+            长期任务列表：
+            {task_list_json}
+            请跨多天生成schedule_items，输出JSON，包含schedule_items、plan_summary。
+            时间格式 YYYY‑MM‑DD HH:MM。
+            """
+            plan_result_raw = self.plan_Agent.run(plan_user_query)
+            plan_result_raw = plan_result_raw.strip().removeprefix("```json").removesuffix("```").strip()
+            try:
+                plan_out = json.loads(plan_result_raw)
+            except json.JSONDecodeError:
+                return None, "修改计划失败，返回格式异常。"
+
+            schedule_dicts = plan_out.get("schedule_items", [])
+            try:
+                schedule_items: List[ScheduleItem] = [ScheduleItem.model_validate(s) for s in schedule_dicts]
+            except Exception:
+                return None, "长期任务日程数据解析失败。"
+
+            long_term_plan = Plan(
+                plan_id=str(uuid.uuid4()),
+                plan_name=plan_name,
+                create_time=target_date if target_date else datetime.now().strftime("%Y‑%m‑%d %H:%M"),
+                source_user_input=f"[长期任务] {user_input}",
+                tasks=task_list,
+                schedule_items=schedule_items,
+                plan_summary=plan_out.get("plan_summary", ""),
+                conflict_check_result="pending"
+            )
+
+            plan_json = json.dumps(long_term_plan.model_dump(), ensure_ascii=False, indent=2)
+            validate_user_query = f"""
+            用户记忆：
+            {memory_json}
+
+            待校验计划：
+            {plan_json}
+
+            请输出JSON：
+            {{
+              "has_error": bool,
+              "need_user_clarify": bool,
+              "clarify_question": str,
+              "error_msg": str
+            }}
+            """
+            validate_raw = self.validator_agent.run(validate_user_query)
+            validate_raw = validate_raw.strip().removeprefix("```json").removesuffix("```").strip()
+            try:
+                validate_out = json.loads(validate_raw)
+            except json.JSONDecodeError:
+                long_term_plan.conflict_check_result = 'unknown'
+                return long_term_plan, long_term_plan.plan_summary + "\n 校验长任务环节解析异常"
+            has_error = validate_out.get("has_error", False)
+            need_user_clarify = validate_out.get("need_user_clarify", False)
+            clarify_question = validate_out.get("clarify_question", "")
+            if need_user_clarify:
+                return None, clarify_question
+            if has_error:
+                long_term_plan.conflict_check_result = "has_error"
+            else:
+                long_term_plan.conflict_check_result = "ok"
+            return long_term_plan, long_term_plan.plan_summary
+
+
+        else:
+            return None, f"无法识别意图：{intent_type}, 请重新描述你的需求"
+
+    def clear_memory(self):
+        """清空本轮对话记忆，保留agent实例，适合开启新会话"""
+        self.dialog_history.clear()
+
+    def save_session(self, save_path: Path | str, memory: UserMemory, current_plan: Optional[Plan]):
+        """将会话保存到JSON文件"""
+        data = {
+            "dialog_history": self.dialog_history,
+            "user_memory": memory.model_dump(),
+            "last_plan": current_plan.model_dump() if current_plan else None
+        }
+        with open(save_path, "w", encoding="utf‑8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+
+    def load_session(self, load_path: Path | str) -> tuple[UserMemory | None, Plan | None]:
+        """从JSON恢复会话，返回(memory, current_plan)，同时填充self.dialog_history"""
+        if not Path(load_path).exists():
+            return None, None
+        with open(load_path, "r", encoding="utf‑8") as f:
+            data = json.load(f)
+        self.dialog_history = data.get("dialog_history", [])
+        mem_data = data.get("user_memory")
+        plan_data = data.get("last_plan")
+        mem = UserMemory(**mem_data) if mem_data else None
+        plan = Plan(**plan_data) if plan_data else None
+        return mem, plan
+
+
+def demo_memory() -> UserMemory:
+    return UserMemory(
+        user_id="u_001",
+        work_start="09:00",
+        work_end="18:00",
+        rest_days=["Saturday", "Sunday"],
+        avoid_time=["12:00‑13:30"],
+        preference={"priority_rule": "工作事务优先"},
+        hobbies=["慢跑", "看书"],
+        dislike=["早起高强度运动"]
+    )
+
+
+def main():
+    agent = MainAgent()
+    memory = demo_memory()
+    current_plan = None
+    user_input = "帮我规划今天的日程，我需要复习90分钟，慢跑40分钟。"
+    plan, output_text = agent.run(user_input, memory, current_plan)
+    if plan is None:
+        print(f"输出：{output_text}")
+    else:
+        print(f"\n计划摘要：{plan.plan_summary}")
+        for item in plan.schedule_items:
+            print(f"{item.start_time} ~ {item.end_time} | {item.task_name}")
+
+
+if __name__ == "__main__":
+    main()

Co-creation-projects/learner0102-LifePlanAgent/src/config.py

@@ -0,0 +1,20 @@
+from dotenv import load_dotenv
+import os
+load_dotenv()
+
+# LLM大模型配置
+LLM_MODEL = "填写你要用的大模型"
+LLM_API_KEY = os.getenv("API_KEY")
+LLM_BASE_URL = os.getenv("BASE_URL")
+LLM_TEMPERATURE = 0.3
+LLM_MAX_TOKENS = 1500
+
+# Agent系统常量
+BUFFER_MINUTE = 30  # 任务之间缓冲时间，30分钟
+MAX_SINGLE_TASK_MIN = 120  # 单次任务最大时长
+
+# 意图枚举
+INTENT_NEW_PLAN = "new_plan"
+INTENT_MODIFY_PLAN = "modify_plan"
+INTENT_QUERY_CONFLICT = "query_conflict"
+INTENT_LONG_TERM_GOAL = "long_term_goal"

Co-creation-projects/learner0102-LifePlanAgent/src/main.py

@@ -0,0 +1,59 @@
+# main.py
+from agent import MainAgent
+from model import UserMemory
+
+
+def build_demo_user_memory() -> UserMemory:
+    return UserMemory(
+        user_id="user_001",
+        work_start="08:00",
+        work_end="18:00",
+        rest_days=["Saturday", "Sunday"],
+        avoid_time=["12:00‑13:30"],
+        preference={"priority_rule": "工作类任务优先安排"},
+        hobbies=["慢跑", "看书"],
+        dislike=["早起高强度运动"]
+    )
+
+
+def main():
+    print("===== 多智能体协同日程规划助手 =====")
+    agent = MainAgent()
+    try:
+        mem, plan = agent.load_session("../data/session.json")
+    except:
+        mem, plan = None, None
+    memory = build_demo_user_memory()
+    current_plan = None
+
+    while True:
+        print("\n--------------------------------")
+        user_input = input("请输入你的需求(exit退出)：").strip()
+        if user_input.lower() in ("exit", "quit"):
+            agent.save_session("../data/session.json", memory, current_plan)
+            print("程序退出。")
+            break
+        if not user_input:
+            print("输入不能为空，请重新输入。")
+            continue
+        if mem and plan:
+            plan, output_text = agent.run(user_input, memory, current_plan, mem, plan)
+        else:
+            plan, output_text = agent.run(user_input, memory, current_plan)
+        print(f"\n【助手回复】：{output_text}")
+
+        if plan is not None:
+            current_plan = plan
+            print("\n=====📋 当前生效计划 =====")
+            print(f"计划ID：{plan.plan_id}")
+            print(f"创建时间：{plan.create_time}")
+            print(f"冲突校验结果：{plan.conflict_check_result}")
+            print("---日程列表---")
+            for item in plan.schedule_items:
+                print(f"{item.start_time} ~ {item.end_time} | {item.task_name}")
+        else:
+            print("\n⚠️未生成有效计划，请根据提示补充信息或者重新输入。")
+
+
+if __name__ == "__main__":
+    main()

Co-creation-projects/learner0102-LifePlanAgent/src/model.py

@@ -0,0 +1,60 @@
+from typing import Dict, Optional, List
+
+from pydantic import BaseModel, Field
+
+
+class UserMemory(BaseModel):
+    """用户长期记忆：保存用户基础信息、习惯、约束、历史计划反馈"""
+    user_id: str
+    work_start: str
+    work_end: str
+    rest_days: List[str]
+    avoid_time: List[str]
+    preference: Dict[str,str]
+    hobbies: List[str]
+    dislike: List[str]
+    history_task: List[Dict] = Field(default_factory=list)
+    custorm_constraints: List[Dict] = Field(default_factory=list)
+
+
+class Task(BaseModel):
+    """解析用户自然语言后得到的结构化任务实体"""
+    task_id: str
+    task_name: str
+    description: str
+    estimated_duration_min: int
+    deadline: Optional[str] = None
+    priority: str = Field(default="medium")
+    fixed_time: Optional[str] = None
+    allowed_time: Optional[List[str]] = None
+    depend_on: List[str] = Field(default_factory=list)
+
+
+class ScheduleItem(BaseModel):
+    """分配好具体起止时间的日程条目，Task经过规划之后生成"""
+    task_id: str
+    task_name: str
+    start_time: str
+    end_time: str
+    duration_min: int
+    note: str = Field(default="")
+
+
+class Plan(BaseModel):
+    """完整的一份用户生成计划"""
+    plan_id: str
+    plan_name: str
+    create_time: str
+    source_user_input: str
+    tasks: List[Task]
+    schedule_items: List[ScheduleItem]
+    plan_summary: str
+    conflict_check_result: str
+    modify_log: List[str] = Field(default_factory=list)
+
+
+class AgentMessage(BaseModel):
+    """Agent多轮对话消息结构体，用于保存对话上下文"""
+    role: str
+    content: str
+    history_text: Optional[Dict] = None

Co-creation-projects/learner0102-LifePlanAgent/src/prompt.py

@@ -0,0 +1,180 @@
+from datetime import datetime
+
+# NLParseAgent：自然语言解析prompt
+NL_PARSE_PROMPT = """
+你是自然语言解析子Agent。
+接收：用户记忆、完整对话上下文、用户最新输入
+输入payload包含字段：
+- current_system_time: 当前时间
+- user_memory: 用户记忆画像
+- current_plan: 当前已存在计划
+- dialog_history: 本轮对话历史数组，包含user/assistant/tool记录，处理指代、省略的用户请求时参考此历史。
+输出严格JSON，不要多余文字，不要markdown以外的内容。
+字段：
+{
+  "intent_type": "new_plan | modify_plan | long_term_goal",
+  "target_date": "YYYY‑MM‑DD HH:MM:SS | null",
+  "tasks": [],
+  "modify_operation": {"action":"","task_id":"","new_task":{},"remark":""},
+  "long_term_meta": {"total_days":0,"daily_duration_min":0,"goal_desc":""}
+}
+intent_type说明：
+new_plan：新建单日计划；
+modify_plan：修改已有计划；
+long_term_goal：多日长期目标。
+tasks数组输出Task的字段。解析不出target_date输出null。
+Task对象：
+- task_id: str，任务唯一标识
+- task_name: str，任务名称
+- description: str，任务描述
+- estimated_duration_min: int，预估耗时分钟
+- deadline: str|null，截止时间
+- priority: str，优先级，默认medium
+- fixed_time: str|null，固定执行时间
+- allowed_time: str[]|null，允许执行时间段
+- depend_on: str[]，依赖任务id列表
+modify_operation仅modify_plan时填充；long_term_meta仅long_term_goal填充。
+"""
+
+# PlanReasonAgent 规划推理prompt
+PLAN_REASON_PROMPT = """
+你是日程规划子Agent。
+输入：用户记忆、目标日期、任务列表 / 旧计划+修改操作 /长期任务列表
+输入payload包含字段：
+- current_system_time: 当前时间
+- user_memory: 用户记忆画像
+- current_plan: 当前已存在计划
+- dialog_history: 本轮对话历史数组，包含user/assistant/tool记录，处理指代、省略的用户请求时参考此历史。
+职责：生成合理schedule_items日程条目，遵守用户记忆的工作时间、休息、避峰时间。
+⚠️【强时间约束，必须遵守】
+1. 当前系统时间会作为输入给到你。
+2. 如果目标日期就是**今天**：
+   - 绝对不允许生成 start_time 早于当前系统时间的任务片段；
+   - 已经过去的时刻不能安排任何任务；
+   - 如果今日剩余可用时间不足以放下全部任务，不要强行把任务塞进过去时间；plan_summary中告知用户：部分任务今日时间不足，建议延后到后续日期。
+3. 遵守用户记忆的工作时间、休息日、午休避峰时间约束。
+4. 任务duration_min要和任务estimated_duration_min保持一致。
+输出JSON：
+{
+    "schedule_items":[],
+    "plan_summary": "计划摘要，给用户阅读"
+}
+schedule_items每一条包含start_time、end_time、task_name；时间格式 YYYY‑MM‑DD HH:MM。
+ScheduleItem对象：
+- task_id: str，关联任务ID
+- task_name: str，任务名称
+- start_time: str，开始时间，格式YYYY‑MM‑DD HH:MM
+- end_time: str，结束时间，格式YYYY‑MM‑DD HH:MM
+- duration_min: int，持续分钟数
+- note: str，备注
+如果是modify_plan，基于旧计划，按照modify_operation修改后全部重生成schedule_items。
+如果是long_term_goal，跨多天排布任务。
+"""
+
+# ValidateAgent 校验prompt
+VALIDATE_PROMPT = """
+你是计划校验子Agent。
+输入：current_system_time、用户记忆 + 完整Plan JSON。
+输入payload包含字段：
+- current_system_time: 当前时间
+- user_memory: 用户记忆画像
+- current_plan: 当前已存在计划
+- dialog_history: 本轮对话历史数组，包含user/assistant/tool记录，处理指代、省略的用户请求时参考此历史。
+检查项：
+1. 时间冲突：多个schedule_items时间段重叠；
+2. 是否违反用户工作/休息、避峰时间段；
+3. ⚠️时间合法性：如果计划是今日任务，不允许存在start_time早于current_system_time的条目；存在则标记has_error=true，并在error_msg写明“存在已经过去的时间点的任务”。
+
+输出JSON：
+{
+    "has_error": bool,
+    "need_user_clarify": bool,
+    "clarify_question": "",
+    "error_msg": ""
+}
+need_user_clarify=true代表需要向用户追问补充信息；
+has_error=true代表存在冲突/违规。
+
+只输出JSON，允许```json包裹，禁止额外文字。
+"""
+
+
+# MainAgent调度提示词
+MAIN_AGENT_PROMPT = """
+你是**总控调度智能体**，负责调度下面3个子智能体完成用户日程规划任务。
+子智能体列表：
+1. nl_parser：自然语言解析子Agent，输入用户对话、用户记忆，输出解析后的意图、任务、目标日期、修改操作、长期目标元数据JSON。
+2. planner：日程规划子Agent，输入任务、旧计划、目标日期、修改指令，输出schedule_items日程列表与plan_summary。
+3. validator：校验子Agent，输入完整Plan对象，检查时间冲突、用户记忆约束，输出是否冲突、是否需要向用户澄清。
+
+【对象字段说明，仅用于你理解数据结构，禁止输出Python class代码】
+Plan对象：
+- plan_id: str，计划唯一ID
+- plan_name: str，计划名称
+- create_time: str，创建时间
+- source_user_input: str，来源用户输入
+- tasks: Task[]，任务列表
+- schedule_items: ScheduleItem[]，日程时间条目
+- plan_summary: str，计划摘要
+- conflict_check_result: str，冲突校验结果
+- modify_log: str[]，修改记录
+
+Task对象：
+- task_id: str，任务唯一标识
+- task_name: str，任务名称
+- description: str，任务描述
+- estimated_duration_min: int，预估耗时分钟
+- deadline: str|null，截止时间
+- priority: str，优先级，默认medium
+- fixed_time: str|null，固定执行时间
+- allowed_time: str[]|null，允许执行时间段
+- depend_on: str[]，依赖任务id列表
+
+ScheduleItem对象：
+- task_id: str，关联任务ID
+- task_name: str，任务名称
+- start_time: str，开始时间，格式YYYY‑MM‑DD HH:MM
+- end_time: str，结束时间，格式YYYY‑MM‑DD HH:MM
+- duration_min: int，持续分钟数
+- note: str，备注
+
+你的工作规则：
+1. 输入上下文包含：当前系统时间、对话历史、用户记忆、当前已存在计划current_plan、用户最新输入。
+2. 对话历史中，role为 tool:nl_parser / tool:planner / tool:validator 的消息，是子Agent返回的原始输出，**你必须读取、复制其中JSON内容填入pass_to_sub，严禁凭空编造子Agent输出**。
+3. 你**只输出JSON**，允许被```json ```代码块包裹；禁止输出任何思考、解释、自然语言、Python代码。
+
+输出JSON固定结构：
+{
+    "call_sub_agent": "nl_parser | planner | validator | finish",
+    "pass_to_sub": {},
+    "remark": "内部思考备注，仅调试用"
+}
+
+字段行为说明：
+- call_sub_agent = "nl_parser"：需要解析用户输入意图，把全部上下文放入pass_to_sub。
+- call_sub_agent = "planner"：需要生成/修改计划，**从tool:nl_parser消息读取解析结果，把解析结果、memory、current_plan全部放入pass_to_sub**。
+- call_sub_agent = "validator"：需要校验生成好的plan，**从tool:planner消息读取完整plan json，把plan完整json与memory放入pass_to_sub**。
+- call_sub_agent = "finish"：全部流程结束。pass_to_sub必须包含两个key：
+  1. final_plan_json：Plan完整JSON字符串；没有有效计划则填字符串"null"，**绝对不能省略该字段**。
+  2. output_text：展示给用户的回复文本。
+> ️重要：finish时不允许只填写output_text而丢弃final_plan_json，即使output_text内容很完整，也必须带上final_plan_json。
+
+流转逻辑：
+1. 用户新输入，优先调用 nl_parser；
+2. 读取 tool:nl_parser 的输出解析结果，根据intent_type(new_plan/modify_plan/long_term_goal)调用planner；
+3. 读取 tool:planner 的输出计划，调用validator做冲突校验；
+4. 读取 tool:validator 返回结果后：
+   - 如果need_user_clarify=true：直接finish，final_plan_json="null"，output_text为澄清提问；
+   - 否则finish，把planner输出的完整plan填入final_plan_json，同时填写output_text。
+
+边界规则：
+- intent_type=modify_plan，但current_plan为null：finish，final_plan_json="null"，output_text="当前没有已生成的计划，请先生成一份日程计划之后再执行修改。"
+- long_term_goal要注意参数合法性，total_days、daily_duration_min必须大于0，非法则finish返回错误提示，final_plan_json="null"。
+- 如果任意子Agent返回JSON解析失败，直接finish，final_plan_json="null"，output_text输出错误提示。
+
+硬性约束：
+1. 不要自己生成日程、不要自己解析任务，全部交给子Agent完成。
+2. 所有子Agent输出来自对话历史tool:*消息，禁止编造。
+3. finish输出必须同时存在final_plan_json、output_text两个key，final_plan_json不可缺失。
+4. 禁止输出Python class定义、禁止输出markdown以外的附加文本。
+"""

Co-creation-projects/learner0102-LifePlanAgent/src/requirements.txt

@@ -0,0 +1,4 @@
+python==3.12
+hello_agents==0.2.8
+pydantic==2.13.5
+python-dotenv==1.2.3

Co-creation-projects/learner0102-LifePlanAgent/src/tools.py

@@ -0,0 +1,222 @@
+from datetime import datetime
+from typing import List, Dict, Tuple, Any, Optional
+from hello_agents.tools import Tool
+from hello_agents.tools.base import ToolParameter
+from model import ScheduleItem, Task, UserMemory
+
+
+def parse_time(time_str: str) -> datetime:
+    return datetime.strptime(time_str, "%Y-%m-%d %H:%M")
+
+
+class get_Tools(Tool):
+    """
+    多种工具可以供给Agent使用：
+    - time_conflict_detect：检测日程冲突，返回冲突信息
+    - task_priority_sort：按优先级 high>medium>low 排序任务
+    - time_available_calc：计算目标日期可用时间段
+    - task_duration_check：检查任务是否超过单次最大时长
+    - subtask_split_helper：长周期任务拆分辅助工具，返回阶段描述
+    """
+
+    def __init__(self, name: str = "schedule_tools", description: Optional[str] = None):
+        """
+        初始化日程工具集
+
+        Args:
+            name: 工具名称
+            description: 工具描述
+        """
+        if description is None:
+            description = self._generate_description()
+
+        super().__init__(name=name, description=description)
+
+    def _generate_description(self) -> str:
+        """生成工具描述"""
+        return """日程管理工具集，提供以下功能：
+        1. time_conflict_detect：检测日程冲突
+        2. task_priority_sort：按优先级排序任务
+        3. time_available_calc：计算可用时间段
+        4. task_duration_check：检查任务时长
+        5. subtask_split_helper：拆分长周期任务
+        
+        调用格式：{"action": "工具名", "arguments": {...}}"""
+
+    def get_parameters(self) -> List[ToolParameter]:
+        """获取工具参数定义"""
+        return [
+            ToolParameter(
+                name="action",
+                type="string",
+                description="操作类型: time_conflict_detect, task_priority_sort, time_available_calc, task_duration_check, subtask_split_helper",
+                required=True
+            ),
+            ToolParameter(
+                name="arguments",
+                type="object",
+                description="各工具对应的参数",
+                required=True
+            )
+        ]
+
+    def run(self, parameters: Dict[str, Any]) -> str:
+        """
+        执行工具操作
+
+        Args:
+            parameters: 包含 action 和 arguments 的字典
+                - action: 工具名称
+                - arguments: 工具参数
+
+        Returns:
+            操作结果（字符串格式）
+        """
+        action = parameters.get("action", "").lower()
+        arguments = parameters.get("arguments", {})
+
+        if not action:
+            return "错误：必须指定 action 参数"
+
+        try:
+            if action == "time_conflict_detect":
+                return self._format_conflict_result(
+                    self.time_conflict_detect_tool(**arguments)
+                )
+            elif action == "task_priority_sort":
+                result = self.task_priority_sort_tool(**arguments)
+                return self._format_priority_result(result)
+            elif action == "time_available_calc":
+                result = self.time_available_calc_tool(**arguments)
+                return self._format_available_time_result(result)
+            elif action == "task_duration_check":
+                result = self.task_duration_check_tool(**arguments)
+                return self._format_duration_check_result(result)
+            elif action == "subtask_split_helper":
+                result = self.subtask_split_helper_tool(**arguments)
+                return self._format_subtask_result(result)
+            else:
+                return f"错误：不支持的操作 '{action}'，支持的操作: time_conflict_detect, task_priority_sort, time_available_calc, task_duration_check, subtask_split_helper"
+        except Exception as e:
+            return f"执行 '{action}' 失败: {str(e)}"
+
+    # ========== 工具方法 ==========
+
+    def time_conflict_detect_tool(self, schedule_items: List[ScheduleItem]) -> Dict:
+        """检测日程冲突，返回冲突信息"""
+        conflicts = []
+        for i in range(len(schedule_items)):
+            start_time_i = parse_time(schedule_items[i].start_time)
+            end_time_i = parse_time(schedule_items[i].end_time)
+            for j in range(i + 1, len(schedule_items)):
+                start_time_j = parse_time(schedule_items[j].start_time)
+                end_time_j = parse_time(schedule_items[j].end_time)
+                if start_time_i < end_time_j and end_time_i > start_time_j:
+                    conflicts.append({
+                        "task_a_id": schedule_items[i].task_id,
+                        "task_b_id": schedule_items[j].task_id,
+                        "task_a_name": schedule_items[i].task_name,
+                        "task_b_name": schedule_items[j].task_name,
+                        "note": f"{schedule_items[i].start_time} ~ {schedule_items[i].end_time} 与 {schedule_items[j].start_time} ~ {schedule_items[j].end_time} 存在时间重叠"
+                    })
+        return {
+            "has_conflict": len(conflicts) > 0,
+            "conflict_pairs": conflicts
+        }
+
+    def task_priority_sort_tool(self, task_list: List[Task], memory: UserMemory) -> List[Task]:
+        """按优先级 high>medium>low 排序任务"""
+        prio_map = {"high": 0, "medium": 1, "low": 2}
+        return sorted(task_list, key=lambda x: prio_map.get(x.priority, 1))
+
+    def time_available_calc_tool(self, memory: UserMemory, target_date_str: str) -> List[Tuple[str, str]]:
+        """
+        计算目标日期可用时间段
+        target_date_str: "2026-09-05"
+        返回 [(start_time_str, end_time_str), ...]，格式 "2026-09-05 HH:MM"
+        """
+        base_date = datetime.strptime(target_date_str, "%Y-%m-%d")
+        work1_start_time = base_date.replace(hour=8, minute=0)
+        work1_end_time = base_date.replace(hour=12, minute=0)
+        work2_start_time = base_date.replace(hour=14, minute=0)
+        work2_end_time = base_date.replace(hour=18, minute=0)
+        slots = [
+            (work1_start_time.strftime("%Y-%m-%d %H:%M"), work1_end_time.strftime("%Y-%m-%d %H:%M")),
+            (work2_start_time.strftime("%Y-%m-%d %H:%M"), work2_end_time.strftime("%Y-%m-%d %H:%M"))
+        ]
+        return slots
+
+    def task_duration_check_tool(self, task_list: List[Task], max_time: int) -> Dict:
+        """检查任务是否超过单次最大时长"""
+        over_list = []
+        for task in task_list:
+            if task.estimated_duration_min > max_time:
+                over_list.append({
+                    "task_id": task.task_id,
+                    "task_name": task.task_name,
+                    "task_duration_time": task.estimated_duration_min
+                })
+        return {
+            "is_over": len(over_list) > 0,
+            "over_tasks": over_list
+        }
+
+    def subtask_split_helper_tool(self, total_days: int, main_desc: str) -> List[Dict]:
+        """长周期任务拆分辅助工具，返回阶段描述"""
+        chunk = total_days // 5
+        res = []
+        for i in range(5):
+            res.append({"sub_desc": f"{main_desc} 阶段{i + 1}", "days": chunk})
+        return res
+
+    # ========== 格式化方法（用于返回字符串结果） ==========
+
+    def _format_conflict_result(self, result: Dict) -> str:
+        """格式化冲突检测结果"""
+        if not result["has_conflict"]:
+            return "✅ 没有检测到日程冲突"
+
+        output = f"⚠️ 发现 {len(result['conflict_pairs'])} 个冲突：\n"
+        for idx, conflict in enumerate(result['conflict_pairs'], 1):
+            output += f"{idx}. {conflict['note']}\n"
+        return output
+
+    def _format_priority_result(self, result: List[Task]) -> str:
+        """格式化优先级排序结果"""
+        if not result:
+            return "没有任务需要排序"
+
+        output = "任务按优先级排序（高>中>低）：\n"
+        for idx, task in enumerate(result, 1):
+            output += f"{idx}. {task.task_name} (优先级: {task.priority}, 预计: {task.estimated_duration_min}分钟)\n"
+        return output
+
+    def _format_available_time_result(self, result: List[Tuple[str, str]]) -> str:
+        """格式化可用时间段结果"""
+        if not result:
+            return "当天没有可用时间段"
+
+        output = "当天可用时间段：\n"
+        for idx, (start, end) in enumerate(result, 1):
+            output += f"{idx}. {start} ~ {end}\n"
+        return output
+
+    def _format_duration_check_result(self, result: Dict) -> str:
+        """格式化时长检查结果"""
+        if not result["is_over"]:
+            return "✅ 所有任务都在最大时长限制内"
+
+        output = f"⚠️ 发现 {len(result['over_tasks'])} 个任务超过最大时长：\n"
+        for idx, task in enumerate(result['over_tasks'], 1):
+            output += f"{idx}. {task['task_name']}: {task['task_duration_time']}分钟\n"
+        return output
+
+    def _format_subtask_result(self, result: List[Dict]) -> str:
+        """格式化子任务拆分结果"""
+        if not result:
+            return "无法拆分任务"
+
+        output = "任务拆分阶段：\n"
+        for idx, phase in enumerate(result, 1):
+            output += f"{idx}. {phase['sub_desc']}: {phase['days']}天\n"
+        return output