From 0ec3c37912f2f07af863530b364eafedb6522a5f Mon Sep 17 00:00:00 2001 From: Brandon Hancock Date: Thu, 27 Mar 2025 17:18:33 -0400 Subject: [PATCH] 99% done. Need to make docs match new example --- docs/concepts/lite-agent.mdx | 177 ++++++++++++++++++++++++ examples/flow_lite_agent.py | 81 +++++++++++ src/crewai/lite_agent.py | 253 ++++++++++++++++------------------- 3 files changed, 373 insertions(+), 138 deletions(-) create mode 100644 docs/concepts/lite-agent.mdx create mode 100644 examples/flow_lite_agent.py diff --git a/docs/concepts/lite-agent.mdx b/docs/concepts/lite-agent.mdx new file mode 100644 index 000000000..c88462638 --- /dev/null +++ b/docs/concepts/lite-agent.mdx @@ -0,0 +1,177 @@ +--- +title: LiteAgent +description: A lightweight, single-purpose agent for simple autonomous tasks within the CrewAI framework. +icon: feather +--- + +## Overview + +A `LiteAgent` is a streamlined version of CrewAI's Agent, designed for simpler, standalone tasks that don't require the full complexity of a crew-based workflow. It's perfect for quick automations, single-purpose tasks, or when you need a lightweight solution. + + +Think of a LiteAgent as a specialized worker that excels at individual tasks. While regular Agents are team players in a crew, LiteAgents are solo performers optimized for specific operations. + + +## LiteAgent Attributes + +| Attribute | Parameter | Type | Description | +| :------------------------------------ | :----------------- | :---------------------- | :--------------------------------------------------------------------------------------------- | +| **Role** | `role` | `str` | Defines the agent's function and expertise. | +| **Goal** | `goal` | `str` | The specific objective that guides the agent's actions. | +| **Backstory** | `backstory` | `str` | Provides context and personality to the agent. | +| **LLM** _(optional)_ | `llm` | `Union[str, LLM, Any]` | Language model powering the agent. Defaults to "gpt-4". | +| **Tools** _(optional)_ | `tools` | `List[BaseTool]` | Capabilities available to the agent. Defaults to an empty list. | +| **Verbose** _(optional)_ | `verbose` | `bool` | Enable detailed execution logs. Default is False. | +| **Response Format** _(optional)_ | `response_format` | `Type[BaseModel]` | Pydantic model for structured output. Optional. | + +## Creating a LiteAgent + +Here's a simple example of creating and using a LiteAgent in a flow: + +```python +from typing import List +from pydantic import BaseModel, Field +from crewai.flow.flow import Flow, start, listen +from crewai.lite_agent import LiteAgent +from crewai.tools import WebSearchTool + +# Define a structured output format +class MarketAnalysis(BaseModel): + key_trends: List[str] = Field(description="List of identified market trends") + market_size: str = Field(description="Estimated market size") + competitors: List[str] = Field(description="Major competitors in the space") + +# Define flow state +class MarketResearchState(BaseModel): + product: str = "" + analysis: MarketAnalysis = None + +# Create a flow class +class MarketResearchFlow(Flow[MarketResearchState]): + @start() + def initialize_research(self, product: str): + print(f"Starting market research for {product}") + self.state.product = product + + @listen(initialize_research) + async def analyze_market(self): + # Create a LiteAgent for market research + analyst = LiteAgent( + role="Market Research Analyst", + goal=f"Analyze the market for {self.state.product}", + backstory="You are an experienced market analyst with expertise in " + "identifying market trends and opportunities.", + tools=[WebSearchTool()], + verbose=True, + response_format=MarketAnalysis + ) + + # Define the research query + query = f""" + Research the market for {self.state.product}. Include: + 1. Key market trends + 2. Market size + 3. Major competitors + + Format your response according to the specified structure. + """ + + # Execute the analysis + result = await analyst.kickoff_async(query) + self.state.analysis = result.pydantic + return result.pydantic + + @listen(analyze_market) + def present_results(self): + analysis = self.state.analysis + print("\nMarket Analysis Results") + print("=====================") + + print("\nKey Market Trends:") + for trend in analysis.key_trends: + print(f"- {trend}") + + print(f"\nMarket Size: {analysis.market_size}") + + print("\nMajor Competitors:") + for competitor in analysis.competitors: + print(f"- {competitor}") + +# Usage example +import asyncio + +async def run_flow(): + flow = MarketResearchFlow() + result = await flow.kickoff(inputs={"product": "AI-powered chatbots"}) + return result + +# Run the flow +if __name__ == "__main__": + asyncio.run(run_flow()) + +## Key Features + +### 1. Simplified Setup +Unlike regular Agents, LiteAgents are designed for quick setup and standalone operation. They don't require crew configuration or task management. + +### 2. Structured Output +LiteAgents support Pydantic models for response formatting, making it easy to get structured, type-safe data from your agent's operations. + +### 3. Tool Integration +Just like regular Agents, LiteAgents can use tools to enhance their capabilities: +```python +from crewai.tools import SerperDevTool, CalculatorTool + +agent = LiteAgent( + role="Research Assistant", + goal="Find and analyze information", + tools=[SerperDevTool(), CalculatorTool()], + verbose=True +) +``` + +### 4. Async Support +LiteAgents support asynchronous execution through the `kickoff_async` method, making them suitable for non-blocking operations in your application. + +## Response Formatting + +LiteAgents support structured output through Pydantic models using the `response_format` parameter. This feature ensures type safety and consistent output structure, making it easier to work with agent responses in your application. + +### Basic Usage + +```python +from pydantic import BaseModel, Field + +class SearchResult(BaseModel): + title: str = Field(description="The title of the found content") + summary: str = Field(description="A brief summary of the content") + relevance_score: float = Field(description="Relevance score from 0 to 1") + +agent = LiteAgent( + role="Search Specialist", + goal="Find and summarize relevant information", + response_format=SearchResult +) + +result = await agent.kickoff_async("Find information about quantum computing") +print(f"Title: {result.pydantic.title}") +print(f"Summary: {result.pydantic.summary}") +print(f"Relevance: {result.pydantic.relevance_score}") +``` + +### Handling Responses + +When using `response_format`, the agent's response will be available in two forms: + +1. **Raw Response**: Access the unstructured string response + ```python + result = await agent.kickoff_async("Analyze the market") + print(result.raw) # Original LLM response + ``` + +2. **Structured Response**: Access the parsed Pydantic model + ```python + print(result.pydantic) # Parsed response as Pydantic model + print(result.pydantic.dict()) # Convert to dictionary + ``` + \ No newline at end of file diff --git a/examples/flow_lite_agent.py b/examples/flow_lite_agent.py new file mode 100644 index 000000000..85f0a03a0 --- /dev/null +++ b/examples/flow_lite_agent.py @@ -0,0 +1,81 @@ +from typing import List, cast + +from crewai_tools.tools.website_search.website_search_tool import WebsiteSearchTool +from pydantic import BaseModel, Field + +from crewai.flow.flow import Flow, listen, start +from crewai.lite_agent import LiteAgent + + +# Define a structured output format +class MarketAnalysis(BaseModel): + key_trends: List[str] = Field(description="List of identified market trends") + market_size: str = Field(description="Estimated market size") + competitors: List[str] = Field(description="Major competitors in the space") + + +# Define flow state +class MarketResearchState(BaseModel): + product: str = "" + analysis: MarketAnalysis | None = None + + +# Create a flow class +class MarketResearchFlow(Flow[MarketResearchState]): + @start() + def initialize_research(self): + print(f"Starting market research for {self.state.product}") + + @listen(initialize_research) + def analyze_market(self): + # Create a LiteAgent for market research + analyst = LiteAgent( + role="Market Research Analyst", + goal=f"Analyze the market for {self.state.product}", + backstory="You are an experienced market analyst with expertise in " + "identifying market trends and opportunities.", + llm="gpt-4o", + tools=[WebsiteSearchTool()], + verbose=True, + response_format=MarketAnalysis, + ) + + # Define the research query + query = f""" + Research the market for {self.state.product}. Include: + 1. Key market trends + 2. Market size + 3. Major competitors + + Format your response according to the specified structure. + """ + + # Execute the analysis + result = analyst.kickoff(query) + self.state.analysis = cast(MarketAnalysis, result.pydantic) + return result.pydantic + + @listen(analyze_market) + def present_results(self): + analysis = self.state.analysis + if analysis is None: + print("No analysis results available") + return + + print("\nMarket Analysis Results") + print("=====================") + + print("\nKey Market Trends:") + for trend in analysis.key_trends: + print(f"- {trend}") + + print(f"\nMarket Size: {analysis.market_size}") + + print("\nMajor Competitors:") + for competitor in analysis.competitors: + print(f"- {competitor}") + + +# Usage example +flow = MarketResearchFlow() +result = flow.kickoff(inputs={"product": "AI-powered chatbots"}) diff --git a/src/crewai/lite_agent.py b/src/crewai/lite_agent.py index e6cffef41..812219680 100644 --- a/src/crewai/lite_agent.py +++ b/src/crewai/lite_agent.py @@ -54,9 +54,6 @@ from crewai.utilities.events.tool_usage_events import ( ToolUsageFinishedEvent, ToolUsageStartedEvent, ) -from crewai.utilities.exceptions.context_window_exceeding_exception import ( - LLMContextLengthExceededException, -) from crewai.utilities.llm_utils import create_llm from crewai.utilities.printer import Printer from crewai.utilities.token_counter_callback import TokenCalcHandler @@ -116,8 +113,8 @@ 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, InstanceOf[LLM], Any] = Field( - description="Language model that will run the agent" + llm: Optional[Union[str, InstanceOf[LLM], Any]] = Field( + default=None, description="Language model that will run the agent" ) tools: List[BaseTool] = Field( default_factory=list, description="Tools at agent's disposal" @@ -199,6 +196,117 @@ class LiteAgent(BaseModel): """Return the original role for compatibility with tool interfaces.""" return self.role + def kickoff(self, messages: Union[str, List[Dict[str, str]]]) -> LiteAgentOutput: + """ + Execute the agent with the given messages. + + Args: + messages: Either a string query or a list of message dictionaries. + If a string is provided, it will be converted to a user message. + If a list is provided, each dict should have 'role' and 'content' keys. + + Returns: + LiteAgentOutput: The result of the agent execution. + """ + # Create agent info for event emission + agent_info = { + "role": self.role, + "goal": self.goal, + "backstory": self.backstory, + "tools": self._parsed_tools, + "verbose": self.verbose, + } + + try: + # Reset state for this run + self._iterations = 0 + self.tools_results = [] + + # Format messages for the LLM + self._messages = self._format_messages(messages) + + # Emit event for agent execution start + crewai_event_bus.emit( + self, + event=LiteAgentExecutionStartedEvent( + agent_info=agent_info, + tools=self._parsed_tools, + messages=messages, + ), + ) + + # Execute the agent using invoke loop + agent_finish = self._invoke_loop() + + formatted_result: Optional[BaseModel] = None + if self.response_format: + try: + # Cast to BaseModel to ensure type safety + result = self.response_format.model_validate_json( + agent_finish.output + ) + if isinstance(result, BaseModel): + formatted_result = result + except Exception as e: + self._printer.print( + content=f"Failed to parse output into response format: {str(e)}", + color="yellow", + ) + + # Calculate token usage metrics + usage_metrics = self._token_process.get_summary() + + # Create output + output = LiteAgentOutput( + raw=agent_finish.output, + pydantic=formatted_result, + agent_role=self.role, + usage_metrics=usage_metrics.model_dump() if usage_metrics else None, + ) + + # Emit completion event + crewai_event_bus.emit( + self, + event=LiteAgentExecutionCompletedEvent( + agent_info=agent_info, + output=agent_finish.output, + ), + ) + + return output + + except Exception as e: + self._printer.print( + content="Agent failed to reach a final answer. This is likely a bug - please report it.", + color="red", + ) + handle_unknown_error(self._printer, e) + # Emit error event + crewai_event_bus.emit( + self, + event=LiteAgentExecutionErrorEvent( + agent_info=agent_info, + error=str(e), + ), + ) + raise e + + async def kickoff_async( + self, messages: Union[str, List[Dict[str, str]]] + ) -> LiteAgentOutput: + """ + Execute the agent asynchronously with the given messages. + + Args: + messages: Either a string query or a list of message dictionaries. + If a string is provided, it will be converted to a user message. + If a list is provided, each dict should have 'role' and 'content' keys. + + Returns: + LiteAgentOutput: The result of the agent execution. + """ + return await asyncio.to_thread(self.kickoff, messages) + def _get_default_system_prompt(self) -> str: """Get the default system prompt for the agent.""" base_prompt = "" @@ -247,140 +355,13 @@ class LiteAgent(BaseModel): return formatted_messages - def kickoff(self, messages: Union[str, List[Dict[str, str]]]) -> LiteAgentOutput: - """ - Execute the agent with the given messages. - - Args: - messages: Either a string query or a list of message dictionaries. - If a string is provided, it will be converted to a user message. - If a list is provided, each dict should have 'role' and 'content' keys. - - Returns: - LiteAgentOutput: The result of the agent execution. - """ - return asyncio.run(self.kickoff_async(messages)) - - async def kickoff_async( - self, messages: Union[str, List[Dict[str, str]]] - ) -> LiteAgentOutput: - """ - Execute the agent asynchronously with the given messages. - - Args: - messages: Either a string query or a list of message dictionaries. - If a string is provided, it will be converted to a user message. - If a list is provided, each dict should have 'role' and 'content' keys. - - Returns: - LiteAgentOutput: The result of the agent execution. - - Raises: - Exception: If agent execution fails - """ - # Create agent info for event emission - agent_info = { - "role": self.role, - "goal": self.goal, - "backstory": self.backstory, - "tools": self._parsed_tools, - "verbose": self.verbose, - } - - try: - # Reset state for this run - self._iterations = 0 - self.tools_results = [] - - # Format messages for the LLM - self._messages = self._format_messages(messages) - - # Emit event for agent execution start - crewai_event_bus.emit( - self, - event=LiteAgentExecutionStartedEvent( - agent_info=agent_info, - tools=self._parsed_tools, - messages=messages, - ), - ) - - # Execute the agent using invoke loop - try: - agent_finish = await self._invoke_loop() - except Exception as e: - self._printer.print( - content="Agent failed to reach a final answer. This is likely a bug - please report it.", - color="red", - ) - handle_unknown_error(self._printer, e) - # Emit error event - crewai_event_bus.emit( - self, - event=LiteAgentExecutionErrorEvent( - agent_info=agent_info, - error=str(e), - ), - ) - raise e - - formatted_result: Optional[BaseModel] = None - if self.response_format: - try: - # Cast to BaseModel to ensure type safety - result = self.response_format.model_validate_json( - agent_finish.output - ) - if isinstance(result, BaseModel): - formatted_result = result - except Exception as e: - self._printer.print( - content=f"Failed to parse output into response format: {str(e)}", - color="yellow", - ) - - # Calculate token usage metrics - usage_metrics = self._token_process.get_summary() - - # Create output - output = LiteAgentOutput( - raw=agent_finish.output, - pydantic=formatted_result, - agent_role=self.role, - usage_metrics=usage_metrics.model_dump() if usage_metrics else None, - ) - - # Emit completion event - crewai_event_bus.emit( - self, - event=LiteAgentExecutionCompletedEvent( - agent_info=agent_info, - output=agent_finish.output, - ), - ) - - return output - - except Exception as e: - handle_unknown_error(self._printer, e) - # Emit error event - crewai_event_bus.emit( - self, - event=LiteAgentExecutionErrorEvent( - agent_info=agent_info, - error=str(e), - ), - ) - raise e - - async def _invoke_loop(self) -> AgentFinish: + def _invoke_loop(self) -> AgentFinish: """ Run the agent's thought process until it reaches a conclusion or max iterations. Returns: - str: The final result of the agent execution. + AgentFinish: The final result of the agent execution. """ - # Execute the agent loop formatted_answer = None while not isinstance(formatted_answer, AgentFinish): @@ -518,10 +499,6 @@ class LiteAgent(BaseModel): finally: self._iterations += 1 - # During the invoke loop, formatted_answer alternates between AgentAction - # (when the agent is using tools) and eventually becomes AgentFinish - # (when the agent reaches a final answer). This assertion confirms we've - # reached a final answer and helps type checking understand this transition. assert isinstance(formatted_answer, AgentFinish) self._show_logs(formatted_answer) return formatted_answer