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

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()