Rastreamento da CrewAI
CrewAI é uma estrutura de código aberto para orquestrar o agente AI autônomo e de interpretação de papéis.
O MLflow Tracing fornece o recurso de rastreamento automático para o CrewAI, uma estrutura de código aberto para a criação de aplicativos multiagentes. Ao ativar o rastreamento automático
para CrewAI chamando a função mlflow.crewai.autolog, MLflow capturará os rastreamentos aninhados para a execução do fluxo de trabalho CrewAI e os registrará na experiência MLflow ativa.
import mlflow
mlflow.crewai.autolog()
MLflow O rastreamento captura automaticamente as seguintes informações sobre os agentes do CrewAI:
- tarefa e o agente que executa cada tarefa
- Cada chamada de LLM com prompts de entrada, respostas de conclusão e vários metadados
- Operações de carga e gravação de memória
- Latência de cada operação
- Qualquer exceção, se levantada
Atualmente, a integração do MLflow CrewAI suporta apenas o rastreamento para a execução de tarefas síncronas. A tarefa assíncrona e o kickoff não são suportados no momento.
Pré-requisitos
Para usar o MLflow Tracing com o CrewAI, o senhor precisa instalar o MLflow e a biblioteca crewai (que inclui crewai_tools).
- Development
- Production
Para ambientes de desenvolvimento, instale o pacote completo do MLflow com os extras do Databricks e crewai:
pip install --upgrade "mlflow[databricks]>=3.1" crewai
O pacote completo do mlflow[databricks] inclui todos os recursos para desenvolvimento local e experimentação no Databricks.
Para implantações de produção, instale mlflow-tracing e crewai:
pip install --upgrade mlflow-tracing crewai
O pacote mlflow-tracing é otimizado para uso na produção.
O MLflow 3 é altamente recomendado para obter a melhor experiência de rastreamento com o CrewAI.
Antes de executar os exemplos, você precisará configurar seu ambiente:
Para usuários fora do Databricks Notebook : Defina seu Databricks variável de ambiente:
export DATABRICKS_HOST="https://your-workspace.cloud.databricks.com"
export DATABRICKS_TOKEN="your-personal-access-token"
Para usuários do Databricks Notebook : Essas credenciais são definidas automaticamente para o senhor.
API chave : Certifique-se de que todas as chaves necessárias do provedor LLM API estejam configuradas. Para uso em produção, use segredos do AI Gateway ou do Databricks em vez de valores codificados:
export OPENAI_API_KEY="your-openai-api-key"
export SERPER_API_KEY="your-serper-api-key"
# Add other provider keys as needed
Exemplo de uso
Primeiro, ative o rastreamento automático para o CrewAI e, opcionalmente, crie um experimento MLflow para gravar os rastreamentos. Isso ajuda a organizar melhor seus traços.
import mlflow
import os
# Ensure your API keys (e.g., OPENAI_API_KEY, SERPER_API_KEY) are set in your environment
# os.environ["OPENAI_API_KEY"] = "your-openai-key"
# os.environ["SERPER_API_KEY"] = "your-serper-key"
# Turn on auto tracing by calling mlflow.crewai.autolog()
mlflow.crewai.autolog()
# Set up MLflow tracking to Databricks
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment("/Shared/crewai-demo")
Em seguida, defina um fluxo de trabalho multiagente usando o CrewAI. O exemplo a seguir define um agente de planejamento de viagens que usa o recurso de pesquisa na web como uma ferramenta.
from crewai import Agent, Crew, Task
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
from crewai_tools import SerperDevTool, WebsiteSearchTool
from textwrap import dedent
content = "Users name is John. He is 30 years old and lives in San Francisco."
string_source = StringKnowledgeSource(
content=content, metadata={"preference": "personal"}
)
search_tool = WebsiteSearchTool()
class TripAgents:
def city_selection_agent(self):
return Agent(
role="City Selection Expert",
goal="Select the best city based on weather, season, and prices",
backstory="An expert in analyzing travel data to pick ideal destinations",
tools=[
search_tool,
],
verbose=True,
)
def local_expert(self):
return Agent(
role="Local Expert at this city",
goal="Provide the BEST insights about the selected city",
backstory="""A knowledgeable local guide with extensive information
about the city, it's attractions and customs""",
tools=[search_tool],
verbose=True,
)
class TripTasks:
def identify_task(self, agent, origin, cities, interests, range):
return Task(
description=dedent(
f"""
Analyze and select the best city for the trip based
on specific criteria such as weather patterns, seasonal
events, and travel costs. This task involves comparing
multiple cities, considering factors like current weather
conditions, upcoming cultural or seasonal events, and
overall travel expenses.
Your final answer must be a detailed
report on the chosen city, and everything you found out
about it, including the actual flight costs, weather
forecast and attractions.
Traveling from: {origin}
City Options: {cities}
Trip Date: {range}
Traveler Interests: {interests}
"""
),
agent=agent,
expected_output="Detailed report on the chosen city including flight costs, weather forecast, and attractions",
)
def gather_task(self, agent, origin, interests, range):
return Task(
description=dedent(
f"""
As a local expert on this city you must compile an
in-depth guide for someone traveling there and wanting
to have THE BEST trip ever!
Gather information about key attractions, local customs,
special events, and daily activity recommendations.
Find the best spots to go to, the kind of place only a
local would know.
This guide should provide a thorough overview of what
the city has to offer, including hidden gems, cultural
hotspots, must-visit landmarks, weather forecasts, and
high level costs.
The final answer must be a comprehensive city guide,
rich in cultural insights and practical tips,
tailored to enhance the travel experience.
Trip Date: {range}
Traveling from: {origin}
Traveler Interests: {interests}
"""
),
agent=agent,
expected_output="Comprehensive city guide including hidden gems, cultural hotspots, and practical travel tips",
)
class TripCrew:
def __init__(self, origin, cities, date_range, interests):
self.cities = cities
self.origin = origin
self.interests = interests
self.date_range = date_range
def run(self):
agents = TripAgents()
tasks = TripTasks()
city_selector_agent = agents.city_selection_agent()
local_expert_agent = agents.local_expert()
identify_task = tasks.identify_task(
city_selector_agent,
self.origin,
self.cities,
self.interests,
self.date_range,
)
gather_task = tasks.gather_task(
local_expert_agent, self.origin, self.interests, self.date_range
)
crew = Crew(
agents=[city_selector_agent, local_expert_agent],
tasks=[identify_task, gather_task],
verbose=True,
memory=True,
knowledge={
"sources": [string_source],
"metadata": {"preference": "personal"},
},
)
result = crew.kickoff()
return result
trip_crew = TripCrew("California", "Tokyo", "Dec 12 - Dec 20", "sports")
result = trip_crew.run()
Para ambientes de produção, use AI Gateway ou Databricks secrets em vez de valores codificados ou variáveis de ambiente para um gerenciamento seguro API key .
Desativar o rastreamento automático
O rastreamento automático para CrewAI pode ser desativado globalmente chamando mlflow.crewai.autolog(disable=True) ou mlflow.autolog(disable=True).