diff --git a/examples/flow_lite_agent.py b/examples/flow_lite_agent.py new file mode 100644 index 000000000..2df35d119 --- /dev/null +++ b/examples/flow_lite_agent.py @@ -0,0 +1,80 @@ +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 + + +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/examples/lite_agent_example.py b/examples/lite_agent_example.py new file mode 100644 index 000000000..40e064579 --- /dev/null +++ b/examples/lite_agent_example.py @@ -0,0 +1,140 @@ +""" +Example script demonstrating how to use the LiteAgent. + +This example shows how to create and use a LiteAgent for simple interactions +without the need for a full crew or task-based workflow. +""" + +import asyncio +from typing import Any, Dict, cast + +from pydantic import BaseModel, Field + +from crewai.lite_agent import LiteAgent +from crewai.tools.base_tool import BaseTool + + +# Define custom tools +class WebSearchTool(BaseTool): + """Tool for searching the web for information.""" + + name: str = "search_web" + description: str = "Search the web for information about a topic." + + def _run(self, query: str) -> str: + """Search the web for information about a topic.""" + # This is a mock implementation + if "tokyo" in query.lower(): + return "Tokyo's population in 2023 was approximately 21 million people in the city proper, and 37 million in the greater metropolitan area." + elif "climate change" in query.lower() and "coral" in query.lower(): + return "Climate change severely impacts coral reefs through: 1) Ocean warming causing coral bleaching, 2) Ocean acidification reducing calcification, 3) Sea level rise affecting light availability, 4) Increased storm frequency damaging reef structures. Sources: NOAA Coral Reef Conservation Program, Global Coral Reef Alliance." + else: + return f"Found information about {query}: This is a simulated search result for demonstration purposes." + + +class CalculatorTool(BaseTool): + """Tool for performing calculations.""" + + name: str = "calculate" + description: str = "Calculate the result of a mathematical expression." + + def _run(self, expression: str) -> str: + """Calculate the result of a mathematical expression.""" + try: + result = eval(expression, {"__builtins__": {}}) + return f"The result of {expression} is {result}" + except Exception as e: + return f"Error calculating {expression}: {str(e)}" + + +# Define a custom response format using Pydantic +class ResearchResult(BaseModel): + """Structure for research results.""" + + main_findings: str = Field(description="The main findings from the research") + key_points: list[str] = Field(description="List of key points") + sources: list[str] = Field(description="List of sources used") + + +async def main(): + # Create tools + web_search_tool = WebSearchTool() + calculator_tool = CalculatorTool() + + # Create a LiteAgent with a specific role, goal, and backstory + agent = LiteAgent( + role="Research Analyst", + goal="Provide accurate and concise information on requested topics", + backstory="You are an expert research analyst with years of experience in gathering and synthesizing information from various sources.", + llm="gpt-4", + tools=[web_search_tool, calculator_tool], + verbose=True, + response_format=ResearchResult, # Optional: Use a structured output format + ) + + # # Example 1: Simple query with raw text response + # print("\n=== Example 1: Simple Query ===") + # result = await agent.kickoff_async("What is the population of Tokyo in 2023?") + # print(f"Raw response: {result.raw}") + + # Example 2: Query with structured output + print("\n=== Example 2: Structured Output ===") + structured_query = """ + # Research the impact of climate change on coral reefs. + + # YOU MUST format your response as a valid JSON object with the following structure: + # { + # "main_findings": "A summary of the main findings", + # "key_points": ["Point 1", "Point 2", "Point 3"], + # "sources": ["Source 1", "Source 2"] + # } + + # Include at least 3 key points and 2 sources. Wrap your JSON in ```json and ``` tags. + # """ + + result = await agent.kickoff_async(structured_query) + + if result.pydantic: + # Cast to the specific type for better IDE support + research_result = cast(ResearchResult, result.pydantic) + print(f"Main findings: {research_result.main_findings}") + print("\nKey points:") + for i, point in enumerate(research_result.key_points, 1): + print(f"{i}. {point}") + print("\nSources:") + for i, source in enumerate(research_result.sources, 1): + print(f"{i}. {source}") + else: + print(f"Raw response: {result.raw}") + print( + "\nNote: Structured output was not generated. The LLM may need more explicit instructions to format the response as JSON." + ) + print("Usage metrics:") + print(result.usage_metrics) + + # # Example 3: Multi-turn conversation + # print("\n=== Example 3: Multi-turn Conversation ===") + # messages = [ + # {"role": "user", "content": "I'm planning a trip to Japan."}, + # { + # "role": "assistant", + # "content": "That sounds exciting! Japan is a beautiful country with rich culture, delicious food, and stunning landscapes. What would you like to know about Japan to help with your trip planning?", + # }, + # { + # "role": "user", + # "content": "What are the best times to visit Tokyo and Kyoto?", + # }, + # ] + + # result = await agent.kickoff_async(messages) + # print(f"Response: {result.raw}") + + # # Print usage metrics if available + # if result.usage_metrics: + # print("\nUsage metrics:") + # for key, value in result.usage_metrics.items(): + # print(f"{key}: {value}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/lite_agent_example_2.py b/examples/lite_agent_example_2.py new file mode 100644 index 000000000..f01589007 --- /dev/null +++ b/examples/lite_agent_example_2.py @@ -0,0 +1,46 @@ +from typing import List, cast + +from crewai_tools.tools.website_search.website_search_tool import WebsiteSearchTool +from pydantic import BaseModel, Field + +from crewai.lite_agent import LiteAgent + + +# Define a structured output format +class MovieReview(BaseModel): + title: str = Field(description="The title of the movie") + rating: float = Field(description="Rating out of 10") + pros: List[str] = Field(description="List of positive aspects") + cons: List[str] = Field(description="List of negative aspects") + + +# Create a LiteAgent +critic = LiteAgent( + role="Movie Critic", + goal="Provide insightful movie reviews", + backstory="You are an experienced film critic known for balanced, thoughtful reviews.", + tools=[WebsiteSearchTool()], + verbose=True, + response_format=MovieReview, +) + +# Use the agent +query = """ +Review the movie 'Inception'. Include: +1. Your rating out of 10 +2. Key positive aspects +3. Areas that could be improved +""" + +result = critic.kickoff(query) + +# Access the structured output +review = cast(MovieReview, result.pydantic) +print(f"\nMovie Review: {review.title}") +print(f"Rating: {review.rating}/10") +print("\nPros:") +for pro in review.pros: + print(f"- {pro}") +print("\nCons:") +for con in review.cons: + print(f"- {con}") diff --git a/src/crewai/tools/tool_usage.py b/src/crewai/tools/tool_usage.py index eb31a4cc4..6ab5139cc 100644 --- a/src/crewai/tools/tool_usage.py +++ b/src/crewai/tools/tool_usage.py @@ -170,7 +170,7 @@ class ToolUsage: started_at = time.time() from_cache = False - result = None + result: str | None = None # check if cache is available if self.tools_handler and self.tools_handler.cache: result = self.tools_handler.cache.read( @@ -287,7 +287,7 @@ class ToolUsage: if self.agent and hasattr(self.agent, "tools_results"): self.agent.tools_results.append(data) - return result # type: ignore # No return value expected + return result def _format_result(self, result: Any) -> str: if self.task: