feat: improve Crew search while resetting their memories

Some memories couldn't be reset due to their reliance on relative external sources like `PDFKnowledge`. This was caused by the need to run the reset memories command from the `src` directory, which could break when external files weren't accessible from that path.

This commit allows the reset command to be executed from the root of the project — the same location typically used to run a crew — improving compatibility and reducing friction.
This commit is contained in:
Lucas Gomide
2025-06-24 10:07:56 -03:00
parent 01e208ee77
commit a20ec5ed5b

View File

@@ -255,50 +255,69 @@ def write_env_file(folder_path, env_vars):
def get_crews(crew_path: str = "crew.py", require: bool = False) -> list[Crew]:
"""Get the crew instances from the a file."""
"""Get the crew instances from a file."""
crew_instances = []
try:
import importlib.util
for root, _, files in os.walk("."):
if crew_path in files:
crew_os_path = os.path.join(root, crew_path)
try:
spec = importlib.util.spec_from_file_location(
"crew_module", crew_os_path
)
if not spec or not spec.loader:
continue
module = importlib.util.module_from_spec(spec)
# Add the current directory to sys.path to ensure imports resolve correctly
current_dir = os.getcwd()
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
# If we're not in src directory but there's a src directory, add it to path
src_dir = os.path.join(current_dir, "src")
if os.path.isdir(src_dir) and src_dir not in sys.path:
sys.path.insert(0, src_dir)
# Search in both current directory and src directory if it exists
search_paths = [".", "src"] if os.path.isdir("src") else ["."]
for search_path in search_paths:
for root, _, files in os.walk(search_path):
if crew_path in files:
crew_os_path = os.path.join(root, crew_path)
try:
sys.modules[spec.name] = module
spec.loader.exec_module(module)
for attr_name in dir(module):
module_attr = getattr(module, attr_name)
try:
crew_instances.extend(fetch_crews(module_attr))
except Exception as e:
print(f"Error processing attribute {attr_name}: {e}")
continue
except Exception as exec_error:
print(f"Error executing module: {exec_error}")
import traceback
print(f"Traceback: {traceback.format_exc()}")
except (ImportError, AttributeError) as e:
if require:
console.print(
f"Error importing crew from {crew_path}: {str(e)}",
style="bold red",
spec = importlib.util.spec_from_file_location(
"crew_module", crew_os_path
)
if not spec or not spec.loader:
continue
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
try:
spec.loader.exec_module(module)
for attr_name in dir(module):
module_attr = getattr(module, attr_name)
try:
crew_instances.extend(fetch_crews(module_attr))
except Exception as e:
print(f"Error processing attribute {attr_name}: {e}")
continue
# If we found crew instances, break out of the loop
if crew_instances:
break
except Exception as exec_error:
print(f"Error executing module: {exec_error}")
except (ImportError, AttributeError) as e:
if require:
console.print(
f"Error importing crew from {crew_path}: {str(e)}",
style="bold red",
)
continue
# If we found crew instances in this search path, break out of the search paths loop
if crew_instances:
break
if require:
if require and not crew_instances:
console.print("No valid Crew instance found in crew.py", style="bold red")
raise SystemExit
@@ -318,11 +337,15 @@ def get_crew_instance(module_attr) -> Crew | None:
and module_attr.is_crew_class
):
return module_attr().crew()
if (ismethod(module_attr) or isfunction(module_attr)) and get_type_hints(
module_attr
).get("return") is Crew:
return module_attr()
elif isinstance(module_attr, Crew):
try:
if (ismethod(module_attr) or isfunction(module_attr)) and get_type_hints(
module_attr
).get("return") is Crew:
return module_attr()
except Exception:
return None
if isinstance(module_attr, Crew):
return module_attr
else:
return None