CrewAI new features

Agentic AI framework

LLM support

If your LLM is not in supported providers in the above page, you can check LiteLLM support as CrewAI uses LiteLLM to connect to LLMs, e.g. to connect LLM on Databricks model serving, refer to this page.

from litellm import completion
import os
## set ENV variables
os.environ["DATABRICKS_API_KEY"] = "databricks key"
os.environ["DATABRICKS_API_BASE"] = "databricks base url" # e.g.: https://adb-3064715882934586.6.azuredatabricks.net/serving-endpoints

# Databricks dbrx-instruct call
response = completion(
    model="databricks/databricks-dbrx-instruct", 
    messages = [{ "content": "Hello, how are you?","role": "user"}]
)

Below is how you can identify 3 key elements:

model name: just set model=databricks/<any-model-on-databricks> as a prefix when sending litellm requests. Supported foundation models on Mosaic AI Model Serving

Databricks foundational model API

DATABRICKS_API_BASE: use databricks sdk

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
openai_client = w.serving_endpoints.get_open_ai_client()

DATABRICKS_API_BASE = str(openai_client.base_url)

DATABRICKS_API_KEY: it is access key, follow this page.

Flow

Flows - CrewAI

Adding workflow management capability to CrewAI, mainly (multiple crews (each crew with multiple agents)) can be orchestrated with flows which can have control logic, state management (which is similar to graph in Langgraph)

crewAI-examples/self_evaluation_loop_flow at main · crewAIInc/crewAI-examples

Implement a workflow with evaluate-iterate-until-valid.

A flow can orchestrate between different functions which calls crew.kickoff and stores state. A flow has different events indicate different states, e.g.

from crewai.flow.flow import Flow, listen, router, start

start: Marks a method as a flow’s starting point, designating a method as an entry point for the flow execution, optionally specifying conditions that trigger the start based on other method executions.

@start("retry")
def generate_shakespeare_x_post(self):

listen: a listener that executes when specified conditions are met, e.g. @listen(“process_data”) # Listen to single method, @listen(or_(“success”, “failure”)) # Listen to multiple methods

@listen("complete")
def save_result(self):

router: a routing method that directs flow execution based on conditions

@router(generate_shakespeare_x_post)
def evaluate_x_post(self):

Generally, router will check result from generate_shakespeare_x_post method, then output conditions of “complete” and “retry” based on evaluation. It also tracks retry_count state, adding 1 with each iteration and if max retry is reached, outputs “max_retry_exceeded”.

    @router(generate_shakespeare_x_post)
    def evaluate_x_post(self):
        if self.state.retry_count > 3:
            return "max_retry_exceeded"

        result = XPostReviewCrew().crew().kickoff(inputs={"x_post": self.state.x_post})
        self.state.valid = result["valid"]
        self.state.feedback = result["feedback"]

        print("valid", self.state.valid)
        print("feedback", self.state.feedback)
        self.state.retry_count += 1

        if self.state.valid:
            return "complete"

        return "retry"

LLM type in crew

Manager LLM

Manager Agent

Hierarchical Process: A manager agent coordinates the crew, delegating tasks and validating outcomes before proceeding. Note: A manager_llm or manager_agent is required for this process and it’s essential for validating the process flow.

Planning LLM: The language model used by the AgentPlanner in a planning process.

Function calling LLM

If passed, the crew will use this LLM to do function calling for tools for all agents in the crew. Each agent can have its own LLM, which overrides the crew’s LLM for function calling.

Multi Agents and Function Calling with watsonx.ai

Knowledge

Knowledge - CrewAI

Basically, it will automatically converts source data to chunks, compute embeddings, and save them for later query.

Integration

If we can easily combine best of different LLM tooling features, we have more tools and capabilities. Other LLM agent frameworks are there: e.g. langchain/langgraph, llamaindex, autogen. Can Crewai leverage them?

Langchain

Using LangChain Tools - CrewAI

Approach: in CrewAI tool definition, create langchain tool and call tool.run in CrewAI tool ._run method.

from crewai.tools import BaseTool
from pydantic import Field
from langchain_community.utilities import GoogleSerperAPIWrapper

# Set up your SERPER_API_KEY key in an .env file, eg:
# SERPER_API_KEY=<your api key>
load_dotenv()

search = GoogleSerperAPIWrapper()

class SearchTool(BaseTool):
    name: str = "Search"
    description: str = "Useful for search-based queries. Use this to find current information about markets, companies, and trends."
    search: GoogleSerperAPIWrapper = Field(default_factory=GoogleSerperAPIWrapper)

    def _run(self, query: str) -> str:
        """Execute the search query and return results"""
        try:
            return self.search.run(query)
        except Exception as e:
            return f"Error performing search: {str(e)}"

Langraph

crewAI-examples/CrewAI-LangGraph at main · crewAIInc/crewAI-examples

Langraph defines the main graph, each node is a function (which can call CrewAI crew.kickoff)

LlamaIndex

Using LlamaIndex Tools - CrewAI

More straightforward approach to use LlamaIndex tools with LlamaIndexTool.from_tool

from crewai_tools import LlamaIndexTool

# Example 1: Initialize from FunctionTool
from llama_index.core.tools import FunctionTool

og_tool = FunctionTool.from_defaults(
    your_python_function, 
    name="<name>", 
    description='<description>'
)
tool = LlamaIndexTool.from_tool(og_tool)

Include LlamaIndex RAG as tool, with LlamaIndexTool.from_query_engine

llm = OpenAI(model="gpt-4o")
index = VectorStoreIndex.from_documents(docs)
query_engine = index.as_query_engine(similarity_top_k=5, llm=llm)
# try out query engine tool

query_tool = LlamaIndexTool.from_query_engine(
    query_engine,
    name="Uber 2019 10K Query Tool",
    description="Use this tool to lookup the 2019 Uber 10K Annual Report",
)

Schema Validation errors while using delegation

Tool input behavior is different across different LLMs and can cause CrewAI not functioning. As described in above article

GPT-4o

## Tool Input:
"{\"user\": \"USER_a1s2d3f4\"}"

Other LLMs

# Tool Input:
"{\"user\": {\"description\": \"User ID to fetch chat history for.\", \"type\": \"str\"}}"

Which caused pydantic input validation error

error Input should be a valid string [type=string_type, input_value={'description': '...t

Therefore, when using tool, we need to explicitly instruct agent to generate tool-argument-compatible parameters when passing to tool, which can be inconvenient when using multiple tools.

Appendix

https://platform.openai.com/docs/models