From f7dd74ea20a74aa2cff9dd9b0b73ca6870529956 Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Mon, 1 Sep 2025 22:54:38 +0800 Subject: [PATCH] feat(cookbook): implement task memory functionality - Rename and rebrand experience-related variables and functions to task memory - Update Appworld and FrozenLake agents to use task memory instead of experience - Add new functions for handling API responses, deleting workspaces, and dumping/loading memories - Modify run scripts to incorporate task memory creation and usage - Update logging and print statements to reflect new task memory terminology --- cookbook/appworld/appworld_react_agent.py | 97 +++++++++++----- cookbook/appworld/run_appworld.py | 109 ++++++++++++++++-- cookbook/frozenlake/frozenlake_react_agent.py | 79 ++++++------- cookbook/frozenlake/run_frozenlake.py | 74 ++++++------ 4 files changed, 244 insertions(+), 115 deletions(-) diff --git a/cookbook/appworld/appworld_react_agent.py b/cookbook/appworld/appworld_react_agent.py index 118e546b..19174350 100644 --- a/cookbook/appworld/appworld_react_agent.py +++ b/cookbook/appworld/appworld_react_agent.py @@ -36,9 +36,9 @@ class AppworldReactAgent: max_interactions: int = 30, max_response_size: int = 2048, num_runs: int = 1, - use_experience: bool = False, - make_experience: bool = False, - exp_url: str = "http://0.0.0.0:8001/", + use_task_memory: bool = False, + make_task_memory: bool = False, + api_url: str = "http://0.0.0.0:8002/", workspace_id: str="appworld_v1"): self.index: int = index @@ -49,9 +49,9 @@ class AppworldReactAgent: self.max_interactions: int = max_interactions self.max_response_size: int = max_response_size self.num_runs: int = num_runs - self.use_experience: bool = use_experience - self.make_experience: bool = make_experience - self.exp_url = exp_url + self.use_task_memory: bool = use_task_memory + self.make_task_memory: bool = make_task_memory + self.api_url = api_url self.workspace_id = workspace_id self.llm_client = OpenAI() @@ -75,10 +75,10 @@ class AppworldReactAgent: return "call llm error" def prompt_messages(self,world: AppWorld) -> list[dict]: - if self.use_experience: - experience = self.get_experience(world.task.instruction) - logger.info(f"loaded experience: {experience}") - dictionary = {"supervisor": world.task.supervisor, "instruction": world.task.instruction, "experience": experience} + if self.use_task_memory: + task_memory = self.get_task_memory(world.task.instruction) + logger.info(f"loaded task_memory: {task_memory}") + dictionary = {"supervisor": world.task.supervisor, "instruction": world.task.instruction, "experience": task_memory} else: dictionary = {"supervisor": world.task.supervisor, "instruction": world.task.instruction ,"experience": ""} print(dictionary) @@ -144,30 +144,75 @@ class AppworldReactAgent: } result.append(t_result) - if self.make_experience: - self.make_experience(result) + if self.make_task_memory: + memory_list = self.make_task_memory(result) + logger.info(f"Created {len(memory_list) if memory_list else 0} task memories") return result - def get_experience(self, query: str): - response = requests.post(url=self.exp_url + "retriever", json={ - "workspace_id": self.workspace_id, - "query": query, - "top_k": 5 - }) - + def handle_api_response(self, response: requests.Response): + """Handle API response with proper error checking""" if response.status_code != 200: + print(f"Error: {response.status_code}") print(response.text) + return None + + return response.json() + + def get_task_memory(self, query: str): + """Retrieve relevant task memories based on a query""" + response = requests.post( + url=f"{self.api_url}retrieve_task_memory", + json={ + "workspace_id": self.workspace_id, + "query": query, + } + ) + + result = self.handle_api_response(response) + if not result: return "" - response = response.json() - print(response) - experience_merged: str = response["experience_merged"] - print(f"experience_merged={experience_merged}") - return experience_merged + # Extract and return the answer + answer = result.get("answer", "") + print(f"Retrieved task memory: {answer}") + return answer - def make_experience(self, result): - pass + def make_task_memory(self, result): + """Generate a summary of conversation messages and create task memories""" + if not result: + print("No results to summarize") + return + + # Prepare trajectories from results + trajectories = [] + for r in result: + if "task_history" in r: + trajectories.append({ + "messages": r["task_history"], + "score": float(r.get("uplift_score", 0.0)) + }) + + if not trajectories: + print("No trajectories to summarize") + return + + response = requests.post( + url=f"{self.api_url}summary_task_memory", + json={ + "workspace_id": self.workspace_id, + "trajectories": trajectories + } + ) + + result = self.handle_api_response(response) + if not result: + return + + # Extract memory list from response + memory_list = result.get("metadata", {}).get("memory_list", []) + print(f"Task memory list created: {len(memory_list)} memories") + return memory_list def main(): diff --git a/cookbook/appworld/run_appworld.py b/cookbook/appworld/run_appworld.py index d16bcd9b..a3dcb7db 100644 --- a/cookbook/appworld/run_appworld.py +++ b/cookbook/appworld/run_appworld.py @@ -1,5 +1,6 @@ import os import time +import requests import ray from ray import logger @@ -17,7 +18,64 @@ from appworld import load_task_ids from appworld_react_agent import AppworldReactAgent -def run_agent(dataset_name: str, experiment_suffix: str, max_workers: int, num_runs: int = 1, use_experience: bool = False, workspace_id: str="appworld", exp_url: str = "http://0.0.0.0:8001/") : +def handle_api_response(response: requests.Response): + """Handle API response with proper error checking""" + if response.status_code != 200: + print(f"Error: {response.status_code}") + print(response.text) + return None + + return response.json() + + +def delete_workspace(workspace_id: str, api_url: str = "http://0.0.0.0:8002/"): + """Delete the current workspace from the vector store""" + response = requests.post( + url=f"{api_url}vector_store", + json={ + "workspace_id": workspace_id, + "action": "delete", + } + ) + + result = handle_api_response(response) + if result: + print(f"Workspace '{workspace_id}' deleted successfully") + + +def dump_memory(workspace_id: str, path: str = "./", api_url: str = "http://0.0.0.0:8002/"): + """Dump the vector store memories to disk""" + response = requests.post( + url=f"{api_url}vector_store", + json={ + "workspace_id": workspace_id, + "action": "dump", + "path": path, + } + ) + + result = handle_api_response(response) + if result: + print(f"Memory dumped to {path}") + + +def load_memory(workspace_id: str, path: str = "./", api_url: str = "http://0.0.0.0:8002/"): + """Load memories from disk into the vector store""" + response = requests.post( + url=f"{api_url}vector_store", + json={ + "workspace_id": workspace_id, + "action": "load", + "path": path, + } + ) + + result = handle_api_response(response) + if result: + print(f"Memory loaded from {path}") + + +def run_agent(dataset_name: str, experiment_suffix: str, max_workers: int, num_runs: int = 1, use_task_memory: bool = False, make_task_memory: bool = False, workspace_id: str="appworld_v1", api_url: str = "http://0.0.0.0:8002/") : experiment_name = dataset_name + "_" + experiment_suffix path: Path = Path(f"./exp_result") path.mkdir(parents=True, exist_ok=True) @@ -39,9 +97,10 @@ def run_agent(dataset_name: str, experiment_suffix: str, max_workers: int, num_r task_ids=worker_task_ids, experiment_name=experiment_name, num_runs=num_runs, - use_experience=use_experience, + use_task_memory=use_task_memory, + make_task_memory=make_task_memory, workspace_id=workspace_id, - exp_url=exp_url) + api_url=api_url) future = actor.execute.remote() future_list.append(future) time.sleep(1) @@ -64,7 +123,10 @@ def run_agent(dataset_name: str, experiment_suffix: str, max_workers: int, num_r task_ids=[task_id], experiment_name=experiment_name, num_runs=num_runs, - use_experience=use_experience) + use_task_memory=use_task_memory, + make_task_memory=make_task_memory, + workspace_id=workspace_id, + api_url=api_url) task_results = agent.execute() if isinstance(task_results, list): result.extend(task_results) @@ -75,18 +137,43 @@ def run_agent(dataset_name: str, experiment_suffix: str, max_workers: int, num_r def main(): max_workers = 8 - num_runs = 1 # Run each task 4 times + num_runs = 1 # Run each task once + workspace_id = "appworld_v1" + api_url = "http://0.0.0.0:8002/" + if max_workers > 1: ray.init(num_cpus=8) - - logger.info("Start running experiments without experience") + + # Clean up workspace before starting + logger.info("Deleting workspace...") + delete_workspace(workspace_id=workspace_id, api_url=api_url) + + # First run to build task memories + logger.info("Start running experiments to build task memories") + run_agent(dataset_name="dev", experiment_suffix="build-memory", + max_workers=max_workers, num_runs=1, + use_task_memory=False, make_task_memory=True, + workspace_id=workspace_id, api_url=api_url) + + # Dump memories to disk for persistence + logger.info("Dumping memories to disk...") + dump_memory(workspace_id=workspace_id, api_url=api_url) + + # Run experiments without task memory + logger.info("Start running experiments without task memory") for i in range(num_runs): - run_agent(dataset_name="dev", experiment_suffix=f"no-exp", max_workers=max_workers, num_runs=1, - use_experience=False, workspace_id="appworld_v1") + run_agent(dataset_name="dev", experiment_suffix=f"no-memory", + max_workers=max_workers, num_runs=1, + use_task_memory=False, make_task_memory=False, + workspace_id=workspace_id, api_url=api_url) - logger.info("Start running experiments with experience") + # Run experiments with task memory + logger.info("Start running experiments with task memory") for i in range(num_runs): - run_agent(dataset_name="dev", experiment_suffix=f"add-exp", max_workers=max_workers, num_runs=1, use_experience=True,workspace_id="appworld_v1") + run_agent(dataset_name="dev", experiment_suffix=f"with-memory", + max_workers=max_workers, num_runs=1, + use_task_memory=True, make_task_memory=False, + workspace_id=workspace_id, api_url=api_url) diff --git a/cookbook/frozenlake/frozenlake_react_agent.py b/cookbook/frozenlake/frozenlake_react_agent.py index 2e215db6..4f6538d5 100644 --- a/cookbook/frozenlake/frozenlake_react_agent.py +++ b/cookbook/frozenlake/frozenlake_react_agent.py @@ -32,7 +32,7 @@ class GameResult: @ray.remote class FrozenLakeReactAgent: - """A ReAct Agent for FrozenLake game with experience learning.""" + """A ReAct Agent for FrozenLake game with task memory learning.""" def __init__(self, index: int, @@ -42,8 +42,8 @@ class FrozenLakeReactAgent: temperature: float = 0.7, max_steps: int = 50, num_runs: int = 1, - use_experience: bool = False, - make_experience: bool = False): + use_task_memory: bool = False, + make_task_memory: bool = False): self.index = index self.task_configs = task_configs @@ -52,8 +52,8 @@ class FrozenLakeReactAgent: self.temperature = temperature self.max_steps = max_steps self.num_runs = num_runs - self.use_experience = use_experience - self.make_experience = make_experience + self.use_task_memory = use_task_memory + self.make_task_memory = make_task_memory self.llm_client = OpenAI() self.action_map = {0: "LEFT", 1: "DOWN", 2: "RIGHT", 3: "UP"} @@ -121,35 +121,34 @@ class FrozenLakeReactAgent: else: return self.prompts["frozenlake_sys_prompt_no_slippery"] - def get_experience(self, map_desc: str, is_slippery: bool) -> str: - """Retrieve relevant experience from experience service""" - if not self.use_experience: + def get_task_memory(self, map_desc: str, is_slippery: bool) -> str: + """Retrieve relevant task memory from task memory service""" + if not self.use_task_memory: return "" try: query = f"FrozenLake game map: {map_desc}, slippery: {is_slippery}" - base_url = "http://0.0.0.0:8001/" + base_url = "http://0.0.0.0:8002/" workspace_id = self.experiment_name response = requests.post( - url=base_url + "retriever", + url=base_url + "retrieve_task_memory", json={ "workspace_id": workspace_id, "query": query, - "top_k": 3 }, timeout=60 ) if response.status_code == 200: data = response.json() - return data.get("experience_merged", "") + return data.get("answer", "") else: - logger.warning(f"Experience retrieval failed: {response.status_code}") + logger.warning(f"Task memory retrieval failed: {response.status_code}") return "" except Exception as e: - logger.warning(f"Failed to get experience: {e}") + logger.warning(f"Failed to get task memory: {e}") return "" def action_parser(self, response: str) -> int: @@ -194,19 +193,19 @@ class FrozenLakeReactAgent: env = gym.make("FrozenLake-v1", **env_kwargs) - # Get map description for experience + # Get map description for task memory map_str = '\n'.join([''.join([cell.decode('utf-8') for cell in row]) for row in env.unwrapped.desc]) # Build messages system_prompt = self.build_system_prompt(is_slippery) - experience = self.get_experience(map_str, is_slippery) + task_memory = self.get_task_memory(map_str, is_slippery) messages = [{"role": "system", "content": system_prompt}] - if experience: - exp_content = f"Here are some relevant tips from previous successful games:\n\n{experience}\n\nUse these tips to help you succeed." - messages.append({"role": "user", "content": exp_content}) + if task_memory: + memory_content = f"Here are some relevant tips from previous successful games:\n\n{task_memory}\n\nUse these tips to help you succeed." + messages.append({"role": "user", "content": memory_content}) messages.append( {"role": "assistant", "content": "I'll use these tips to navigate the frozen lake successfully."}) @@ -279,56 +278,54 @@ class FrozenLakeReactAgent: "map_id": map_id, "is_slippery": is_slippery, "map_size": map_size, - "use_experience": self.use_experience + "use_task_memory": self.use_task_memory } ) return result, messages - def save_experience(self, results: List[GameResult], messages_list: List[List[Dict]]): - """Save successful trajectories as experience""" - if not self.make_experience: + def save_task_memory(self, results: List[GameResult], messages_list: List[List[Dict]]): + """Save successful trajectories as task memory""" + if not self.make_task_memory: return - trajs = [] + trajectories = [] for result, messages in zip(results, messages_list): if result.success: - # Create trajectory for experience service + # Create trajectory for task memory service traj = { "messages": messages, - "query" : result.map_config["map_desc"], "score": 1.0, # Success } - trajs.append(traj) + trajectories.append(traj) else: traj = { "messages": messages, - "query" : result.map_config["map_desc"], - "score": 0.0, # Success + "score": 0.0, # Failure } - trajs.append(traj) + trajectories.append(traj) - if trajs: + if trajectories: try: - base_url = "http://0.0.0.0:8001/" + base_url = "http://0.0.0.0:8002/" workspace_id = self.experiment_name response = requests.post( - url=base_url + "summarizer", + url=base_url + "summary_task_memory", json={ "workspace_id": workspace_id, - "traj_list": trajs + "trajectories": trajectories }, timeout=300 ) if response.status_code == 200: - logger.info(f"Saved {len(trajs)} trajectories as experience") + logger.info(f"Saved {len(trajectories)} trajectories as task memory") else: - logger.warning(f"Failed to save experience: {response.status_code}") + logger.warning(f"Failed to save task memory: {response.status_code}") except Exception as e: - logger.error(f"Error saving experience: {e}") + logger.error(f"Error saving task memory: {e}") def execute(self) -> List[Dict]: """Execute all tasks""" @@ -357,9 +354,9 @@ class FrozenLakeReactAgent: } all_results[-1] = result_dict - # Save experience if needed - if self.make_experience: - # Convert back to GameResult objects for experience saving + # Save task memory if needed + if self.make_task_memory: + # Convert back to GameResult objects for task memory saving game_results = [] for i, result_dict in enumerate(all_results): game_result = GameResult( @@ -374,6 +371,6 @@ class FrozenLakeReactAgent: ) game_results.append(game_result) - self.save_experience(game_results, all_messages) + self.save_task_memory(game_results, all_messages) return all_results \ No newline at end of file diff --git a/cookbook/frozenlake/run_frozenlake.py b/cookbook/frozenlake/run_frozenlake.py index 7be7117c..66f30fb8 100644 --- a/cookbook/frozenlake/run_frozenlake.py +++ b/cookbook/frozenlake/run_frozenlake.py @@ -13,7 +13,7 @@ from map_manager import MapManager def generate_training_configs(num_maps: int = 20, map_size: int = 4, is_slippery: bool=False) -> List[Dict]: - """Generate random maps for training/experience generation""" + """Generate random maps for training/task memory generation""" configs = [] for i in range(num_maps): @@ -46,15 +46,15 @@ def generate_test_configs(num_test_maps: int = 100, is_slippery: bool = False) - map_desc = np.array([list(row) for row in map_data["map_desc"]], dtype='c') map_id = map_data["map_id"] - for use_exp in [True, False]: + for use_memory in [True, False]: config = { "task_type": "test", "map_desc": map_desc, "map_size": 4, "is_slippery": is_slippery, - "use_experience": use_exp, + "use_task_memory": use_memory, "map_id": map_id, - "task_id": f"test_map{map_id}_slip{is_slippery}_exp{use_exp}" + "task_id": f"test_map{map_id}_slip{is_slippery}_mem{use_memory}" } configs.append(config) @@ -63,8 +63,8 @@ def generate_test_configs(num_test_maps: int = 100, is_slippery: bool = False) - def train(experiment_name: str, max_workers: int = 2, num_runs: int = 3, num_training_maps= 15, is_slippery: bool= False) -> None: - """Phase 1: Generate experience from random maps""" - logger.info("๐ŸŽฏ Starting Training Phase - Generating Experience") + """Phase 1: Generate task memory from random maps""" + logger.info("๐ŸŽฏ Starting Training Phase - Generating Task Memory") logger.info("=" * 60) training_configs = generate_training_configs(num_maps=num_training_maps, map_size=4, is_slippery=is_slippery) @@ -91,8 +91,8 @@ def train(experiment_name: str, max_workers: int = 2, num_runs: int = 3, num_tra task_configs=worker_configs, experiment_name=experiment_name, num_runs=num_runs, - use_experience=False, # No experience in training phase - make_experience=True,# Generate experience + use_task_memory=False, # No task memory in training phase + make_task_memory=True, # Generate task memory ) future = agent.execute.remote() future_list.append(future) @@ -115,8 +115,8 @@ def train(experiment_name: str, max_workers: int = 2, num_runs: int = 3, num_tra task_configs=training_configs, experiment_name=experiment_name, num_runs=num_runs, - use_experience=False, - make_experience=True + use_task_memory=False, + make_task_memory=True ) results = agent.execute() dump_results() @@ -131,7 +131,7 @@ def train(experiment_name: str, max_workers: int = 2, num_runs: int = 3, num_tra def test(experiment_name: str, max_workers: int = 2, num_runs: int = 5, num_test_maps: int = 100, is_slippery: bool=False) -> None: - """Phase 2: Test on fixed maps with/without experience""" + """Phase 2: Test on fixed maps with/without task memory""" logger.info("๐Ÿงช Starting Test Phase - Evaluating Performance") logger.info(f"๐Ÿ“Š Testing on {num_test_maps} maps with {num_runs} runs each") logger.info("=" * 60) @@ -140,12 +140,12 @@ def test(experiment_name: str, max_workers: int = 2, num_runs: int = 5, num_test path = Path("./exp_result") path.mkdir(parents=True, exist_ok=True) - # Group configs by experience usage for separate experiments - exp_configs = [c for c in test_configs if c.get("use_experience", False)] - no_exp_configs = [c for c in test_configs if not c.get("use_experience", False)] + # Group configs by task memory usage for separate experiments + memory_configs = [c for c in test_configs if c.get("use_task_memory", False)] + no_memory_configs = [c for c in test_configs if not c.get("use_task_memory", False)] - logger.info(f"๐Ÿ“ Configs without experience: {len(no_exp_configs)}") - logger.info(f"๐Ÿ“ Configs with experience: {len(exp_configs)}") + logger.info(f"๐Ÿ“ Configs without task memory: {len(no_memory_configs)}") + logger.info(f"๐Ÿ“ Configs with task memory: {len(memory_configs)}") @@ -156,37 +156,37 @@ def test(experiment_name: str, max_workers: int = 2, num_runs: int = 5, num_test f.write(json.dumps(result) + "\n") logger.info(f"๐Ÿ’พ Test results saved to {output_file}") - # Test without experience first - logger.info("๐Ÿšซ Testing WITHOUT experience...") + # Test without task memory first + logger.info("๐Ÿšซ Testing WITHOUT task memory...") all_results = [] - results_no_exp = run_test_configs( - configs=no_exp_configs, + results_no_memory = run_test_configs( + configs=no_memory_configs, experiment_name=experiment_name, max_workers=max_workers, num_runs=num_runs, - use_experience=False + use_task_memory=False ) - all_results.extend(results_no_exp) - dump_results("no_exp") + all_results.extend(results_no_memory) + dump_results("no_memory") - # Test with experience - logger.info("โœ… Testing WITH experience...") + # Test with task memory + logger.info("โœ… Testing WITH task memory...") all_results = [] - results_with_exp = run_test_configs( - configs=exp_configs, + results_with_memory = run_test_configs( + configs=memory_configs, experiment_name=experiment_name, max_workers=max_workers, num_runs=num_runs, - use_experience=True + use_task_memory=True ) - all_results.extend(results_with_exp) - dump_results("with_exp") + all_results.extend(results_with_memory) + dump_results("with_memory") return all_results def run_test_configs(configs: List[Dict], experiment_name: str, max_workers: int, - num_runs: int, use_experience: bool) -> List[Dict]: + num_runs: int, use_task_memory: bool) -> List[Dict]: """Run a set of test configurations""" results = [] @@ -200,8 +200,8 @@ def run_test_configs(configs: List[Dict], experiment_name: str, max_workers: int task_configs=worker_configs, experiment_name=experiment_name, num_runs=num_runs, - use_experience=use_experience, - make_experience=False + use_task_memory=use_task_memory, + make_task_memory=False ) future = agent.execute.remote() future_list.append(future) @@ -219,8 +219,8 @@ def run_test_configs(configs: List[Dict], experiment_name: str, max_workers: int task_configs=configs, experiment_name=experiment_name, num_runs=num_runs, - use_experience=use_experience, - make_experience=False + use_task_memory=use_task_memory, + make_task_memory=False ) results = agent.execute() @@ -258,8 +258,8 @@ def main(): is_slippery=is_slippery ) - # Wait a bit for experience service to process - logger.info("โฐ Waiting for experience service to process data...") + # Wait a bit for task memory service to process + logger.info("โฐ Waiting for task memory service to process data...") time.sleep(10)