mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-21 05:48:14 +00:00
58fb717ab2
- Add ExecutionStep and ExecutionTrace models to track crew execution steps - Add ExecutionTraceCollector to capture events and build execution traces - Add trace_execution parameter to Crew class (disabled by default) - Add execution_trace field to CrewOutput to return trace data - Integrate trace collection into crew.kickoff() method - Add comprehensive tests covering execution tracing functionality - Add example demonstrating how to use execution tracing - Export new classes in __init__.py Addresses issue #3268: Users can now track the sequence of steps/actions that a crew takes to complete a goal, including agent thoughts, tool calls, and intermediate results, similar to LangGraph's conversation state. Co-Authored-By: João <joao@crewai.com>
62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
import json
|
|
from typing import Any, Dict, Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from crewai.tasks.output_format import OutputFormat
|
|
from crewai.tasks.task_output import TaskOutput
|
|
from crewai.types.usage_metrics import UsageMetrics
|
|
from crewai.crews.execution_trace import ExecutionTrace
|
|
|
|
|
|
class CrewOutput(BaseModel):
|
|
"""Class that represents the result of a crew."""
|
|
|
|
raw: str = Field(description="Raw output of crew", default="")
|
|
pydantic: Optional[BaseModel] = Field(
|
|
description="Pydantic output of Crew", default=None
|
|
)
|
|
json_dict: Optional[Dict[str, Any]] = Field(
|
|
description="JSON dict output of Crew", default=None
|
|
)
|
|
tasks_output: list[TaskOutput] = Field(
|
|
description="Output of each task", default=[]
|
|
)
|
|
token_usage: UsageMetrics = Field(description="Processed token summary", default={})
|
|
execution_trace: Optional[ExecutionTrace] = Field(
|
|
description="Detailed execution trace of crew steps", default=None
|
|
)
|
|
|
|
@property
|
|
def json(self) -> Optional[str]:
|
|
if self.tasks_output[-1].output_format != OutputFormat.JSON:
|
|
raise ValueError(
|
|
"No JSON output found in the final task. Please make sure to set the output_json property in the final task in your crew."
|
|
)
|
|
|
|
return json.dumps(self.json_dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert json_output and pydantic_output to a dictionary."""
|
|
output_dict = {}
|
|
if self.json_dict:
|
|
output_dict.update(self.json_dict)
|
|
elif self.pydantic:
|
|
output_dict.update(self.pydantic.model_dump())
|
|
return output_dict
|
|
|
|
def __getitem__(self, key):
|
|
if self.pydantic and hasattr(self.pydantic, key):
|
|
return getattr(self.pydantic, key)
|
|
elif self.json_dict and key in self.json_dict:
|
|
return self.json_dict[key]
|
|
else:
|
|
raise KeyError(f"Key '{key}' not found in CrewOutput.")
|
|
|
|
def __str__(self):
|
|
if self.pydantic:
|
|
return str(self.pydantic)
|
|
if self.json_dict:
|
|
return str(self.json_dict)
|
|
return self.raw
|