mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-05-03 16:22:49 +00:00
Fix static typing errors (#187)
Co-authored-by: João Moura <joaomdmoura@gmail.com>
This commit is contained in:
@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field, PrivateAttr, ValidationError, model_valid
|
||||
|
||||
|
||||
class I18N(BaseModel):
|
||||
_translations: Optional[Dict[str, str]] = PrivateAttr()
|
||||
_translations: Dict[str, Dict[str, str]] = PrivateAttr()
|
||||
language: Optional[str] = Field(
|
||||
default="en",
|
||||
description="Language used to load translations",
|
||||
@@ -25,10 +25,14 @@ class I18N(BaseModel):
|
||||
self._translations = json.load(f)
|
||||
except FileNotFoundError:
|
||||
raise ValidationError(
|
||||
f"Trasnlation file for language '{self.language}' not found."
|
||||
f"Translation file for language '{self.language}' not found."
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
raise ValidationError(f"Error decoding JSON from the prompts file.")
|
||||
|
||||
if not self._translations:
|
||||
self._translations = {}
|
||||
|
||||
return self
|
||||
|
||||
def slice(self, slice: str) -> str:
|
||||
@@ -40,8 +44,8 @@ class I18N(BaseModel):
|
||||
def tools(self, error: str) -> str:
|
||||
return self.retrieve("tools", error)
|
||||
|
||||
def retrieve(self, kind, key):
|
||||
def retrieve(self, kind, key) -> str:
|
||||
try:
|
||||
return self._translations[kind].get(key)
|
||||
return self._translations[kind][key]
|
||||
except:
|
||||
raise ValidationError(f"Translation for '{kind}':'{key}' not found.")
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from typing import ClassVar
|
||||
|
||||
from langchain.prompts import PromptTemplate
|
||||
from langchain.prompts import PromptTemplate, BasePromptTemplate
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from crewai.utilities import I18N
|
||||
@@ -13,19 +13,19 @@ class Prompts(BaseModel):
|
||||
|
||||
SCRATCHPAD_SLICE: ClassVar[str] = "\n{agent_scratchpad}"
|
||||
|
||||
def task_execution_with_memory(self) -> str:
|
||||
def task_execution_with_memory(self) -> BasePromptTemplate:
|
||||
"""Generate a prompt for task execution with memory components."""
|
||||
return self._build_prompt(["role_playing", "tools", "memory", "task"])
|
||||
|
||||
def task_execution_without_tools(self) -> str:
|
||||
def task_execution_without_tools(self) -> BasePromptTemplate:
|
||||
"""Generate a prompt for task execution without tools components."""
|
||||
return self._build_prompt(["role_playing", "task"])
|
||||
|
||||
def task_execution(self) -> str:
|
||||
def task_execution(self) -> BasePromptTemplate:
|
||||
"""Generate a standard prompt for task execution."""
|
||||
return self._build_prompt(["role_playing", "tools", "task"])
|
||||
|
||||
def _build_prompt(self, components: [str]) -> str:
|
||||
def _build_prompt(self, components: list[str]) -> BasePromptTemplate:
|
||||
"""Constructs a prompt string from specified components."""
|
||||
prompt_parts = [self.i18n.slice(component) for component in components]
|
||||
prompt_parts.append(self.SCRATCHPAD_SLICE)
|
||||
|
||||
@@ -12,7 +12,7 @@ class RPMController(BaseModel):
|
||||
max_rpm: Union[int, None] = Field(default=None)
|
||||
logger: Logger = Field(default=None)
|
||||
_current_rpm: int = PrivateAttr(default=0)
|
||||
_timer: threading.Timer = PrivateAttr(default=None)
|
||||
_timer: threading.Timer | None = PrivateAttr(default=None)
|
||||
_lock: threading.Lock = PrivateAttr(default=None)
|
||||
|
||||
@model_validator(mode="after")
|
||||
|
||||
Reference in New Issue
Block a user