解锁crewai的潜力：5个新功能提升您的AI工作流程

Agentic AI 框架

如果您的 LLM 不在上述页面中的 supported providers 中，您可以检查 LiteLLM 支持情况，因为 CrewAI 使用 LiteLLM 连接到 LLM，例如，要连接 Databricks 模型服务上的 LLM，请参阅 此页面。

from litellm import completion
import os

os.environ["DATABRICKS_API_KEY"] = "databricks key"
os.environ["DATABRICKS_API_BASE"] = "databricks base url"

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

以下是如何确定 3 个关键元素：

model name：只需在发送 litellm 请求时将 model=databricks/ 设置为前缀。Supported foundation models on Mosaic AI Model Serving

Databricks foundational model API

DATABRICKS_API_BASE：使用 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：它是访问密钥，请遵循 此页面。

流程

Flows - CrewAI

了解如何使用 CrewAI Flows 创建和管理 AI 工作流程。

docs.crewai.com

为 CrewAI 添加工作流程管理功能，主要是（多个 crew（每个 crew 有多个 agent））可以使用 flow 进行编排，这些 flow 可以具有控制逻辑、状态管理（类似于 Langgraph 中的 graph）

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

展示如何使用 CrewAI 框架自动化工作流程的示例集合。…

github.com

实现一个带有 evaluate-iterate-until-valid 的工作流程。

一个 flow 可以在调用 crew.kickoff 并存储状态的不同函数之间进行编排。一个 flow 具有指示不同状态的不同事件，例如

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

start：将方法标记为 flow 的起始点，将方法指定为 flow 执行的入口点，可选择指定基于其他方法执行触发启动的条件。

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

listen：当满足指定条件时执行的监听器，例如 @listen(“process_data”) # 监听单个方法，@listen(or_(“success”, “failure”)) # 监听多个方法

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

router：一个路由方法，根据条件指导 flow 执行

@router(generate_shakespeare_x_post)
def evaluate_x_post(self):

通常，router 将检查来自 generate_shakespeare_x_post 方法的结果，然后根据评估输出“complete”和“retry”的条件。它还跟踪 retry_count 状态，每次迭代加 1，如果达到最大重试次数，则输出“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"

Crew 中的 LLM 类型

Manager LLM

Manager Agent

分层流程：一个 manager agent 协调 crew，委派任务并在继续之前验证结果。注意：此流程需要 manager_llm 或 manager_agent，并且对于验证流程至关重要。

Planning LLM：AgentPlanner 在规划流程中使用的语言模型。

Function calling LLM

如果传递，crew 将使用此 LLM 为 crew 中的所有 agent 调用工具的函数。每个 agent 都可以有自己的 LLM，它会覆盖 crew 的 LLM 用于函数调用。

Multi Agents and Function Calling with watsonx.ai

LLM Agents 和 Function Calling 是构建现代 AI 应用程序的强大技术。这篇文章描述了一个示例…

heidloff.net

知识

Knowledge - CrewAI

CrewAI 中的知识是什么以及如何使用它。

docs.crewai.com

基本上，它将自动将源数据转换为块，计算嵌入，并保存它们以供以后查询。

使用 Azure OpenAI

CrewAI 知识库使用嵌入作为后端。当您将默认的 OpenAI 迁移到 Azure OpenAI 时，您需要同时提供 LLM 和嵌入配置。

.env

OPENAI_API_TYPE=azure
AZURE_OPENAI_ENDPOINT=https://<endpoint>.openai.azure.com/
AZURE_OPENAI_API_VERSION=2024-08-01-preview
AZURE_OPENAI_API_KEY=
AZURE_OPENAI_GPT4O_MODEL_NAME=gpt-4o-mini
AZURE_OPENAI_GPT4O_DEPLOYMENT_NAME=gpt-4o-mini
AZURE_OPENAI_EMBEDDING_API_VERSION=2023-05-15
AZURE_OPENAI_EMBEDDING_MODEL_NAME=
AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME=

设置 Azure OpenAI LLM 和嵌入配置

import os
from crewai import LLM

from dotenv import load_dotenv
load_dotenv(".env")

os.environ["AZURE_API_KEY"] = os.getenv("AZURE_OPENAI_API_KEY") # "my-azure-api-key"
os.environ["AZURE_API_BASE"] = os.getenv("AZURE_OPENAI_ENDPOINT") # "https://example-endpoint.openai.azure.com"
os.environ["AZURE_API_VERSION"] = os.getenv("AZURE_OPENAI_API_VERSION")

crewai_llm = LLM(model=f"azure/{model_name}", temperature=0.0)
crewai_embedder_config=dict(
## provider="azure_openai", # or openai, ollama, ...
provider="azure",
config=dict(
model=os.getenv("AZURE_OPENAI_EMBEDDING_MODEL_NAME"),
deployment_name=os.getenv("AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME"),
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
api_base=os.getenv("AZURE_OPENAI_ENDPOINT"),
api_version=os.getenv("AZURE_OPENAI_EMBEDDING_API_VERSION"),
openai_api_version=os.getenv("AZURE_OPENAI_EMBEDDING_API_VERSION"),
openai_api_type="azure"
)
)

修改 CrewAI 知识库示例以使用 Azure OpenAI（注意 llm 和 embedder）

from crewai import Agent, Task, Crew, Process, LLM
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource

content = "Users name is John. He is 30 years old and lives in San Francisco."
string_source = StringKnowledgeSource(
    content=content,
)

agent = Agent(
    role="About User",
    goal="You know everything about the user.",
    backstory="""You are a master at understanding people and their preferences.""",
    verbose=True,
    allow_delegation=False,
    llm=crewai_llm,
    embedder=crewai_embedder_config
)
task = Task(
    description="Answer the following questions about the user: {question}",
    expected_output="An answer to the question.",
    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    verbose=True,
    process=Process.sequential,
    knowledge_sources=[string_source], 
    embedder=crewai_embedder_config,
)

result = crew.kickoff(inputs={"question": "What city does John live in and how old is he?"})

结果

## Agent: About User
#

## Agent: About User
## 
John lives in San Francisco and he is 30 years old.

集成

如果我们能够轻松地结合不同 LLM 工具的特性，我们将拥有更多工具和能力。还有其他 LLM 代理框架：例如 langchain/langgraph、llamaindex、autogen。CrewAI 可以利用它们吗？

Langchain

使用 LangChain 工具 - CrewAI

了解如何将 LangChain 工具与 CrewAI 代理集成，以增强基于搜索的查询等。

docs.crewai.com

方法：在 CrewAI 工具定义 中，创建 langchain 工具并在 CrewAI 工具的 ._run 方法中调用 tool.run。

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

展示如何使用 CrewAI 框架自动化工作流程的示例集合。…

github.com

Langraph 定义了主图，每个节点都是一个函数（可以调用 CrewAI crew.kickoff）

LlamaIndex

使用 LlamaIndex 工具 - CrewAI

了解如何将 LlamaIndex 工具与 CrewAI 代理集成，以增强基于搜索的查询等。

docs.crewai.com

使用 LlamaIndexTool.from_tool 更直接地使用 LlamaIndex 工具

from crewai_tools import LlamaIndexTool

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)

将 LlamaIndex RAG 作为工具包含，使用 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)

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",
)

使用委托时出现模式验证错误

我按照文档使用 manager 和几个简单的 agent 进行了标准设置，但它却因严格的错误而失败…

community.crewai.com

不同的 LLM 之间工具输入行为不同，这可能导致 CrewAI 无法正常工作。如上文所述

GPT-4o

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

其他 LLM

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

这导致了 pydantic 输入验证错误

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

因此，在使用工具时，我们需要明确指示 agent 在传递给工具时生成与工具参数兼容的参数，这在使用多个工具时可能会带来不便。

附录

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