Compare commits

...

4 Commits

Author SHA1 Message Date
Greyson LaLonde
4315f33e88 fix: cast dict values to str in _format_prompt
- Add str() casts for type safety
- These values are always strings when called from invoke
2025-07-22 10:34:10 -04:00
Greyson LaLonde
cf0a17f099 fix: update CrewAgentExecutor.invoke type signature
- Change inputs parameter from Dict[str, str] to Dict[str, Union[str, bool, None]]
- Matches actual usage where ask_for_human_input can be bool or None
2025-07-22 10:27:58 -04:00
Greyson LaLonde
a893e6030b fix: handle None agent_executor and type mismatch
- Add None check before accessing agent_executor attributes
- Convert task.human_input to bool for type compatibility
2025-07-22 10:21:31 -04:00
Greyson LaLonde
767bbd693d fix: add type annotation for agent_executor field
- Fixes 'Unresolved attribute reference' IDE warning
2025-07-22 10:16:53 -04:00
2 changed files with 33 additions and 10 deletions

View File

@@ -1,7 +1,18 @@
import shutil
import subprocess
import time
from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Tuple, Type, Union
from typing import (
Any,
Callable,
Dict,
List,
Literal,
Optional,
Sequence,
Tuple,
Type,
Union,
)
from pydantic import Field, InstanceOf, PrivateAttr, model_validator
@@ -76,6 +87,12 @@ class Agent(BaseAgent):
"""
_times_executed: int = PrivateAttr(default=0)
agent_executor: Optional[CrewAgentExecutor] = Field(
default=None,
init=False, # Not included in __init__ as it's created dynamically in create_agent_executor()
exclude=True, # Excluded from serialization to avoid circular references
description="The agent executor instance for running tasks. Created dynamically when needed.",
)
max_execution_time: Optional[int] = Field(
default=None,
description="Maximum execution time for an agent to execute a task",
@@ -162,7 +179,7 @@ class Agent(BaseAgent):
)
guardrail: Optional[Union[Callable[[Any], Tuple[bool, Any]], str]] = Field(
default=None,
description="Function or string description of a guardrail to validate agent output"
description="Function or string description of a guardrail to validate agent output",
)
guardrail_max_retries: int = Field(
default=3, description="Maximum number of retries when guardrail fails"
@@ -340,7 +357,6 @@ class Agent(BaseAgent):
self.knowledge_config.model_dump() if self.knowledge_config else {}
)
if self.knowledge or (self.crew and self.crew.knowledge):
crewai_event_bus.emit(
self,
@@ -531,6 +547,11 @@ class Agent(BaseAgent):
Returns:
The output of the agent.
"""
if not self.agent_executor:
raise ValueError(
"Agent executor not initialized. Call create_agent_executor() first."
)
return self.agent_executor.invoke(
{
"input": task_prompt,

View File

@@ -96,7 +96,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
)
)
def invoke(self, inputs: Dict[str, str]) -> Dict[str, Any]:
def invoke(self, inputs: Dict[str, Union[str, bool, None]]) -> Dict[str, Any]:
if "system" in self.prompt:
system_prompt = self._format_prompt(self.prompt.get("system", ""), inputs)
user_prompt = self._format_prompt(self.prompt.get("user", ""), inputs)
@@ -122,7 +122,6 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
handle_unknown_error(self._printer, e)
raise
if self.ask_for_human_input:
formatted_answer = self._handle_human_feedback(formatted_answer)
@@ -156,7 +155,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
messages=self.messages,
callbacks=self.callbacks,
printer=self._printer,
from_task=self.task
from_task=self.task,
)
formatted_answer = process_llm_response(answer, self.use_stop_words)
@@ -372,10 +371,13 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
training_data[agent_id] = agent_training_data
training_handler.save(training_data)
def _format_prompt(self, prompt: str, inputs: Dict[str, str]) -> str:
prompt = prompt.replace("{input}", inputs["input"])
prompt = prompt.replace("{tool_names}", inputs["tool_names"])
prompt = prompt.replace("{tools}", inputs["tools"])
def _format_prompt(
self, prompt: str, inputs: Dict[str, Union[str, bool, None]]
) -> str:
# Cast to str to satisfy type checker - these are always strings when called
prompt = prompt.replace("{input}", str(inputs["input"]))
prompt = prompt.replace("{tool_names}", str(inputs["tool_names"]))
prompt = prompt.replace("{tools}", str(inputs["tools"]))
return prompt
def _handle_human_feedback(self, formatted_answer: AgentFinish) -> AgentFinish: