How to Use Conditional Tasks to Check for Clients in Singapore?

CrewAI Community Support
, ,
1

Hello CrewAI community,

I’m working on a task flow where I need to determine if a company has business clients based in Singapore. Here’s the step-by-step breakdown of what I’m trying to achieve:

  1. Input the name of the company and fetch a list of its clients (e.g., 10 clients).
  2. For each client, search for their address and check if it’s in Singapore.
  3. As soon as one client is identified as being from Singapore, stop further checks and return TRUE.
  4. If no clients are found in Singapore after processing the list, return FALSE.

I believe this can be implemented using Conditional Tasks, but I’m struggling with how to dynamically handle tasks for each client and stop the flow upon meeting the condition.

2

Hello,

i’m new to crewai, from the documentation i think you can do this with Conditional Tasks - CrewAI

Check the following for inspiration

from typing import List, Optional
from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process
from crewai.tasks.conditional_task import ConditionalTask
from crewai.tasks.task_output import TaskOutput
from crewai_tools import SerperDevTool

# Define data models
class Client(BaseModel):
    name: str
    address: Optional[str] = None
    is_singapore: Optional[bool] = None

class ClientList(BaseModel):
    clients: List[Client]
    has_singapore_client: Optional[bool] = False

# Define agents
researcher = Agent(
    role="Client Researcher",
    goal="Find and validate company client information",
    backstory="""You are an expert at finding company client information 
        and validating business addresses.""",
    tools=[SerperDevTool()],
    verbose=True
)

# Create tasks
def fetch_clients_task(company_name: str) -> Task:
    return Task(
        description=f"""Find a list of up to 10 clients for {company_name}.
            Only include client company names that you are confident about.""",
        expected_output="A list of client company names",
        agent=researcher,
        output_pydantic=ClientList
    )

def check_client_location(client: Client) -> ConditionalTask:
    def is_not_singapore(output: TaskOutput) -> bool:
        # If client is not from Singapore, continue checking other clients
        return not output.pydantic.is_singapore

    return ConditionalTask(
        description=f"""Find the business address for {client.name} and determine 
            if it's located in Singapore. Be thorough in your verification.""",
        expected_output="Client location information and Singapore status",
        condition=is_not_singapore,
        agent=researcher,
        output_pydantic=Client
    )

def create_crew(company_name: str) -> Crew:
    # Initial task to fetch client list
    initial_task = fetch_clients_task(company_name)
    
    # Create crew with sequential process
    crew = Crew(
        agents=[researcher],
        tasks=[initial_task],
        process=Process.sequential,
        verbose=True
    )
    
    # Add conditional tasks after getting initial client list
    result = crew.kickoff()
    clients = result.pydantic.clients
    
    # Create conditional tasks for each client
    location_tasks = [check_client_location(client) for client in clients]
    
    # Update crew with all tasks
    crew.tasks.extend(location_tasks)
    
    return crew

def check_singapore_presence(company_name: str) -> bool:
    """
    Main function to check if a company has clients in Singapore.
    Returns True if at least one client is found in Singapore, False otherwise.
    """
    crew = create_crew(company_name)
    result = crew.kickoff()
    
    # Check results from all completed tasks
    for task in crew.tasks[1:]:  # Skip the initial fetch task
        if task.output and task.output.pydantic.is_singapore:
            return True
    
    return False

# Example usage
if __name__ == "__main__":
    company_name = "Example Corp"
    has_singapore_clients = check_singapore_presence(company_name)
    print(f"Does {company_name} have clients in Singapore? {has_singapore_clients}")

To use this code:

  1. First, ensure you have the required environment variables set up:
export OPENAI_API_KEY=secret_key
export SERPER_API_KEY=secret_key
  1. Then you can use it like this:
ptyhon main.py