scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-09-08 22:21:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #42 from agentscope-ai/dev_czy_1126

Browse source 
Fix TypeError in DeleteMemoryOp && Update Cookbook
This commit is contained in:
jinliyl 2025-11-27 16:27:14 +08:00 • committed by GitHub
commit 0ee91e912c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 762 additions and 515 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

271
cookbook/appworld/appworld_react_agent.py
View file

@ -1,25 +1,27 @@
# flake8: noqa: E402, E501
import os
from typing import List
from typing import List, Any
from tqdm import tqdm
os.environ["APPWORLD_ROOT"] = "."
from dotenv import load_dotenv
load_dotenv("../../../.env")
load_dotenv("../../.env")
import re
import time
import json
import ray
import requests
import datetime
from appworld import AppWorld, load_task_ids
from jinja2 import Template
from loguru import logger
from openai import OpenAI
from prompt import PROMPT_TEMPLATE_WITH_EXPERIENCE
from prompt import NEW_PROMPT_TEMPLATE
@ray.remote
@ -35,11 +37,15 @@ class AppworldReactAgent:
temperature: float = 0.9,
max_interactions: int = 30,
max_response_size: int = 2048,
num_runs: int = 1,
use_task_memory: bool = False,
make_task_memory: bool = False,
api_url: str = "http://0.0.0.0:8002/",
workspace_id: str = "appworld_v1",
num_trials: int = 1,
use_memory: bool = False,
use_memory_addition: bool = False,
use_memory_deletion: bool = False,
delete_freq: int = 10,
freq_threshold: int = 5,
utility_threshold: float = 0.5,
memory_base_url: str = "http://0.0.0.0:8002/",
memory_workspace_id: str = "appworld_v1",
):
self.index: int = index
@ -49,14 +55,26 @@ class AppworldReactAgent:
self.temperature: float = temperature
self.max_interactions: int = max_interactions
self.max_response_size: int = max_response_size
self.num_runs: int = num_runs
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.num_trials: int = num_trials
self.use_memory: bool = use_memory
self.use_memory_addition: bool = use_memory_addition if use_memory else False
self.use_memory_deletion: bool = use_memory_deletion if use_memory else False
self.delete_freq: int = delete_freq
self.freq_threshold: int = freq_threshold
self.utility_threshold: float = utility_threshold
self.memory_base_url: str = memory_base_url
self.memory_workspace_id: str = memory_workspace_id
self.llm_client = OpenAI()
self.history: List[List[List[dict]]] = [[] for _ in range(num_trials)]
self.retrieved_memory_list: List[List[List[Any]]] = [[] for _ in range(num_trials)]
for run_id in range(num_trials):
for _ in range(len(task_ids)):
self.retrieved_memory_list[run_id].append([])
self.history[run_id].append([])
def call_llm(self, messages: list) -> str:
for i in range(100):
try:
@ -76,36 +94,39 @@ class AppworldReactAgent:
return "call llm error"
def prompt_messages(self, world: AppWorld) -> list[dict]:
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)
prompt = Template(PROMPT_TEMPLATE_WITH_EXPERIENCE.lstrip()).render(dictionary)
messages: list[dict] = []
# last_start = 0
# for match in re.finditer("(USER|ASSISTANT|SYSTEM):\n", prompt):
# last_end = match.span()[0]
# if len(messages) == 0:
# if last_end != 0:
# raise ValueError(
# f"Start of the prompt has no assigned role: {prompt[:last_end]}"
# )
# else:
# messages[-1]["content"] = prompt[last_start:last_end]
# role_type = match.group(1).lower()
# messages.append({"role": role_type, "content": None})
# last_start = match.span()[1]
# messages[-1]["content"] = prompt[last_start:]
messages.append({"role": "user", "content": prompt})
return messages
def prompt_messages(self, run_id, task_index, previous_memories: None, world: AppWorld):
app_descriptions = json.dumps(
[
{"name": k, "description": v}
for (k, v) in world.task.app_descriptions.items()
],
indent=1,
)
dictionary = {"supervisor": world.task.supervisor, "app_descriptions": app_descriptions}
sys_prompt = Template(NEW_PROMPT_TEMPLATE.lstrip()).render(dictionary)
query = world.task.instruction
if self.use_memory:
if len(previous_memories) == 0:
response = self.get_memory(world.task.instruction)
if response and "memory_list" in response["metadata"]:
self.retrieved_memory_list[run_id][task_index] = response["metadata"]["memory_list"]
task_memory = response["answer"]
logger.info(f"loaded task_memory: {task_memory}")
query = "Task:\n" + query + "\n\nSome Related Experience to help you to complete the task:\n" + task_memory
else:
formatted_memories = []
for i, memory in enumerate(previous_memories, 1):
condition = memory["when_to_use"]
memory_content = memory["content"]
memory_text = f"Experience {i}:\n When to use: {condition}\n Content: {memory_content}\n"
formatted_memories.append(memory_text)
query = "Task:\n" + query + "\n\nSome Related Experience to help you to complete the task:\n" + "\n".join(formatted_memories)
messages = [
{"role": "system", "content": sys_prompt},
{"role": "user", "content": query}
]
self.history[run_id][task_index] = messages
@staticmethod
def get_reward(world) -> float:
@ -114,45 +135,97 @@ class AppworldReactAgent:
num_failures = len(tracker.failures)
return num_passes / (num_passes + num_failures)
def extract_code_and_fix_content(
self, text: str, ignore_multiple_calls=True
) -> tuple[str, str]:
full_code_regex = r"```python\n(.*?)```"
partial_code_regex = r".*```python\n(.*)"
original_text = text
output_code = ""
match_end = 0
# Handle multiple calls
for re_match in re.finditer(full_code_regex, original_text, flags=re.DOTALL):
code = re_match.group(1).strip()
if ignore_multiple_calls:
text = original_text[: re_match.end()]
return code, text
output_code += code + "\n"
match_end = re_match.end()
# check for partial code match at end (no terminating ```) following the last match
partial_match = re.match(
partial_code_regex, original_text[match_end:], flags=re.DOTALL
)
if partial_match:
output_code += partial_match.group(1).strip()
# terminated due to stop condition. Add stop condition to output.
if not text.endswith("\n"):
text = text + "\n"
text = text + "```"
if len(output_code) == 0:
return text, text
else:
return output_code, text
def execute(self):
result = []
counter = 0
for task_index, task_id in enumerate(tqdm(self.task_ids, desc=f"ray_index={self.index}")):
# Run each task num_runs times
for run_id in range(self.num_runs):
t_result = None
previous_memories = []
# Run each task num_trials times
for run_id in range(self.num_trials):
start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with AppWorld(task_id=task_id, experiment_name=f"{self.experiment_name}_run_{run_id}") as world:
history = self.prompt_messages(world=world)
before_score = self.get_reward(world)
for i in range(self.max_interactions):
code = self.call_llm(history)
history.append({"role": "assistant", "content": code})
if i == 0:
self.prompt_messages(run_id=run_id, task_index=task_index, previous_memories=previous_memories, world=world)
code_msg = self.call_llm(self.history[run_id][task_index])
code, text = self.extract_code_and_fix_content(code_msg)
self.history[run_id][task_index].append({"role": "assistant", "content": code})
output = world.execute(code)
if len(output) > self.max_response_size:
# logger.warning(f"output exceed max size={len(output)}")
output = output[: self.max_response_size]
history.append({"role": "user", "content": output})
self.history[run_id][task_index].append({"role": "user", "content": "Output:\n```\n" + output + "```\n\n"})
if world.task_completed():
break
after_score = self.get_reward(world)
uplift_score = after_score - before_score
if self.use_memory:
if self.use_memory_addition:
new_traj_list = [self.get_traj_from_task_history(task_id, self.history[run_id][task_index], after_score)]
previous_memories = self.add_memory(new_traj_list)
if after_score != 1:
self.delete_memory_by_ids([mem["memory_id"] for mem in previous_memories])
# update the freq & utility attributes of retrieved memories
update_utility: bool = after_score == 1
self.update_memory_information(self.retrieved_memory_list[run_id][task_index], update_utility)
counter += 1
if self.use_memory_deletion: # and counter % self.delete_freq == 0:
self.delete_memory()
t_result = {
"task_id": world.task_id,
"run_id": run_id, # Add run_id field
"run_id": run_id,
"experiment_name": self.experiment_name,
"task_completed": world.task_completed(),
"before_score": before_score,
"after_score": after_score,
"uplift_score": uplift_score,
"task_history": history,
"task_history": self.history[run_id][task_index],
"task_start_time": start_time,
}
result.append(t_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")
if after_score == 1:
break
result.append(t_result)
return result
@ -165,68 +238,90 @@ class AppworldReactAgent:
return response.json()
def get_task_memory(self, query: str):
def get_memory(self, query: str):
"""Retrieve relevant task memories based on a query"""
response = requests.post(
url=f"{self.api_url}retrieve_task_memory",
url=f"{self.memory_base_url}retrieve_task_memory",
json={
"workspace_id": self.workspace_id,
"workspace_id": self.memory_workspace_id,
"query": query,
},
)
result = self.handle_api_response(response)
if not result:
return ""
return None
# Extract and return the answer
answer = result.get("answer", "")
print(f"Retrieved task memory: {answer}")
return answer
logger.info(f"query: {query}, response: {result}")
return result
def make_task_memory(self, result):
def get_traj_from_task_history(self, task_id: str, task_history: list, reward: float):
pattern = r"\n\nSome Related Experience to help you to complete the task:.*"
task_history[1]["content"] = re.sub(pattern, "", task_history[1]["content"], flags=re.DOTALL)
return {
"task_id": task_id,
"messages": task_history,
"score": reward
}
def add_memory(self, trajectories):
"""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",
url=f"{self.memory_base_url}summary_task_memory",
json={
"workspace_id": self.workspace_id,
"workspace_id": self.memory_workspace_id,
"trajectories": trajectories,
},
)
result = self.handle_api_response(response)
if not result:
return
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 delete_memory_by_ids(self, memory_ids):
response = requests.post(
url=f"{self.memory_base_url}vector_store",
json={
"workspace_id": self.memory_workspace_id,
"action": "delete_ids",
"memory_ids": memory_ids
}
)
response.raise_for_status()
def update_memory_information(self, memory_list, update_utility: bool = False):
response = requests.post(
url=f"{self.memory_base_url}record_task_memory",
json={
"workspace_id": self.memory_workspace_id,
"memory_dicts": memory_list,
"update_utility": update_utility,
},
)
response.raise_for_status()
logger.info(response.json())
def delete_memory(self):
response = requests.post(
url=f"{self.memory_base_url}delete_task_memory",
json={
"workspace_id": self.memory_workspace_id,
"freq_threshold": self.freq_threshold,
"utility_threshold": self.utility_threshold,
},
)
response.raise_for_status()
def main():
dataset_name = "train"
task_ids = load_task_ids(dataset_name)
agent = AppworldReactAgent(index=0, task_ids=task_ids[0:1], experiment_name=dataset_name, num_runs=4)
agent = AppworldReactAgent(index=0, task_ids=task_ids[0:1], experiment_name=dataset_name, num_trials=1)
result = agent.execute()
logger.info(f"result={json.dumps(result)}")

346
cookbook/appworld/prompt.py
View file

@ -311,3 +311,349 @@ Task:
{{ instruction }}
"""
NEW_PROMPT_TEMPLATE = """
USER:
I am your supervisor and you are a super intelligent AI Assistant whose job is to achieve my day-to-day tasks completely autonomously.
To do this, you will need to interact with app/s (e.g., spotify, venmo etc) using their associated APIs on my behalf. For this you will undertake a *multi-step conversation* using a python REPL environment. That is, you will write the python code and the environment will execute it and show you the result, based on which, you will write python code for the next step and so on, until you've achieved the goal. This environment will let you interact with app/s using their associated APIs on my behalf.
Here are three key APIs that you need to know to get more information
# To get a list of apps that are available to you.
```python
print(apis.api_docs.show_app_descriptions())
```
# To get the list of apis under any app listed above, e.g. spotify
```python
print(apis.api_docs.show_api_descriptions(app_name='spotify'))
```
# To get the specification of a particular api, e.g. spotify app's login api
```python
print(apis.api_docs.show_api_doc(app_name='spotify', api_name='login'))
```
Each code execution will produce an output that you can use in subsequent calls. Using these APIs, you can now generate code, that I will execute, to solve the task. Let's start with the task
My name is: {{ supervisor.first_name }} {{ supervisor.last_name }}. My personal email is {{ supervisor.email }} and phone number is {{ supervisor.phone_number }}.
Task: How many playlists do I have in Spotify?
ASSISTANT:
Okay. Lets first find which APIs are available to use in Spotify.
Code:
```python
print(apis.api_docs.show_api_descriptions(app_name='spotify'))
```
USER:
Output:
```
[
...
"login : Login to your account.",
"logout : Logout from your account.",
...
]
```
ASSISTANT:
Okay. Looks like I can use the `login` api. Lets find its specifications.
Code:
```python
print(apis.api_docs.show_api_doc(app_name='spotify', api_name='login'))
```
USER:
Output:
```
{
"app_name": "spotify",
"api_name": "login",
"path": "/auth/token",
"method": "POST",
"description": "Login to your account.",
"parameters": [
{
"name": "username",
"type": "string",
"required": true,
"description": "Your account email.",
"default": null,
"constraints": []
},
{
"name": "password",
"type": "string",
"required": true,
"description": "Your account password.",
"default": null,
"constraints": []
}
],
"response_schemas": {
"success": {
"token_type": "string",
"access_token": "string"
},
"failure": {
"message": "string"
}
}
}
```
ASSISTANT:
I need the supervisor's username and password. Lets see if any app can help me get that.
Code:
```python
print(apis.api_docs.show_app_descriptions())
```
USER:
Output:
```
{{ app_descriptions }}
```
ASSISTANT:
Looks like the supervisor app could help me with that.
Code:
```python
print(apis.api_docs.show_api_descriptions(app_name='supervisor'))
```
USER:
Output:
```
[
...
"show_account_passwords : Show your supervisor's account passwords."
...
]
```
ASSISTANT:
I can use `show_account_passwords` to get the passwords. So I will call that and save it in a variable for future reference.
Code:
```python
passwords=apis.supervisor.show_account_passwords()
print(passwords)
```
USER:
Output:
```
[
{
"account_name": "spotify",
"password": "dummy_spotify_pass"
},
{
"account_name": "file_system",
"password": "dummy_fs_pass"
}
]
```
ASSISTANT:
So the Spotify password is an entry in the `passwords` list with the account_name=spotify. I can use it to login now with the provided email address: {{ supervisor.email }}.
Code:
```python
spotify_password = [account_password["account_name"] == "spotify" for account_password in passwords][0]["password"]
login_result = apis.spotify.login(username='{{ supervisor.email }}', password=spotify_password)
print(login_result)
```
USER:
Output:
```
{
"token_type": "Bearer",
"access_token": "fake_access_token"
}
```
ASSISTANT:
First, let's check the available APIs for the Spotify app.
Code:
```python
print(apis.api_docs.show_api_descriptions(app_name='spotify'))
```
USER:
Output:
```
[
...
{
"name": "show_playlist_library",
"description": "Get a list of playlists in the user's playlist library."
},
{
"name": "show_playlist",
"description": "Get detailed information about a specific playlist. You can view your own playlists or others' playlists if they are public."
},
...
]
```
ASSISTANT:
The `show_playlist_library` API seems to be the one we need. Let's check its specifications.
Code:
```python
print(apis.api_docs.show_api_doc(app_name='spotify', api_name='show_playlist_library'))
```
USER:
Output:
```
{
"app_name": "spotify",
"api_name": "show_playlist_library",
"path": "/private_playlists",
"method": "GET",
"description": "Get a list of playlists in the user's playlist library.",
"parameters": [
{
"name": "access_token",
"type": "string",
"required": true,
"description": "Access token obtained from spotify app login.",
"default": null,
"constraints": []
},
{
"name": "page_index",
"type": "integer",
"required": false,
"description": "The index of the page to retrieve.",
"default": 0,
"constraints": [
"value >= 0.0"
]
},
{
"name": "page_limit",
"type": "integer",
"required": false,
"description": "The maximum number of results to return per page.",
"default": 5,
"constraints": [
"value >= 1.0, <= 20.0"
]
},
{
"name": "is_public",
"type": "boolean",
"required": false,
"description": "Whether to show public playlists or private playlists.",
"default": null,
"constraints": []
}
],
"response_schema": [
{
"title": "string",
"created_at": "2019-01-01T00:00:00",
"is_public": true,
"rating": 0.0,
"like_count": 1,
"owner_email": "user@example.com",
"playlist_id": 1,
"song_ids": [
1
]
}
]
}
```
ASSISTANT:
I need to page through all the playlists to get the list of playlists and save it in `playlists`.
Code:
```python
page_index = 0
playlists = []
while page_index < 10:
playlist_page = apis.spotify.show_playlist_library(access_token=spotify_access_token, page_index=page_index)
if playlist_page:
playlists.extend(playlist_page)
page_index += 1
else:
break
num_playlists = len(playlists)
print(num_playlists)
```
USER:
Output:
```
23
```
ASSISTANT:
Now that the task is completed, I can call apis.supervisor.complete_task(). Since this task has an answer to be returned, I will pass that as an argument.
Code:
```python
apis.supervisor.complete_task(answer=num_playlists)
```
USER:
Output:
Marked the active task complete.
----------------------------------------------
USER:
**Key instructions**:
(1) Make sure to end code blocks with ``` followed by a newline(\n).
(2) Remember you can use the variables in your code in subsequent code blocks.
(3) Remember that the email addresses, access tokens and variables (e.g. spotify_password) in the example above are not valid anymore.
(4) You can use the "supervisor" app to get information about my accounts and use the "phone" app to get information about friends and family.
(5) Always look at API specifications (using apis.api_docs.show_api_doc) before calling an API.
(6) Write small chunks of code and only one chunk of code in every step. Make sure everything is working correctly before making any irreversible change.
(7) Many APIs return items in "pages". Make sure to run through all the pages by looping over `page_index`.
(8) Once you have completed the task, make sure to call apis.supervisor.complete_task(). If the task asked for some information, return it as the answer argument, i.e. call apis.supervisor.complete_task(answer=<answer>). Many tasks do not require an answer, so in those cases, just call apis.supervisor.complete_task() i.e. do not pass any argument.
USER:
Using these APIs, now generate code to solve the actual task:
My name is: {{ supervisor.first_name }} {{ supervisor.last_name }}. My personal email is {{ supervisor.email }} and phone number is {{ supervisor.phone_number }}.
"""

160
cookbook/appworld/run_appworld.py
View file

@ -77,17 +77,23 @@ def load_memory(workspace_id: str, path: str = "docs/library", api_url: str = "h
def run_agent(
model_name: str,
dataset_name: str,
experiment_suffix: str,
max_workers: int,
num_runs: int = 1,
use_task_memory: bool = False,
make_task_memory: bool = False,
num_trials: int = 1,
use_memory: bool = False,
use_memory_addition: bool = False,
use_memory_deletion: bool = False,
delete_freq: int = 10,
freq_threshold: int = 5,
utility_threshold: float = 0.5,
workspace_id: str = "appworld_v1",
api_url: str = "http://0.0.0.0:8002/",
batch_size: int = 4
):
experiment_name = dataset_name + "_" + experiment_suffix
path: Path = Path(f"./exp_result")
path: Path = Path(f"./exp_result/{model_name}")
path.mkdir(parents=True, exist_ok=True)
task_ids = load_task_ids(dataset_name)
@ -99,45 +105,84 @@ def run_agent(
f.write(json.dumps(x) + "\n")
if max_workers > 1:
future_list: list = []
for i in range(max_workers):
# Assign tasks to each worker, ensuring each task runs num_runs times
worker_task_ids = task_ids[i::max_workers]
actor = AppworldReactAgent.remote(
index=i,
task_ids=worker_task_ids,
experiment_name=experiment_name,
num_runs=num_runs,
use_task_memory=use_task_memory,
make_task_memory=make_task_memory,
workspace_id=workspace_id,
api_url=api_url,
)
future = actor.execute.remote()
future_list.append(future)
time.sleep(1)
logger.info("submit complete")
# Process tasks in batches
total_tasks = len(task_ids)
num_batches = (total_tasks + batch_size - 1) // batch_size # Ceiling division
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"Total tasks: {total_tasks}, Batch size: {batch_size}, Number of batches: {num_batches}")
for batch_idx in range(num_batches):
# Initialize Ray for this batch
start_idx = batch_idx * batch_size
end_idx = min(start_idx + batch_size, total_tasks)
batch_task_ids = task_ids[start_idx:end_idx]
logger.info(f"Starting batch {batch_idx + 1}/{num_batches} with {len(batch_task_ids)} tasks")
# Initialize Ray with the number of CPUs needed for this batch
ray.init(num_cpus=len(batch_task_ids))
future_list: list = []
for i, task_id in enumerate(batch_task_ids):
actor = AppworldReactAgent.remote(
index=start_idx+i,
model_name=model_name,
task_ids=[task_id],
experiment_name=experiment_name,
num_trials=num_trials,
use_memory=use_memory,
use_memory_addition=use_memory_addition,
use_memory_deletion=use_memory_deletion,
delete_freq=delete_freq,
freq_threshold=freq_threshold,
utility_threshold=utility_threshold,
memory_workspace_id=workspace_id,
memory_base_url=api_url,
)
future = actor.execute.remote()
future_list.append(future)
time.sleep(1)
logger.info(f"Batch {batch_idx + 1} submit complete, waiting for results...")
# Collect results from this batch
for i, (task_id, future) in enumerate(zip(batch_task_ids, future_list)):
try:
t_result = ray.get(future)
if t_result:
if isinstance(t_result, list):
result.extend(t_result)
else:
result.append(t_result)
except Exception as e:
logger.exception(f"run ray error with task_id={task_id}")
logger.info(f"Batch {batch_idx + 1}: task {i + 1}/{len(batch_task_ids)} complete")
# Shutdown Ray to free resources before next batch
ray.shutdown()
logger.info(f"Batch {batch_idx + 1}/{num_batches} complete, Ray resources released")
# Optional: small delay between batches
if batch_idx < num_batches - 1:
time.sleep(2)
logger.info(f"worker {i + 1}/{max_workers} complete")
dump_file()
else:
for index, task_id in enumerate(task_ids):
agent = AppworldReactAgent(
index=index,
model_name=model_name,
task_ids=[task_id],
experiment_name=experiment_name,
num_runs=num_runs,
use_task_memory=use_task_memory,
make_task_memory=make_task_memory,
num_trials=num_trials,
use_memory=use_memory,
use_memory_addition=use_memory_addition,
use_memory_deletion=use_memory_deletion,
delete_freq=delete_freq,
freq_threshold=freq_threshold,
utility_threshold=utility_threshold,
workspace_id=workspace_id,
api_url=api_url,
)
@ -148,56 +193,47 @@ def run_agent(
result.append(task_results)
dump_file()
def main():
max_workers = 8
num_runs = 1 # Run each task once
num_runs = 1 # Number of runs
batch_size = 8 # Number of concurrent tasks per batch
num_trials = 2
model_name = "qwen3-8b"
use_memory = True
use_memory_addition = True
use_memory_deletion = True
workspace_id = "appworld"
api_url = "http://0.0.0.0:8002/"
if max_workers > 1:
ray.init(num_cpus=8)
# Clean up workspace before starting
logger.info("Deleting workspace...")
delete_workspace(workspace_id=workspace_id, api_url=api_url)
time.sleep(5)
# First run to build task memories
logger.info("Start load experiments to build task memories")
load_memory(workspace_id=workspace_id, api_url=api_url)
# 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)
for i in range(num_runs):
# Run experiments with task memory
logger.info("Start running experiments with task memory")
run_agent(
dataset_name="dev",
model_name=model_name,
dataset_name="test_normal",
experiment_suffix=f"with-memory",
max_workers=max_workers,
num_runs=1,
use_task_memory=True,
make_task_memory=False,
num_trials=num_trials,
use_memory=use_memory,
use_memory_addition=use_memory_addition,
use_memory_deletion=use_memory_deletion,
delete_freq=5,
freq_threshold=5,
utility_threshold=0.5,
workspace_id=workspace_id,
api_url=api_url,
batch_size=batch_size
)
# Run experiments without task memory
logger.info("Start running experiments without task memory")
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,
)
if __name__ == "__main__":
main()

127
cookbook/bfcl/bfcl_agent.py
View file

@ -7,6 +7,7 @@ from dotenv import load_dotenv
load_dotenv("../../.env")
import re
import time
import json
import ray
@ -63,7 +64,7 @@ class BFCLAgent:
temperature: float = 0.9,
max_interactions: int = 30,
max_response_size: int = 2000,
num_runs: int = 1,
num_trials: int = 1,
enable_thinking: bool = False,
use_memory: bool = False,
use_memory_addition: bool = False,
@ -71,8 +72,8 @@ class BFCLAgent:
delete_freq: int = 10,
freq_threshold: int = 5,
utility_threshold: float = 0.5,
memory_base_url: str = "http://0.0.0.0:8001/",
memory_workspace_id: str = "bfcl_8b_0725",
memory_base_url: str = "http://0.0.0.0:8002/",
memory_workspace_id: str = "bfcl_v3",
):
self.index: int = index
@ -85,7 +86,7 @@ class BFCLAgent:
self.temperature: float = temperature
self.max_interactions: int = max_interactions
self.max_response_size: int = max_response_size
self.num_runs: int = num_runs
self.num_trials: int = num_trials
self.enable_thinking: bool = enable_thinking
self.use_memory: bool = use_memory
self.use_memory_addition: bool = use_memory_addition if use_memory else False
@ -96,14 +97,14 @@ class BFCLAgent:
self.memory_base_url: str = memory_base_url
self.memory_workspace_id: str = memory_workspace_id
self.history: List[List[List[dict]]] = [[] for _ in range(num_runs)]
self.retrieved_memory_list: List[List[List[Any]]] = [[] 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)]
self.current_turn = [[0 for _ in range(len(task_ids))] for _ in range(num_runs)]
self.history: List[List[List[dict]]] = [[] for _ in range(num_trials)]
self.retrieved_memory_list: List[List[List[Any]]] = [[] for _ in range(num_trials)]
self.test_entry: List[List[Dict[str, Any]]] = [[] for _ in range(num_trials)]
self.original_test_entry: List[List[Dict[str, Any]]] = [[] for _ in range(num_trials)]
self.tool_schema: List[List[List[dict]]] = [[] for _ in range(num_trials)]
self.current_turn = [[0 for _ in range(len(task_ids))] for _ in range(num_trials)]
for run_id in range(num_runs):
for run_id in range(num_trials):
for task_index in range(len(task_ids)):
self.init_state(run_id, task_index)
@ -112,29 +113,40 @@ class BFCLAgent:
self.original_test_entry[run_id].append(self.test_entry[run_id][i].get("extra", {}))
self.tool_schema[run_id].append(extract_tool_schema(self.test_entry[run_id][i].get("tools", [{}])))
msg = self.test_entry[run_id][i].get("messages", [])[0]
if self.use_memory:
query = msg["content"]
response = self.get_memory(query)
if len(response["metadata"]["memory_list"]):
self.retrieved_memory_list[run_id].append(response["metadata"]["memory_list"])
exp: str = response["answer"]
# print(f"memory_merged={exp}")
self.history[run_id].append([self.get_query_with_memory(query, exp)])
else:
self.retrieved_memory_list[run_id].append([])
self.history[run_id].append([msg])
else:
self.history[run_id].append([msg])
msg = self.test_entry[run_id][i].get("messages", [])
self.history[run_id].append(msg)
self.retrieved_memory_list[run_id].append([])
self.current_turn[run_id][i] = 1
def update_task_history_with_memory(self, run_id, task_index, previous_memories: None):
query = self.history[run_id][task_index][0]["content"]
if len(previous_memories) == 0:
response = self.get_memory(query)
if response and "memory_list" in response["metadata"]:
self.retrieved_memory_list[run_id][task_index] = response["metadata"]["memory_list"]
task_memory = response["answer"]
logger.info(f"loaded task_memory: {task_memory}")
self.history[run_id][task_index][0] = self.get_query_with_memory(query, task_memory)
else:
formatted_memories = []
for i, memory in enumerate(previous_memories, 1):
condition = memory["when_to_use"]
memory_content = memory["content"]
memory_text = f"Experience {i}:\n When to use: {condition}\n Content: {memory_content}\n"
formatted_memories.append(memory_text)
self.history[run_id][task_index][0] = self.get_query_with_memory(query, "\n".join(formatted_memories))
def get_query_with_memory(self, query: str, memory: str):
return {
"role": "user",
"content": "Task:\n" + query + "\n\nSome Related Experience to help you to complete the task:\n" + memory,
}
def get_query_without_experience(self, query: str):
if "\n\nSome Related Experience" in query:
query = query.split("\n\nSome Related Experience")[0].split("Task:\n")[-1]
return query
def get_traj_from_task_history(self, task_id: str, task_history: list, reward: float):
return {
"task_id": task_id,
@ -142,6 +154,15 @@ class BFCLAgent:
"score": reward,
}
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_memory(self, query: str):
response = requests.post(
url=self.memory_base_url + "retrieve_task_memory",
@ -152,13 +173,12 @@ class BFCLAgent:
},
)
if response.status_code != 200:
logger.info(response.text)
return ""
result = self.handle_api_response(response)
if not result:
return None
response = response.json()
logger.info(f"query: {query}, response: {response}")
return response
logger.info(f"query: {query}, response: {result}")
return result
def add_memory(self, trajectories):
response = requests.post(
@ -168,9 +188,26 @@ class BFCLAgent:
"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", [])
logger.info(f'add new memories: {memory_list}')
return memory_list
def delete_memory_by_ids(self, memory_ids):
response = requests.post(
url=self.memory_base_url + "vector_store",
json={
"workspace_id": self.memory_workspace_id,
"action": "delete_ids",
"memory_ids": memory_ids
}
)
response.raise_for_status()
response = response.json()
logger.info(f'add new memorys: {response["metadata"]["memory_list"]}')
def update_memory_information(self, memory_list, update_utility: bool = False):
response = requests.post(
@ -555,10 +592,14 @@ class BFCLAgent:
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):
t_result = None
previous_memories = []
for run_id in range(self.num_trials):
try:
start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
for i in range(self.max_interactions):
if self.use_memory and i == 0:
self.update_task_history_with_memory(run_id, task_index, previous_memories)
llm_output = self.call_llm(
self.history[run_id][task_index],
self.tool_schema[run_id][task_index],
@ -605,11 +646,11 @@ class BFCLAgent:
reward = self.get_reward(run_id, task_index)
if self.use_memory:
if reward == 1 and self.use_memory_addition: # selectively add memories when succeed
new_traj_list = [
self.get_traj_from_task_history(task_id, self.history[run_id][task_index], reward),
]
self.add_memory(new_traj_list)
if self.use_memory_addition: # selectively add memories when succeed
new_traj_list = [self.get_traj_from_task_history(task_id, self.history[run_id][task_index], reward)]
previous_memories = self.add_memory(new_traj_list)
if reward != 1:
self.delete_memory_by_ids([mem["memory_id"] for mem in previous_memories])
# update the freq & utility attributes of retrieved memories
update_utility: bool = reward == 1
@ -628,11 +669,13 @@ class BFCLAgent:
"task_history": self.history[run_id][task_index],
"task_start_time": start_time,
}
result.append(t_result)
if reward == 1:
break
except Exception as e:
logger.exception(f"encounter error with {e.args}")
result.append({})
result.append(t_result)
return result
def task_completed(self, run_id, index):
@ -652,7 +695,7 @@ def main():
agent = BFCLAgent(
index=0,
task_id=task_ids[0],
experiment_name=f"zouying_{dataset_name}",
experiment_name=f"qwen3_8b_{dataset_name}",
)
result = agent.execute()
logger.info(f"result={json.dumps(result)}")

250
cookbook/bfcl/evaluation.ipynb
View file

File diff suppressed because one or more lines are too long

93
cookbook/bfcl/init_exp_pool.py
View file

@ -9,7 +9,9 @@ import requests
def load_task_case(data_path: str, task_id: str | None) -> Dict[str, Any]:
"""按 ID加载单条 JSONL 训练用例。找不到就抛错。"""
"""
load training cases by id
"""
if not Path(data_path).exists():
raise FileNotFoundError(f"BFCL data file '{data_path}' not found")
@ -41,15 +43,14 @@ def get_tool_prompt(tools):
def group_trajectories_by_task_id(jsonl_entries: List[Dict[str, Any]]) -> List[List[Any]]:
"""
根据task_id字段对trajectories进行分组
group trajectories by task_id
Args:
jsonl_entries: JSONL条目列表
jsonl_entries: JSONL entry list
Returns:
List[List[Any]]: 按task_id分组的trajectory列表
List[List[Any]]: trajectory list grouped by task_id
"""
# 按task_id分组
grouped = defaultdict(list)
for entry in jsonl_entries:
@ -62,37 +63,26 @@ def group_trajectories_by_task_id(jsonl_entries: List[Dict[str, Any]]) -> List[L
entry["task_history"][0]["content"] += get_tool_prompt(tool_schema)
grouped[task_id].append(entry)
# 对每组只保留最大和最小reward的两个
# retain only the two with the highest and lowest rewards
filtered_groups = []
for key, trajectories in grouped.items():
if len(trajectories) == 1:
# 只有一个trajectory，直接保留
# when only one trajectory, retain it
filtered_groups.append(trajectories)
elif len(trajectories) == 2:
# 有两个trajectory，直接保留
# when there are two trajectories, retain them
filtered_groups.append(trajectories)
else:
# 多个trajectory，选择最大和最小reward的
# when there are more than two trajectories, choose the two with the highest and lowest rewards
trajectories.sort(key=lambda t: t["reward"])
min_reward_traj = trajectories[0] # 最小reward
max_reward_traj = trajectories[-1] # 最大reward
min_reward_traj = trajectories[0] # highest reward
max_reward_traj = trajectories[-1] # lowest reward
filtered_groups.append([min_reward_traj, max_reward_traj])
return filtered_groups
def post_to_summarizer(trajectories: List[Any], service_url: str, workspace_id: str) -> Dict[str, Any]:
"""
将trajectories发送到summarizer服务
Args:
trajectories: trajectory列表
service_url: 服务URL
workspace_id: 工作空间ID
Returns:
响应结果
"""
trajectory_dicts = [
{
"task_id": traj["task_id"],
@ -103,12 +93,12 @@ def post_to_summarizer(trajectories: List[Any], service_url: str, workspace_id:
]
request_data = {
"traj_list": trajectory_dicts,
"trajectories": trajectory_dicts,
"workspace_id": workspace_id,
}
try:
response = requests.post(f"{service_url}/summarizer", json=request_data)
response = requests.post(f"{service_url}/summary_task_memory", json=request_data)
response.raise_for_status()
return response.json()
except Exception as e:
@ -122,27 +112,25 @@ def process_trajectories_with_threads(
n_threads: int = 4,
) -> List[Dict[str, Any]]:
"""
使用多线程处理trajectories组
use threads to process trajectories
Args:
grouped_trajectories: 按task_id分组的trajectory列表
service_url: summarizer服务URL
workspace_id: 工作空间ID
n_threads: 线程数
grouped_trajectories: group trajectory list by task_id
service_url: memory summarizer service URL
workspace_id: workspace ID
n_threads: number of threads
Returns:
所有结果列表
all results
"""
results = []
with ThreadPoolExecutor(max_workers=n_threads) as executor:
# 提交所有任务
future_to_group = {
executor.submit(post_to_summarizer, group, service_url, workspace_id): i
for i, group in enumerate(grouped_trajectories)
}
# 收集结果
for future in as_completed(future_to_group):
group_index = future_to_group[future]
try:
@ -151,7 +139,7 @@ def process_trajectories_with_threads(
result["group_size"] = len(grouped_trajectories[group_index])
results.append(result)
print(
f"✅ Group {group_index} processed: {result.get('experience_list', 0) if 'experience_list' in result else 'error'}",
f'✅ Group {group_index} processed: {result["metadata"].get("memory_list", 0) if "memory_list" in result["metadata"] else "error"}',
)
except Exception as e:
error_result = {
@ -166,13 +154,10 @@ def process_trajectories_with_threads(
def main():
"""
主函数，支持命令行参数
"""
parser = argparse.ArgumentParser(description="Convert JSONL to experiences using experience maker service")
parser = argparse.ArgumentParser(description="Convert JSONL to memories using ReMe service")
parser.add_argument("--jsonl_file", type=str, required=True, help="Path to the JSONL file")
parser.add_argument("--service_url", type=str, default="http://localhost:8001", help="Experience maker service URL")
parser.add_argument("--workspace_id", type=str, required=True, help="Workspace ID for the experience")
parser.add_argument("--service_url", type=str, default="http://localhost:8001", help="ReMe service URL")
parser.add_argument("--workspace_id", type=str, required=True, help="Workspace ID for the task memory pool")
parser.add_argument("--output_file", type=str, help="Output file to save results (optional)")
parser.add_argument("--n_threads", type=int, default=4, help="Number of threads for processing")
@ -183,20 +168,13 @@ def main():
print(f"Workspace ID: {args.workspace_id}")
print(f"Threads: {args.n_threads}")
# 读取JSONL文件
try:
with open(args.jsonl_file, "r") as f:
data = [json.loads(line) for line in f]
print(f"Loaded {len(data)} entries from JSONL file")
except Exception as e:
print(f"Error reading JSONL file: {e}")
return
with open(args.jsonl_file, "r") as f:
data = [json.loads(line) for line in f]
print(f"Loaded {len(data)} entries from JSONL file")
# 分组处理
grouped_trajectories = group_trajectories_by_task_id(data)
print(f"Total groups: {len(grouped_trajectories)}")
# 多线程处理
results = process_trajectories_with_threads(
grouped_trajectories,
args.service_url,
@ -206,16 +184,14 @@ def main():
print(f"Processed {len(results)} groups")
# 统计结果
success_count = sum(1 for r in results if "error" not in r)
error_count = len(results) - success_count
total_experiences = sum(len(r.get("experiences", [])) for r in results if "experiences" in r)
total_memories = sum(len(r["metadata"].get("memory_list", [])) for r in results if "memory_list" in r["metadata"])
print(f"✅ Success: {success_count}")
print(f"❌ Errors: {error_count}")
print(f"📊 Total experiences created: {total_experiences}")
print(f"📊 Total task memories created: {total_memories}")
# 保存结果到文件
if args.output_file:
try:
summary = {
@ -224,7 +200,7 @@ def main():
"total_groups": len(grouped_trajectories),
"success_count": success_count,
"error_count": error_count,
"total_experiences": total_experiences,
"total_task_memories": total_memories,
"results": results,
}
@ -235,28 +211,23 @@ def main():
print(f"Error saving results: {e}")
# 保持原有的使用示例（向后兼容）
if __name__ == "__main__":
# 检查是否有命令行参数
import sys
if len(sys.argv) > 1:
# 使用新的命令行接口
main()
else:
# 保持原有的行为（向后兼容）
print("Running in compatibility mode...")
with open("exp_result/qwen-max-2025-01-25/no_think/bfcl-multi-turn-base-train50_wo-exp.jsonl", "r") as f:
with open("exp_result/qwen3-8b/with_think/bfcl-multi-turn-base-train50_wo-exp.jsonl", "r") as f:
data = [json.loads(line) for line in f]
# 分组
grouped_trajectories = group_trajectories_by_task_id(data)
print(f"Total groups: {len(grouped_trajectories)}")
results = process_trajectories_with_threads(
grouped_trajectories,
"http://localhost:8001",
"bfcl_train50_qwen_max_2025_01_25_extract_compare_validate",
"bfcl_train50_qwen3_8b_extract_compare_validate",
n_threads=4,
)
print(f"Processed {len(results)} groups")

4
cookbook/bfcl/init_task_memory_pool.py
View file

@ -218,7 +218,7 @@ if __name__ == "__main__":
main()
else:
print("Running in compatibility mode...")
with open("exp_result/qwen3-14b/no_think/bfcl-multi-turn-base_wo-exp.jsonl", "r") as f:
with open("exp_result/qwen3-8b/no_think/bfcl-multi-turn-base_wo-exp.jsonl", "r") as f:
data = [json.loads(line) for line in f]
grouped_trajectories = group_trajectories_by_task_id(data)
@ -227,7 +227,7 @@ if __name__ == "__main__":
results = process_trajectories_with_threads(
grouped_trajectories,
"http://localhost:8001",
"bfcl_test",
"bfcl_v3",
n_threads=4,
)
print(f"Processed {len(results)} groups")

2
cookbook/bfcl/local_file_to_library.py
View file

@ -19,6 +19,8 @@ for exp in bfcl:
new_exp["author"] = exp["metadata"]["author"]
new_exp["metadata"] = exp["metadata"]["metadata"]
new_exp["metadata"]["utility"] = 0
new_exp["metadata"]["freq"] = 0
new_bfcl.append(new_exp)

21
cookbook/bfcl/run_bfcl.py
View file

@ -19,7 +19,7 @@ def run_agent(
dataset_name: str,
experiment_suffix: str,
max_workers: int,
num_runs: int = 4,
num_trials: int = 1,
model_name: str = "qwen3-8b",
data_path: str = "data/multiturn_data_base_val.jsonl",
answer_path: Path = Path("data/possible_answer"),
@ -30,8 +30,8 @@ def run_agent(
freq_threshold: int = 5,
utility_threshold: float = 0.5,
enable_thinking: bool = False,
memory_base_url: str = "http://0.0.0.0:8001/",
memory_workspace_id: str = "bfcl_test",
memory_base_url: str = "http://0.0.0.0:8002/",
memory_workspace_id: str = "bfcl_v3",
):
experiment_name = dataset_name + "_" + experiment_suffix
path: Path = Path(
@ -58,7 +58,7 @@ def run_agent(
data_path=data_path,
answer_path=answer_path,
model_name=model_name,
num_runs=num_runs,
num_trials=num_trials,
use_memory=use_memory,
use_memory_addition=use_memory_addition,
use_memory_deletion=use_memory_deletion,
@ -89,20 +89,25 @@ def run_agent(
def main():
max_workers = 4
num_runs = 1
num_trials = 2
model_name="qwen3-8b"
use_memory = False
use_memory_addition = False
use_memory_deletion = False
memory_base_url = "http://0.0.0.0:8001/"
memory_workspace_id = "bfcl_test"
memory_base_url = "http://0.0.0.0:8002/"
memory_workspace_id = "bfcl_v3"
if max_workers > 1:
ray.init(num_cpus=max_workers)
for run_id in range(num_runs):
run_agent(
dataset_name="bfcl-multi-turn-base",
experiment_suffix=f"wo-exp",
model_name="qwen3-8b",
model_name=model_name,
max_workers=max_workers,
num_runs=1,
num_trials=num_trials,
data_path="data/multiturn_data_base_val.jsonl",
answer_path=Path("data/possible_answer"),
enable_thinking=False,

1
reme_ai/vector_store/delete_memory_op.py
View file

@ -48,7 +48,6 @@ class DeleteMemoryOp(BaseAsyncOp):
deleted_memory_ids = []
for node in nodes:
# VectorNode 对象需要使用属性访问，不是字典访问
freq = node.metadata.get("freq", 0)
utility = node.metadata.get("utility", 0)
if freq >= freq_threshold:

2
reme_ai/vector_store/vector_store_action_op.py
View file

@ -99,4 +99,4 @@ class VectorStoreActionOp(BaseAsyncOp):
else:
raise ValueError(f"invalid action={action}")
self.context.response.metadata["action_result"] = result
self.context.response.metadata["action_result"] = str(result)