diff --git a/.gitignore b/.gitignore index acb57448..f08317e3 100644 --- a/.gitignore +++ b/.gitignore @@ -35,4 +35,5 @@ experiencemaker/cookbook/bfcl/no_exp_result experiencemaker/cookbook/bfcl/data experiencemaker/cookbook/bfcl/gorilla experiencemaker/*.sh -experiencemaker/file_vector_store \ No newline at end of file +experiencemaker/file_vector_store +experiencemaker/*.egg-info \ No newline at end of file diff --git a/experiencemaker/cookbook/bfcl/bfcl_agent.py b/experiencemaker/cookbook/bfcl/bfcl_agent.py index b134fd05..b1ffa5d7 100644 --- a/experiencemaker/cookbook/bfcl/bfcl_agent.py +++ b/experiencemaker/cookbook/bfcl/bfcl_agent.py @@ -64,6 +64,10 @@ class BFCLAgent: enable_thinking: bool = False, use_experience: bool = False, use_fixed_experience: bool = True, + use_experience_deletion: bool = False, + delete_freq: int = 10, + freq_threshold: int = 5, + utility_threshold: float = 0.5, experience_base_url: str = "http://0.0.0.0:8001/", experience_workspace_id: str = "bfcl_8b_0725"): @@ -80,11 +84,16 @@ class BFCLAgent: self.num_runs: int = num_runs self.enable_thinking: bool = enable_thinking self.use_experience: bool = use_experience - self.use_fixed_experience: bool = use_fixed_experience + self.use_fixed_experience: bool = use_fixed_experience if use_experience else True + self.use_experience_deletion: bool = use_experience_deletion + self.delete_freq: int = delete_freq + self.freq_threshold: int = freq_threshold + self.utility_threshold: float = utility_threshold self.experience_base_url: str = experience_base_url self.experience_workspace_id: str = experience_workspace_id self.history: List[List[List[dict]]] = [[] for _ in range(num_runs)] + self.retrieved_experience_ids: List[List[List[str]]] = [[] for _ in range(num_runs)] self.test_entry: List[List[Dict[str, Any]]] = [[] for _ in range(num_runs)] self.original_test_entry: List[List[Dict[str, Any]]] = [[] for _ in range(num_runs)] self.tool_schema: List[List[List[dict]]] = [[] for _ in range(num_runs)] @@ -104,8 +113,16 @@ class BFCLAgent: msg = self.test_entry[run_id][i].get("messages", [])[0] if self.use_experience: query = msg["content"] - exp = self.get_experience(query) - self.history[run_id].append([self.get_query_with_experience(query, exp)]) + response = self.get_experience(query) + if len(response["experience_list"]): + self.retrieved_experience_ids[run_id].append([e["experience_id"] for e in response["experience_list"]]) + exp: str = response["experience_merged"] + print(f"experience_merged={exp}") + self.history[run_id].append([self.get_query_with_experience(query, exp)]) + self.update_experience_freq(self.retrieved_experience_ids[run_id][i]) + else: + self.retrieved_experience_ids[run_id].append([]) + self.history[run_id].append([{"role": "user","content": "Task:\n" + query + "\n"}]) else: self.history[run_id].append([msg]) self.current_turn[run_id][i] = 1 @@ -125,24 +142,48 @@ class BFCLAgent: logger.info(f"query:{query}") if response.status_code != 200: - print(response.text) + logger.info(response.text) return "" response = response.json() - print(response) - experience_merged: str = response["experience_merged"] - print(f"experience_merged={experience_merged}") - return experience_merged + logger.info(response) + return response - def update_experience(self, trajectories): + def add_experience(self, trajectories): response = requests.post(url=self.experience_base_url + "summarizer", json={ "workspace_id": self.experience_workspace_id, "traj_list": trajectories, }) response.raise_for_status() response = response.json() - print(f"add new experiences: {response["experience_list"]}") + logger.info(f"add new experiences: {response["experience_list"]}") + def update_experience_freq(self, experience_ids): + response = requests.post(url=self.experience_base_url + "vector_store", json={ + "workspace_id": self.experience_workspace_id, + "action": "update_freq", + "experience_ids": experience_ids, + }) + response.raise_for_status() + logger.info(response.json()) + + def update_experience_utility(self, experience_ids): + response = requests.post(url=self.experience_base_url + "vector_store", json={ + "workspace_id": self.experience_workspace_id, + "action": "update_utility", + "experience_ids": experience_ids, + }) + response.raise_for_status() + + def delete_experience(self): + response = requests.post(url=self.experience_base_url + "vector_store", json={ + "workspace_id": self.experience_workspace_id, + "action": "utility_based_delete", + "freq_threshold": self.freq_threshold, + "utility_threshold": self.utility_threshold + }) + response.raise_for_status() + def call_llm(self, messages: list, tool_schemas: list[dict]) -> str: for i in range(100): try: @@ -475,6 +516,7 @@ class BFCLAgent: def execute(self): result = [] + counter = 0 for task_index, task_id in enumerate(tqdm(self.task_ids, desc=f"ray_index={self.index}")): for run_id in range(self.num_runs): try: @@ -520,12 +562,21 @@ class BFCLAgent: reward = self.get_reward(run_id, task_index) if reward == 1 and not self.use_fixed_experience: - self.update_experience([{ + # selectively add experiences when succeed + self.add_experience([{ "task_id":task_id, "messages":self.history[run_id][task_index], "score":reward - }]) # selectively add experiences when succeed - + }]) + + if len(self.retrieved_experience_ids[run_id][task_index]): + # update the utility-related attributes of retrieved experiences + self.update_experience_utility(self.retrieved_experience_ids[run_id][task_index]) + + counter += 1 + if self.use_experience_deletion and counter % self.delete_freq == 0: + self.delete_experience() + t_result = { "run_id": run_id, "task_id": self.task_ids[task_index], diff --git a/experiencemaker/cookbook/bfcl/run_bfcl.py b/experiencemaker/cookbook/bfcl/run_bfcl.py index 8796dcd8..a1a9e3f2 100644 --- a/experiencemaker/cookbook/bfcl/run_bfcl.py +++ b/experiencemaker/cookbook/bfcl/run_bfcl.py @@ -22,7 +22,11 @@ def run_agent(dataset_name: str, data_path: str = "data/multiturn_data_base_val.jsonl", answer_path: Path = Path("data/possible_answer"), use_experience: bool = False, - use_fixed_experience: bool = True, + use_fixed_experience: bool = True, + use_experience_deletion: bool = False, + delete_freq: int = 10, + freq_threshold: int = 5, + utility_threshold: float = 0.5, enable_thinking: bool = False, experience_base_url: str = "http://0.0.0.0:8001/", experience_workspace_id: str = "bfcl_8b_0725"): @@ -40,81 +44,69 @@ def run_agent(dataset_name: str, for x in result: f.write(json.dumps(x) + "\n") - if max_workers > 1: - future_list: list = [] - for i in range(max_workers): - actor = BFCLAgent.remote( - index=i, - task_ids=task_ids[i::max_workers], - experiment_name=experiment_name, - data_path=data_path, - answer_path=answer_path, - model_name=model_name, - num_runs=num_runs, - use_experience=use_experience, - use_fixed_experience=use_fixed_experience, - enable_thinking=enable_thinking, - experience_base_url=experience_base_url, - experience_workspace_id=experience_workspace_id - ) - future = actor.execute.remote() - future_list.append(future) - time.sleep(1) - logger.info("submit complete") + future_list: list = [] + for i in range(max_workers): + actor = BFCLAgent.remote( + index=i, + task_ids=task_ids[i::max_workers], + experiment_name=experiment_name, + data_path=data_path, + answer_path=answer_path, + model_name=model_name, + num_runs=num_runs, + use_experience=use_experience, + use_fixed_experience=use_fixed_experience, + use_experience_deletion=use_experience_deletion, + delete_freq=delete_freq, + freq_threshold=freq_threshold, + utility_threshold=utility_threshold, + enable_thinking=enable_thinking, + experience_base_url=experience_base_url, + experience_workspace_id=experience_workspace_id + ) + future = actor.execute.remote() + future_list.append(future) + time.sleep(1) + logger.info("submit complete") - for i, future in enumerate(future_list): - t_result = ray.get(future) - if t_result: - if isinstance(t_result, list): - result.extend(t_result) - else: - result.append(t_result) - - logger.info(f"{i + 1}/{len(task_ids)} complete") - dump_file() - - else: - for index, task_id in enumerate(task_ids): - agent = BFCLAgent(index=index, - task_ids=[task_id], - experiment_name=experiment_name, - num_runs=num_runs, - model_name=model_name, - data_path=data_path, - answer_path=answer_path, - enable_thinking=enable_thinking, - use_experience=use_experience, - use_fixed_experience=use_fixed_experience, - experience_base_url=experience_base_url, - experience_workspace_id=experience_workspace_id) - task_results = agent.execute() - if isinstance(task_results, list): - result.extend(task_results) + for i, future in enumerate(future_list): + t_result = ray.get(future) + if t_result: + if isinstance(t_result, list): + result.extend(t_result) else: - result.append(task_results) - dump_file() + result.append(t_result) + + logger.info(f"{i + 1}/{len(task_ids)} complete") + dump_file() + def main(): max_workers = 4 - num_runs = 4 # Run each task 4 times + num_runs = 1 use_experience = True - use_fixed_experience = True - experience_base_url = "http://0.0.0.0:8001/" + use_fixed_experience = False + use_experience_deletion = True + experience_base_url = "http://0.0.0.0:8002/" experience_workspace_id = "bfcl_v1" if max_workers > 1: ray.init(num_cpus=4) for run_id in range(num_runs): run_agent( dataset_name="bfcl-multi-turn-base-val", - experiment_suffix=f"0813-w-exp-w-think-update-test", + experiment_suffix=f"0815-w-exp-recall-rewrite-update-delete-test", # model_name="qwen3-8b", max_workers=max_workers, num_runs=1, data_path="data/multiturn_data_base_val.jsonl", answer_path=Path("data/possible_answer"), - enable_thinking=True, + enable_thinking=False, use_experience=use_experience, use_fixed_experience=use_fixed_experience, + use_experience_deletion=use_experience_deletion, + delete_freq=5, + freq_threshold=5, + utility_threshold=0.5, experience_base_url=experience_base_url, experience_workspace_id=experience_workspace_id, ) diff --git a/experiencemaker/experiencemaker/op/vector_store/vector_store_action_op.py b/experiencemaker/experiencemaker/op/vector_store/vector_store_action_op.py index 243b3c05..fe0d897b 100644 --- a/experiencemaker/experiencemaker/op/vector_store/vector_store_action_op.py +++ b/experiencemaker/experiencemaker/op/vector_store/vector_store_action_op.py @@ -40,6 +40,15 @@ class VectorStoreActionOp(BaseOp): path=request.path, callback_fn=experience_dict_to_node) + elif request.action == "update_freq": + result = self.vector_store.update_freq(workspace_id=request.workspace_id, node_ids = request.experience_ids) + + elif request.action == "update_utility": + result = self.vector_store.update_utility(workspace_id=request.workspace_id, node_ids = request.experience_ids) + + elif request.action == "utility_based_delete": + result = self.vector_store.utility_based_delete(workspace_id=request.workspace_id, freq_threshold = request.freq_threshold, utility_threshold = request.utility_threshold) + else: raise ValueError(f"invalid action={request.action}") diff --git a/experiencemaker/experiencemaker/schema/request.py b/experiencemaker/experiencemaker/schema/request.py index 44df5008..50468b32 100644 --- a/experiencemaker/experiencemaker/schema/request.py +++ b/experiencemaker/experiencemaker/schema/request.py @@ -20,16 +20,13 @@ class SummarizerRequest(BaseRequest): traj_list: List[Trajectory] = Field(default_factory=list) -class ManagerRequest(BaseRequest): - freq_threshold: int = Field(default=10) - utility_threshold: float = Field(default=0.6) - - class VectorStoreRequest(BaseRequest): action: str = Field(default="") src_workspace_id: str = Field(default="") path: str = Field(default="") experience_ids: List[str] = Field(default_factory=list) + freq_threshold: int = Field(default=5) + utility_threshold: float = Field(default=0.5) class AgentRequest(BaseRequest): diff --git a/experiencemaker/experiencemaker/schema/vector_node.py b/experiencemaker/experiencemaker/schema/vector_node.py index 2244b814..7897dbb0 100644 --- a/experiencemaker/experiencemaker/schema/vector_node.py +++ b/experiencemaker/experiencemaker/schema/vector_node.py @@ -8,4 +8,6 @@ class VectorNode(BaseModel): workspace_id: str = Field(default="") content: str = Field(default="") vector: List[float] | None = Field(default=None) + freq: int = Field(default=0) + utility: int = Field(default=0) metadata: dict = Field(default_factory=dict) diff --git a/experiencemaker/experiencemaker/vector_store/base_vector_store.py b/experiencemaker/experiencemaker/vector_store/base_vector_store.py index f16f2707..cdf32126 100644 --- a/experiencemaker/experiencemaker/vector_store/base_vector_store.py +++ b/experiencemaker/experiencemaker/vector_store/base_vector_store.py @@ -134,3 +134,11 @@ class BaseVectorStore(BaseModel, ABC): def delete(self, node_ids: str | List[str], workspace_id: str, **kwargs): raise NotImplementedError + def update_freq(self, node_ids: str | List[str], workspace_id: str, **kwargs): + raise NotImplementedError + + def update_utility(self, node_ids: str | List[str], workspace_id: str, **kwargs): + raise NotImplementedError + + def utility_based_delete(self, workspace_id: str, freq_threshold: int, utility_threshold: float, **kwargs): + raise NotImplementedError \ No newline at end of file diff --git a/experiencemaker/experiencemaker/vector_store/es_vector_store.py b/experiencemaker/experiencemaker/vector_store/es_vector_store.py index 62043edd..2a1b7ff8 100644 --- a/experiencemaker/experiencemaker/vector_store/es_vector_store.py +++ b/experiencemaker/experiencemaker/vector_store/es_vector_store.py @@ -3,6 +3,7 @@ from typing import List, Tuple, Iterable from elasticsearch import Elasticsearch from elasticsearch.helpers import bulk +from elasticsearch import NotFoundError from loguru import logger from pydantic import Field, PrivateAttr, model_validator @@ -131,7 +132,9 @@ class EsVectorStore(BaseVectorStore): "workspace_id": workspace_id, "content": node.content, "metadata": node.metadata, - "vector": node.vector + "vector": node.vector, + "freq": node.freq, + "utility": node.utility } } for node in embedded_nodes + now_embedded_nodes] status, error = bulk(self._client, docs, chunk_size=self.batch_size, **kwargs) @@ -159,8 +162,107 @@ class EsVectorStore(BaseVectorStore): if refresh: self.refresh(workspace_id=workspace_id) + + def update_freq(self, node_ids: str | List[str], workspace_id: str, refresh: bool = False, **kwargs): + if not self.exist_workspace(workspace_id=workspace_id): + logger.warning(f"workspace_id={workspace_id} is not exists!") + return + if isinstance(node_ids, str): + node_ids = [node_ids] + actions = [ + { + "_op_type": "update", + "_index": workspace_id, + "_id": node_id, + "script": { + "source": "ctx._source.freq += 1", + "lang": "painless" + } + } for node_id in node_ids + ] + status, error = bulk(self._client, actions, chunk_size=self.batch_size, **kwargs) + logger.info(f"update exp.size={len(node_ids)} status={status} error={error}") + + if refresh: + self.refresh(workspace_id=workspace_id) + + + def update_utility(self, node_ids: str | List[str], workspace_id: str, refresh: bool = False, **kwargs): + if not self.exist_workspace(workspace_id=workspace_id): + logger.warning(f"workspace_id={workspace_id} is not exists!") + return + + if isinstance(node_ids, str): + node_ids = [node_ids] + + existing_nodes = [] + for node_id in node_ids: + try: + self._client.get(index=workspace_id, id=node_id) + existing_nodes.append(node_id) + except NotFoundError: + logger.warning(f"Experience_id={node_id} not found in workspace_id={workspace_id}") + + actions = [ + { + "_op_type": "update", + "_index": workspace_id, + "_id": node_id, + "script": { + "source": "ctx._source.utility += 1", + "lang": "painless" + } + } for node_id in existing_nodes + ] + status, error = bulk(self._client, actions, chunk_size=self.batch_size, **kwargs) + logger.info(f"when updating utility, exp.size={len(node_ids)}, status={status}, error={error}") + + if refresh: + self.refresh(workspace_id=workspace_id) + + def utility_based_delete(self, workspace_id: str, freq_threshold: int, utility_threshold: float, refresh: bool = False, **kwargs): + if not self.exist_workspace(workspace_id=workspace_id): + logger.warning(f"workspace_id={workspace_id} is not exists!") + return + + query = { + "query": { + "bool": { + "must": [{ + "script": { + "script": { + "source": "doc.freq.value > params.freq_threshold && (doc.utility.value * 1.0 / doc.freq.value) < params.utility_threshold", + "params": { + "freq_threshold": freq_threshold, + "utility_threshold": utility_threshold + } + } + } + }] + } + } + } + + # response = self._client.search(index=workspace_id, body=query, *kwargs) + # delete_node_ids = [] + # for doc in response['hits']['hits']: + # delete_node_ids.append(doc["_id"]) + from elasticsearch.helpers import scan + results = scan(self._client, index=workspace_id, query=query, _source=False) + delete_node_ids = [hit['_id'] for hit in results] + + if delete_node_ids: + logger.info(f"Found {len(delete_node_ids)} nodes to delete with freq>={freq_threshold} and utility/freq<{utility_threshold}") + self.delete(node_ids=delete_node_ids, workspace_id=workspace_id, refresh=refresh, **kwargs) + else: + logger.info("No nodes found matching the delete criteria") + + if refresh: + self.refresh(workspace_id=workspace_id) + + def main(): from dotenv import load_dotenv load_dotenv() diff --git a/experiencemaker/experiencemaker/vector_store/file_vector_store.py b/experiencemaker/experiencemaker/vector_store/file_vector_store.py index 0ea6bf41..0d9af9f7 100644 --- a/experiencemaker/experiencemaker/vector_store/file_vector_store.py +++ b/experiencemaker/experiencemaker/vector_store/file_vector_store.py @@ -104,7 +104,66 @@ class FileVectorStore(BaseVectorStore): self._dump_to_path(nodes=all_nodes, workspace_id=workspace_id, path=self.store_path, **kwargs) logger.info(f"delete workspace_id={workspace_id} before_size={before_size} after_size={after_size}") + def update_freq(self, node_ids: str | List[str], workspace_id: str, **kwargs): + if not self.exist_workspace(workspace_id=workspace_id): + logger.warning(f"workspace_id={workspace_id} is not exists!") + return + if isinstance(node_ids, str): + node_ids = [node_ids] + + all_nodes: List[VectorNode] = list(self._load_from_path(path=self.store_path, workspace_id=workspace_id)) + all_new_nodes = [] + freq_counter = [] + for n in all_nodes: + if n.unique_id in node_ids: + n.freq += 1 + + if n.freq not in freq_counter: + freq_counter[n.freq] = 0 + freq_counter[n.freq] += 1 + + all_new_nodes.append(n) + + self._dump_to_path(nodes=all_new_nodes, workspace_id=workspace_id, path=self.store_path, **kwargs) + logger.info(f"update workspace_id={workspace_id} update_cnt={len(node_ids)}") + + return freq_counter + + def update_utility(self, node_ids: str | List[str], workspace_id: str, **kwargs): + if not self.exist_workspace(workspace_id=workspace_id): + logger.warning(f"workspace_id={workspace_id} is not exists!") + return + + if isinstance(node_ids, str): + node_ids = [node_ids] + + all_nodes: List[VectorNode] = list(self._load_from_path(path=self.store_path, workspace_id=workspace_id)) + all_new_nodes = [] + for n in all_nodes: + if n.unique_id in node_ids: + n.utility += 1 + all_new_nodes.append(n) + + self._dump_to_path(nodes=all_new_nodes, workspace_id=workspace_id, path=self.store_path, **kwargs) + logger.info(f"update workspace_id={workspace_id} update_utility_cnt={len(node_ids)}") + + def utility_based_delete(self, workspace_id: str, freq_threshold: int, utility_threshold: float, **kwargs): + if not self.exist_workspace(workspace_id=workspace_id): + logger.warning(f"workspace_id={workspace_id} is not exists!") + return + + all_nodes: List[VectorNode] = list(self._load_from_path(path=self.store_path, workspace_id=workspace_id)) + delete_node_ids = [] + for n in all_nodes: + if n.freq >= freq_threshold: + if n.utility*1.0/n.freq < utility_threshold: + delete_node_ids.append(n.unique_id) + + logger.info(f"delete when freq>={freq_threshold} and utility/freq<{utility_threshold}") + self.delete(node_ids=delete_node_ids, workspace_id=workspace_id) + + def main(): from dotenv import load_dotenv load_dotenv()