scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-08-28 05:25:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

add: update mechanism

Browse source
This commit is contained in:
zouying 2025-08-15 11:47:58 +08:00
parent f497ee8dac
commit bb7ca27e42
9 changed files with 298 additions and 77 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -35,4 +35,5 @@ experiencemaker/cookbook/bfcl/no_exp_result
experiencemaker/cookbook/bfcl/data
experiencemaker/cookbook/bfcl/gorilla
experiencemaker/*.sh
experiencemaker/file_vector_store
experiencemaker/file_vector_store
experiencemaker/*.egg-info

77
experiencemaker/cookbook/bfcl/bfcl_agent.py
View file

@ -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],

106
experiencemaker/cookbook/bfcl/run_bfcl.py
View file

@ -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,
)

9
experiencemaker/experiencemaker/op/vector_store/vector_store_action_op.py
View file

@ -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}")

7
experiencemaker/experiencemaker/schema/request.py
View file

@ -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):

2
experiencemaker/experiencemaker/schema/vector_node.py
View file

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

8
experiencemaker/experiencemaker/vector_store/base_vector_store.py
View file

@ -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

104
experiencemaker/experiencemaker/vector_store/es_vector_store.py
View file

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

59
experiencemaker/experiencemaker/vector_store/file_vector_store.py
View file

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