andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-09 16:18:30 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Brandon Hancock
2025-03-20 08:00:32 -04:00
parent e9fa9c5700
commit defb0c55e6
9 changed files with 374 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/crewai/agent.py
View File

@@ -18,6 +18,11 @@ from crewai.task import Task
from crewai.tools import BaseTool
from crewai.tools.agent_tools.agent_tools import AgentTools
from crewai.utilities import Converter, Prompts
from crewai.utilities.agent_utils import (
get_tool_names,
parse_tools,
render_text_description_and_args,
)
from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_FILE
from crewai.utilities.converter import generate_model_description
from crewai.utilities.events.agent_events import (
@@ -301,7 +306,7 @@ class Agent(BaseAgent):
An instance of the CrewAgentExecutor class.
"""
tools = tools or self.tools or []
parsed_tools = self._parse_tools(tools)
parsed_tools = parse_tools(tools)
prompt = Prompts(
agent=self,
@@ -331,8 +336,8 @@ class Agent(BaseAgent):
stop_words=stop_words,
max_iter=self.max_iter,
tools_handler=self.tools_handler,
tools_names=self.__tools_names(parsed_tools),
tools_description=self._render_text_description_and_args(parsed_tools),
tools_names=get_tool_names(parsed_tools),
tools_description=render_text_description_and_args(parsed_tools),
step_callback=self.step_callback,
function_calling_llm=self.function_calling_llm,
respect_context_window=self.respect_context_window,
@@ -367,25 +372,6 @@ class Agent(BaseAgent):
def get_output_converter(self, llm, text, model, instructions):
return Converter(llm=llm, text=text, model=model, instructions=instructions)
def _parse_tools(self, tools: List[Any]) -> List[Any]: # type: ignore
"""Parse tools to be used for the task."""
tools_list = []
try:
# tentatively try to import from crewai_tools import BaseTool as CrewAITool
from crewai.tools import BaseTool as CrewAITool
for tool in tools:
if isinstance(tool, CrewAITool):
tools_list.append(tool.to_structured_tool())
else:
tools_list.append(tool)
except ModuleNotFoundError:
tools_list = []
for tool in tools:
tools_list.append(tool)
return tools_list
def _training_handler(self, task_prompt: str) -> str:
"""Handle training data for the agent task prompt to improve output on Training."""
if data := CrewTrainingHandler(TRAINING_DATA_FILE).load():
@@ -431,23 +417,6 @@ class Agent(BaseAgent):
return description
def _render_text_description_and_args(self, tools: List[BaseTool]) -> str:
"""Render the tool name, description, and args in plain text.
Output will be in the format of:
.. code-block:: markdown
search: This tool is used for search, args: {"query": {"type": "string"}}
calculator: This tool is used for math, \
args: {"expression": {"type": "string"}}
"""
tool_strings = []
for tool in tools:
tool_strings.append(tool.description)
return "\n".join(tool_strings)
def _validate_docker_installation(self) -> None:
"""Check if Docker is installed and running."""
if not shutil.which("docker"):
@@ -467,10 +436,6 @@ class Agent(BaseAgent):
f"Docker is not running. Please start Docker to use code execution with agent: {self.role}"
)
@staticmethod
def __tools_names(tools) -> str:
return ", ".join([t.name for t in tools])
def __repr__(self):
return f"Agent(role={self.role}, goal={self.goal}, backstory={self.backstory})"

8
src/crewai/agents/agent_builder/base_agent.py
View File

@@ -71,8 +71,6 @@ class BaseAgent(ABC, BaseModel):
Interpolate inputs into the agent description and backstory.
set_cache_handler(cache_handler: CacheHandler) -> None:
Set the cache handler for the agent.
increment_formatting_errors() -> None:
Increment formatting errors.
copy() -> "BaseAgent":
Create a copy of the agent.
set_rpm_controller(rpm_controller: RPMController) -> None:
@@ -90,9 +88,6 @@ class BaseAgent(ABC, BaseModel):
_original_backstory: Optional[str] = PrivateAttr(default=None)
_token_process: TokenProcess = PrivateAttr(default_factory=TokenProcess)
id: UUID4 = Field(default_factory=uuid.uuid4, frozen=True)
formatting_errors: int = Field(
default=0, description="Number of formatting errors."
)
role: str = Field(description="Role of the agent")
goal: str = Field(description="Objective of the agent")
backstory: str = Field(description="Backstory of the agent")
@@ -349,9 +344,6 @@ class BaseAgent(ABC, BaseModel):
self.tools_handler.cache = cache_handler
self.create_agent_executor()
def increment_formatting_errors(self) -> None:
self.formatting_errors += 1
def set_rpm_controller(self, rpm_controller: RPMController) -> None:
"""Set the rpm controller for the agent.

143
src/crewai/agents/crew_agent_executor.py
View File

@@ -17,6 +17,15 @@ from crewai.llm import LLM
from crewai.tools.base_tool import BaseTool
from crewai.tools.tool_usage import ToolUsage, ToolUsageErrorException
from crewai.utilities import I18N, Printer
from crewai.utilities.agent_utils import (
enforce_rpm_limit,
format_answer,
format_message_for_llm,
get_llm_response,
handle_max_iterations_exceeded,
has_reached_max_iterations,
process_llm_response,
)
from crewai.utilities.constants import MAX_LLM_RETRY, TRAINING_DATA_FILE
from crewai.utilities.events import (
ToolUsageErrorEvent,
@@ -94,11 +103,11 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
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)
self.messages.append(self._format_msg(system_prompt, role="system"))
self.messages.append(self._format_msg(user_prompt))
self.messages.append(format_message_for_llm(system_prompt, role="system"))
self.messages.append(format_message_for_llm(user_prompt))
else:
user_prompt = self._format_prompt(self.prompt.get("prompt", ""), inputs)
self.messages.append(self._format_msg(user_prompt))
self.messages.append(format_message_for_llm(user_prompt))
self._show_start_logs()
@@ -135,16 +144,25 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
formatted_answer = None
while not isinstance(formatted_answer, AgentFinish):
try:
if self._has_reached_max_iterations():
formatted_answer = self._handle_max_iterations_exceeded(
formatted_answer
if has_reached_max_iterations(self.iterations, self.max_iter):
formatted_answer = handle_max_iterations_exceeded(
formatted_answer,
printer=self._printer,
i18n=self._i18n,
messages=self.messages,
llm=self.llm,
callbacks=self.callbacks,
)
break
self._enforce_rpm_limit()
enforce_rpm_limit(self.request_within_rpm_limit)
answer = self._get_llm_response()
formatted_answer = self._process_llm_response(answer)
answer = get_llm_response(
llm=self.llm,
messages=self.messages,
callbacks=self.callbacks,
printer=self._printer,
)
formatted_answer = process_llm_response(answer, self.use_stop_words)
if isinstance(formatted_answer, AgentAction):
tool_result = self._execute_tool_and_check_finality(
@@ -192,50 +210,6 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
color="red",
)
def _has_reached_max_iterations(self) -> bool:
"""Check if the maximum number of iterations has been reached."""
return self.iterations >= self.max_iter
def _enforce_rpm_limit(self) -> None:
"""Enforce the requests per minute (RPM) limit if applicable."""
if self.request_within_rpm_limit:
self.request_within_rpm_limit()
def _get_llm_response(self) -> str:
"""Call the LLM and return the response, handling any invalid responses."""
try:
answer = self.llm.call(
self.messages,
callbacks=self.callbacks,
)
except Exception as e:
self._printer.print(
content=f"Error during LLM call: {e}",
color="red",
)
raise e
if not answer:
self._printer.print(
content="Received None or empty response from LLM call.",
color="red",
)
raise ValueError("Invalid response from LLM call - None or empty.")
return answer
def _process_llm_response(self, answer: str) -> Union[AgentAction, AgentFinish]:
"""Process the LLM response and format it into an AgentAction or AgentFinish."""
if not self.use_stop_words:
try:
# Preliminary parsing to check for errors.
self._format_answer(answer)
except OutputParserException as e:
if FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE in e.error:
answer = answer.split("Observation:")[0].strip()
return self._format_answer(answer)
def _handle_agent_action(
self, formatted_answer: AgentAction, tool_result: ToolResult
) -> Union[AgentAction, AgentFinish]:
@@ -272,7 +246,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
def _append_message(self, text: str, role: str = "assistant") -> None:
"""Append a message to the message list with the given role."""
self.messages.append(self._format_msg(text, role=role))
self.messages.append(format_message_for_llm(text, role=role))
def _handle_output_parser_exception(self, e: OutputParserException) -> AgentAction:
"""Handle OutputParserException by updating messages and formatted_answer."""
@@ -430,10 +404,10 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
for group in messages_groups:
summary = self.llm.call(
[
self._format_msg(
format_message_for_llm(
self._i18n.slice("summarizer_system_message"), role="system"
),
self._format_msg(
format_message_for_llm(
self._i18n.slice("summarize_instruction").format(group=group),
),
],
@@ -444,7 +418,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
merged_summary = " ".join(str(content) for content in summarized_contents)
self.messages = [
self._format_msg(
format_message_for_llm(
self._i18n.slice("summary").format(merged_summary=merged_summary)
)
]
@@ -517,13 +491,6 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
prompt = prompt.replace("{tools}", inputs["tools"])
return prompt
def _format_answer(self, answer: str) -> Union[AgentAction, AgentFinish]:
return CrewAgentParser(agent=self.agent).parse(answer)
def _format_msg(self, prompt: str, role: str = "user") -> Dict[str, str]:
prompt = prompt.rstrip()
return {"role": role, "content": prompt}
def _handle_human_feedback(self, formatted_answer: AgentFinish) -> AgentFinish:
"""Handle human feedback with different flows for training vs regular use.
@@ -550,7 +517,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
"""Process feedback for training scenarios with single iteration."""
self._handle_crew_training_output(initial_answer, feedback)
self.messages.append(
self._format_msg(
format_message_for_llm(
self._i18n.slice("feedback_instructions").format(feedback=feedback)
)
)
@@ -579,7 +546,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
def _process_feedback_iteration(self, feedback: str) -> AgentFinish:
"""Process a single feedback iteration."""
self.messages.append(
self._format_msg(
format_message_for_llm(
self._i18n.slice("feedback_instructions").format(feedback=feedback)
)
)
@@ -604,45 +571,3 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
),
color="red",
)
def _handle_max_iterations_exceeded(self, formatted_answer):
"""
Handles the case when the maximum number of iterations is exceeded.
Performs one more LLM call to get the final answer.
Parameters:
formatted_answer: The last formatted answer from the agent.
Returns:
The final formatted answer after exceeding max iterations.
"""
self._printer.print(
content="Maximum iterations reached. Requesting final answer.",
color="yellow",
)
if formatted_answer and hasattr(formatted_answer, "text"):
assistant_message = (
formatted_answer.text + f'\n{self._i18n.errors("force_final_answer")}'
)
else:
assistant_message = self._i18n.errors("force_final_answer")
self.messages.append(self._format_msg(assistant_message, role="assistant"))
# Perform one more LLM call to get the final answer
answer = self.llm.call(
self.messages,
callbacks=self.callbacks,
)
if answer is None or answer == "":
self._printer.print(
content="Received None or empty response from LLM call.",
color="red",
)
raise ValueError("Invalid response from LLM call - None or empty.")
formatted_answer = self._format_answer(answer)
# Return the formatted answer, regardless of its type
return formatted_answer

19
src/crewai/agents/parser.py
View File

@@ -65,10 +65,20 @@ class CrewAgentParser:
"""
_i18n: I18N = I18N()
agent: Any = None
def __init__(self, agent: Any):
self.agent = agent
@staticmethod
def parse_text(text: str) -> Union[AgentAction, AgentFinish]:
"""
Static method to parse text into an AgentAction or AgentFinish without needing to instantiate the class.
Args:
text: The text to parse.
Returns:
Either an AgentAction or AgentFinish based on the parsed content.
"""
parser = CrewAgentParser()
return parser.parse(text)
def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
thought = self._extract_thought(text)
@@ -104,21 +114,18 @@ class CrewAgentParser:
return AgentFinish(thought, final_answer, text)
if not re.search(r"Action\s*\d*\s*:[\s]*(.*?)", text, re.DOTALL):
self.agent.increment_formatting_errors()
raise OutputParserException(
f"{MISSING_ACTION_AFTER_THOUGHT_ERROR_MESSAGE}\n{self._i18n.slice('final_answer_format')}",
)
elif not re.search(
r"[\s]*Action\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)", text, re.DOTALL
):
self.agent.increment_formatting_errors()
raise OutputParserException(
MISSING_ACTION_INPUT_AFTER_ACTION_ERROR_MESSAGE,
)
else:
format = self._i18n.slice("format_without_tools")
error = f"{format}"
self.agent.increment_formatting_errors()
raise OutputParserException(
error,
)

242
src/crewai/lite_agent.py
View File

@@ -1,8 +1,8 @@
import asyncio
import json
import re
import uuid # Add import for generating unique keys
from typing import Any, Dict, List, Optional, Type, Union, cast
import uuid
from typing import Any, Callable, Dict, List, Optional, Type, Union, cast
from pydantic import BaseModel, Field, PrivateAttr, model_validator
@@ -18,9 +18,18 @@ from crewai.agents.parser import (
from crewai.agents.tools_handler import ToolsHandler
from crewai.llm import LLM
from crewai.tools.base_tool import BaseTool
from crewai.tools.structured_tool import CrewStructuredTool
from crewai.tools.tool_calling import ToolCalling
from crewai.types.usage_metrics import UsageMetrics
from crewai.utilities import I18N
from crewai.utilities.agent_utils import (
enforce_rpm_limit,
get_llm_response,
get_tool_names,
handle_max_iterations_exceeded,
has_reached_max_iterations,
parse_tools,
process_llm_response,
render_text_description_and_args,
)
from crewai.utilities.events.agent_events import (
LiteAgentExecutionCompletedEvent,
LiteAgentExecutionErrorEvent,
@@ -29,6 +38,7 @@ from crewai.utilities.events.agent_events import (
from crewai.utilities.events.crewai_event_bus import crewai_event_bus
from crewai.utilities.events.tool_usage_events import ToolUsageStartedEvent
from crewai.utilities.llm_utils import create_llm
from crewai.utilities.printer import Printer
from crewai.utilities.token_counter_callback import TokenCalcHandler
@@ -85,7 +95,6 @@ class LiteAgent(BaseModel):
max_iterations: Maximum number of iterations for tool usage.
max_execution_time: Maximum execution time in seconds.
response_format: Optional Pydantic model for structured output.
system_prompt: Custom system prompt to override the default.
"""
model_config = {"arbitrary_types_allowed": True}
@@ -93,9 +102,7 @@ class LiteAgent(BaseModel):
role: str = Field(description="Role of the agent")
goal: str = Field(description="Goal of the agent")
backstory: str = Field(description="Backstory of the agent")
llm: Union[str, LLM, Any] = Field(
description="Language model that will run the agent", default=None
)
llm: LLM = Field(description="Language model that will run the agent")
tools: List[BaseTool] = Field(
default_factory=list, description="Tools at agent's disposal"
)
@@ -111,9 +118,7 @@ class LiteAgent(BaseModel):
response_format: Optional[Type[BaseModel]] = Field(
default=None, description="Pydantic model for structured output"
)
system_prompt: Optional[str] = Field(
default=None, description="Custom system prompt to override default"
)
step_callback: Optional[Any] = Field(
default=None,
description="Callback to be executed after each step of the agent execution.",
@@ -134,6 +139,17 @@ class LiteAgent(BaseModel):
_formatting_errors: int = PrivateAttr(default=0)
_tools_errors: int = PrivateAttr(default=0)
_delegations: Dict[str, int] = PrivateAttr(default_factory=dict)
# Internationalization
_i18n: I18N = PrivateAttr(default_factory=I18N)
_printer: Printer = PrivateAttr(default_factory=Printer)
request_within_rpm_limit: Optional[Callable[[], bool]] = Field(
default=None,
description="Callback to check if the request is within the RPM limit",
)
use_stop_words: bool = Field(
default=True,
description="Whether to use stop words to prevent the LLM from using tools",
)
@model_validator(mode="after")
def setup_llm(self):
@@ -143,6 +159,7 @@ class LiteAgent(BaseModel):
if not isinstance(self.llm, LLM):
self.llm = create_llm(self.llm)
self.use_stop_words = self.llm.supports_stop_words()
return self
@@ -158,149 +175,22 @@ class LiteAgent(BaseModel):
def _get_default_system_prompt(self) -> str:
"""Get the default system prompt for the agent."""
prompt = f"""You are a helpful AI assistant acting as {self.role}.
Your goal is: {self.goal}
Your backstory: {self.backstory}
When using tools, you MUST follow this EXACT format with the precise spacing and newlines as shown:
Thought: <your reasoning about what needs to be done>
Action: <tool_name>
Action Input: {{
"parameter1": "value1",
"parameter2": "value2"
}}
Observation: [Result of the tool execution will appear here]
You can then continue with another tool:
Thought: <your reasoning about what to do next>
Action: <another_tool_name>
Action Input: {{
"parameter1": "value1"
}}
Observation: [Result of the tool execution will appear here]
When you have a final answer and don't need to use any more tools, respond with:
Thought: <your reasoning about the final answer>
Final Answer: <your final answer to the user>
Here's a concrete example of proper tool usage:
Thought: I need to find out the weather in New York City.
Action: get_weather
Action Input: {{
"city": "New York City"
}}
Observation: [The weather result would appear here]
Thought: Now I need to save this weather data.
Action: save_weather_data
Action Input: {{
"filename": "weather_history.txt"
}}
Observation: [The result of saving would appear here]
Thought: I now have all the information I need to answer the user's question.
Final Answer: The weather in New York City today is [weather details] and I've saved this information to the weather_history.txt file.
Always maintain the exact format shown above, with blank lines between sections and properly formatted inputs for tools.
"""
return prompt
def _format_tools_description(self) -> str:
"""Format tools into a string for the prompt."""
if not self.tools:
return "You don't have any tools available."
tools_str = "You have access to the following tools:\n\n"
for tool in self.tools:
tools_str += f"Tool: {tool.name}\n"
tools_str += f"Description: {tool.description}\n"
if hasattr(tool, "args_schema"):
schema_info = ""
try:
if hasattr(tool.args_schema, "model_json_schema"):
schema = tool.args_schema.model_json_schema()
if "properties" in schema:
schema_info = ", ".join(
[
f"{k}: {v.get('type', 'any')}"
for k, v in schema["properties"].items()
]
)
else:
schema_info = str(schema)
except Exception:
schema_info = "Unable to parse schema"
tools_str += f"Parameters: {schema_info}\n"
tools_str += "\n"
return tools_str
def _get_tools_names(self) -> str:
"""Get a comma-separated list of tool names."""
return ", ".join([tool.name for tool in self.tools])
def _parse_tools(self) -> List[Dict[str, Any]]:
"""Parse tools to be used by the agent."""
tools_list = []
for tool in self.tools:
try:
# First try to use the to_structured_tool method if available
if hasattr(tool, "to_structured_tool"):
structured_tool = tool.to_structured_tool()
if structured_tool and isinstance(structured_tool, dict):
tools_list.append(structured_tool)
continue
# Fall back to manual conversion if to_structured_tool is not available or fails
tool_dict = {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
},
}
# Add args schema if available
if hasattr(tool, "args_schema") and tool.args_schema:
try:
if hasattr(tool.args_schema, "model_json_schema"):
tool_dict["function"][
"parameters"
] = tool.args_schema.model_json_schema()
except Exception as e:
if self.verbose:
print(
f"Warning: Could not get schema for tool {tool.name}: {e}"
)
tools_list.append(tool_dict)
except Exception as e:
if self.verbose:
print(f"Error converting tool {tool.name}: {e}")
return tools_list
if self.tools:
# Use the prompt template for agents with tools
return self._i18n.slice("lite_agent_system_prompt_with_tools").format(
role=self.role,
backstory=self.backstory,
goal=self.goal,
tools=format_tools_description(),
tool_names=self._get_tools_names(),
)
else:
# Use the prompt template for agents without tools
return self._i18n.slice("lite_agent_system_prompt_without_tools").format(
role=self.role,
backstory=self.backstory,
goal=self.goal,
)
def _format_messages(
self, messages: Union[str, List[Dict[str, str]]]
@@ -309,13 +199,10 @@ Always maintain the exact format shown above, with blank lines between sections
if isinstance(messages, str):
messages = [{"role": "user", "content": messages}]
system_prompt = self.system_prompt or self._get_default_system_prompt()
tools_description = self._format_tools_description()
system_prompt = self._get_default_system_prompt()
# Add system message at the beginning
formatted_messages = [
{"role": "system", "content": f"{system_prompt}\n\n{tools_description}"}
]
formatted_messages = [{"role": "system", "content": system_prompt}]
# Add the rest of the messages
formatted_messages.extend(messages)
@@ -453,9 +340,6 @@ Always maintain the exact format shown above, with blank lines between sections
# Format messages for the LLM
self._messages = self._format_messages(messages)
# Get the original query for event emission
query = messages if isinstance(messages, str) else messages[-1]["content"]
# Create agent info for event emission
agent_info = {
"role": self.role,
@@ -471,7 +355,7 @@ Always maintain the exact format shown above, with blank lines between sections
event=LiteAgentExecutionStartedEvent(
agent_info=agent_info,
tools=self.tools,
task_prompt=query,
messages=messages,
),
)
@@ -548,15 +432,35 @@ Always maintain the exact format shown above, with blank lines between sections
callbacks = [token_callback]
# Prepare tool configurations
parsed_tools = self._parse_tools()
tools_description = self._format_tools_description()
tools_names = self._get_tools_names()
# Create a mapping of tool names to tools for easier lookup
tool_map = {tool.name: tool for tool in self.tools}
parsed_tools = parse_tools(self.tools)
tools_description = render_text_description_and_args(parsed_tools)
tools_names = get_tool_names(parsed_tools)
# Execute the agent loop
formatted_answer = None
while not isinstance(formatted_answer, AgentFinish):
try :
if has_reached_max_iterations(self._iterations, self.max_iterations):
formatted_answer = handle_max_iterations_exceeded(
formatted_answer,
printer=self._printer,
i18n=self._i18n,
messages=self._messages,
llm=self.llm,
callbacks=callbacks,
)
enforce_rpm_limit(self.request_within_rpm_limit)
answer = get_llm_response(
llm=self.llm,
messages=self._messages,
callbacks=callbacks,
printer=self._printer,
)
formatted_answer = process_llm_response(answer, self.use_stop_words)
while self._iterations < self.max_iterations:
try:
# Execute the LLM

4
src/crewai/translations/en.json
View File

@@ -24,7 +24,9 @@
"manager_request": "Your best answer to your coworker asking you this, accounting for the context shared.",
"formatted_task_instructions": "Ensure your final answer contains only the content in the following format: {output_format}\n\nEnsure the final output does not include any code block markers like ```json or ```python.",
"conversation_history_instruction": "You are a member of a crew collaborating to achieve a common goal. Your task is a specific action that contributes to this larger objective. For additional context, please review the conversation history between you and the user that led to the initiation of this crew. Use any relevant information or feedback from the conversation to inform your task execution and ensure your response aligns with both the immediate task and the crew's overall goals.",
"feedback_instructions": "User feedback: {feedback}\nInstructions: Use this feedback to enhance the next output iteration.\nNote: Do not respond or add commentary."
"feedback_instructions": "User feedback: {feedback}\nInstructions: Use this feedback to enhance the next output iteration.\nNote: Do not respond or add commentary.",
"lite_agent_system_prompt_with_tools": "You are {role}. {backstory}\nYour personal goal is: {goal}\n\nYou ONLY have access to the following tools, and should NEVER make up tools that are not listed here:\n\n{tools}\n\nIMPORTANT: Use the following format in your response:\n\n```\nThought: you should always think about what to do\nAction: the action to take, only one name of [{tool_names}], just the name, exactly as it's written.\nAction Input: the input to the action, just a simple JSON object, enclosed in curly braces, using \" to wrap keys and values.\nObservation: the result of the action\n```\n\nOnce all necessary information is gathered, return the following format:\n\n```\nThought: I now know the final answer\nFinal Answer: the final answer to the original input question\n```",
"lite_agent_system_prompt_without_tools": "You are {role}. {backstory}\nYour personal goal is: {goal}\n\nTo give my best complete final answer to the task respond using the exact following format:\n\nThought: I now can give a great answer\nFinal Answer: Your final answer must be the great and the most complete as possible, it must be outcome described.\n\nI MUST use these formats, my job depends on it!"
},
"errors": {
"force_final_answer_error": "You can't keep going, here is the best final answer you generated:\n\n {formatted_answer}",

176
src/crewai/utilities/agent_utils.py Normal file
View File

@@ -0,0 +1,176 @@
from typing import Any, Callable, Dict, List, Optional, Union
from crewai.agents.parser import (
FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE,
AgentAction,
AgentFinish,
CrewAgentParser,
OutputParserException,
)
from crewai.llm import LLM
from crewai.tools import BaseTool as CrewAITool
from crewai.tools.base_tool import BaseTool
from crewai.utilities.i18n import I18N
from crewai.utilities.printer import Printer
def parse_tools(tools: List[Any]) -> List[Any]:
"""Parse tools to be used for the task."""
tools_list = []
try:
for tool in tools:
if isinstance(tool, CrewAITool):
tools_list.append(tool.to_structured_tool())
else:
tools_list.append(tool)
except ModuleNotFoundError:
tools_list = []
for tool in tools:
tools_list.append(tool)
return tools_list
def get_tool_names(tools: List[Any]) -> str:
"""Get the names of the tools."""
return ", ".join([t.name for t in tools])
def render_text_description_and_args(tools: List[BaseTool]) -> str:
"""Render the tool name, description, and args in plain text.
Output will be in the format of:
.. code-block:: markdown
search: This tool is used for search, args: {"query": {"type": "string"}}
calculator: This tool is used for math, \
args: {"expression": {"type": "string"}}
"""
tool_strings = []
for tool in tools:
tool_strings.append(tool.description)
return "\n".join(tool_strings)
def has_reached_max_iterations(iterations: int, max_iterations: int) -> bool:
"""Check if the maximum number of iterations has been reached."""
return iterations >= max_iterations
def handle_max_iterations_exceeded(
formatted_answer: Union[AgentAction, AgentFinish, None],
printer: Printer,
i18n: I18N,
messages: List[Dict[str, str]],
llm: LLM,
callbacks: List[Any],
) -> Union[AgentAction, AgentFinish]:
"""
Handles the case when the maximum number of iterations is exceeded.
Performs one more LLM call to get the final answer.
Parameters:
formatted_answer: The last formatted answer from the agent.
Returns:
The final formatted answer after exceeding max iterations.
"""
printer.print(
content="Maximum iterations reached. Requesting final answer.",
color="yellow",
)
if formatted_answer and hasattr(formatted_answer, "text"):
assistant_message = (
formatted_answer.text + f'\n{i18n.errors("force_final_answer")}'
)
else:
assistant_message = i18n.errors("force_final_answer")
messages.append(format_message_for_llm(assistant_message, role="assistant"))
# Perform one more LLM call to get the final answer
answer = llm.call(
messages,
callbacks=callbacks,
)
if answer is None or answer == "":
printer.print(
content="Received None or empty response from LLM call.",
color="red",
)
raise ValueError("Invalid response from LLM call - None or empty.")
formatted_answer = format_answer(answer)
# Return the formatted answer, regardless of its type
return formatted_answer
def format_message_for_llm(prompt: str, role: str = "user") -> Dict[str, str]:
prompt = prompt.rstrip()
return {"role": role, "content": prompt}
def format_answer(answer: str) -> Union[AgentAction, AgentFinish]:
"""Format a response from the LLM into an AgentAction or AgentFinish."""
try:
return CrewAgentParser.parse_text(answer)
except Exception:
# If parsing fails, return a default AgentFinish
return AgentFinish(
thought="Failed to parse LLM response",
output=answer,
text=answer,
)
def enforce_rpm_limit(
request_within_rpm_limit: Optional[Callable[[], bool]] = None
) -> None:
"""Enforce the requests per minute (RPM) limit if applicable."""
if request_within_rpm_limit:
request_within_rpm_limit()
def get_llm_response(
llm: LLM, messages: List[Dict[str, str]], callbacks: List[Any], printer: Printer
) -> str:
"""Call the LLM and return the response, handling any invalid responses."""
try:
answer = llm.call(
messages,
callbacks=callbacks,
)
except Exception as e:
printer.print(
content=f"Error during LLM call: {e}",
color="red",
)
raise e
if not answer:
printer.print(
content="Received None or empty response from LLM call.",
color="red",
)
raise ValueError("Invalid response from LLM call - None or empty.")
return answer
def process_llm_response(
answer: str, use_stop_words: bool
) -> Union[AgentAction, AgentFinish]:
"""Process the LLM response and format it into an AgentAction or AgentFinish."""
if not use_stop_words:
try:
# Preliminary parsing to check for errors.
format_answer(answer)
except OutputParserException as e:
if FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE in e.error:
answer = answer.split("Observation:")[0].strip()
return format_answer(answer)

4
src/crewai/utilities/events/agent_events.py
View File

@@ -1,4 +1,4 @@
from typing import TYPE_CHECKING, Any, Dict, Optional, Sequence, Union
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.tools.base_tool import BaseTool
@@ -46,7 +46,7 @@ class LiteAgentExecutionStartedEvent(CrewEvent):
agent_info: Dict[str, Any]
tools: Optional[Sequence[Union[BaseTool, CrewStructuredTool]]]
task_prompt: str
messages: Union[str, List[Dict[str, str]]]
type: str = "lite_agent_execution_started"
model_config = {"arbitrary_types_allowed": True}

65
test_lite_agent.py Normal file
View File

@@ -0,0 +1,65 @@
from crewai import LLM
from crewai.lite_agent import LiteAgent
from crewai.tools import BaseTool
# A simple test tool
class TestTool(BaseTool):
name = "test_tool"
description = "A simple test tool"
def _run(self, query: str) -> str:
return f"Test result for: {query}"
# Test with tools
def test_with_tools():
llm = LLM(model="gpt-4o")
agent = LiteAgent(
role="Test Agent",
goal="Test the system prompt formatting",
backstory="I am a test agent created to verify the system prompt works correctly.",
llm=llm,
tools=[TestTool()],
verbose=True,
)
# Get the system prompt
system_prompt = agent._get_default_system_prompt()
print("\n=== System Prompt (with tools) ===")
print(system_prompt)
# Test a simple query
response = agent.kickoff("Hello, can you help me?")
print("\n=== Agent Response ===")
print(response)
# Test without tools
def test_without_tools():
llm = LLM(model="gpt-4o")
agent = LiteAgent(
role="Test Agent",
goal="Test the system prompt formatting",
backstory="I am a test agent created to verify the system prompt works correctly.",
llm=llm,
verbose=True,
)
# Get the system prompt
system_prompt = agent._get_default_system_prompt()
print("\n=== System Prompt (without tools) ===")
print(system_prompt)
# Test a simple query
response = agent.kickoff("Hello, can you help me?")
print("\n=== Agent Response ===")
print(response)
if __name__ == "__main__":
print("Testing LiteAgent with tools...")
test_with_tools()
print("\n\nTesting LiteAgent without tools...")
test_without_tools()