Skip to content

datarobot_genai.crewai.ragas_events

ragas_events

CrewAIRagasEventListener

Collects CrewAI events into Ragas messages for pipeline interactions.

Source code in datarobot_genai/crewai/ragas_events.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class CrewAIRagasEventListener:
    """Collects CrewAI events into Ragas messages for pipeline interactions."""

    def __init__(self) -> None:
        self.messages: list[HumanMessage | AIMessage | ToolMessage] = []

    def setup_listeners(self, crewai_event_bus: CrewAIEventsBus) -> None:
        # Lazy import to reduce memory overhead when ragas is not used
        from ragas.messages import AIMessage
        from ragas.messages import HumanMessage
        from ragas.messages import ToolCall
        from ragas.messages import ToolMessage

        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_crew_execution_started(_: Any, event: Any) -> None:
            self.messages.append(
                HumanMessage(content=f"Working on input '{json.dumps(event.inputs)}'")
            )

        @crewai_event_bus.on(AgentExecutionStartedEvent)
        def on_agent_execution_started(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=event.task_prompt, tool_calls=[]))

        @crewai_event_bus.on(AgentExecutionCompletedEvent)
        def on_agent_execution_completed(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=event.output, tool_calls=[]))

        @crewai_event_bus.on(AgentReasoningStartedEvent)
        def on_agent_reasoning_started(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=event.type, tool_calls=[]))

        @crewai_event_bus.on(AgentReasoningCompletedEvent)
        def on_agent_reasoning_completed(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=event.plan, tool_calls=[]))

        @crewai_event_bus.on(AgentReasoningFailedEvent)
        def on_agent_reasoning_failed(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=event.task_id, tool_calls=[]))

        @crewai_event_bus.on(TaskCompletedEvent)
        def on_agent_task_completed(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=f"Task output: '{event.output}'", tool_calls=[]))

        @crewai_event_bus.on(TaskFailedEvent)
        def on_agent_task_failed(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=event.error, tool_calls=[]))

        @crewai_event_bus.on(TaskStartedEvent)
        def on_agent_task_started(_: Any, event: Any) -> None:
            self.messages.append(AIMessage(content=event.context, tool_calls=[]))

        @crewai_event_bus.on(ToolUsageStartedEvent)
        def on_tool_usage_started(_: Any, event: Any) -> None:
            # It's a tool call - add tool call to last AIMessage
            if len(self.messages) == 0:
                logging.warning("Direct tool usage without agent invocation")
                return
            last_message = self.messages[-1]
            if not isinstance(last_message, AIMessage):
                logging.warning(
                    "Tool call must be preceded by an AIMessage somewhere in the conversation."
                )
                return
            if isinstance(event.tool_args, (str, bytes, bytearray)):
                parsed_args: Any = json.loads(event.tool_args)
            else:
                parsed_args = event.tool_args
            tool_call = ToolCall(name=event.tool_name, args=parsed_args)
            if last_message.tool_calls is None:
                last_message.tool_calls = []
            last_message.tool_calls.append(tool_call)

        @crewai_event_bus.on(ToolUsageErrorEvent)
        def on_tool_usage_error(_: Any, event: Any) -> None:
            self.messages.append(ToolMessage(content=event.error))

        @crewai_event_bus.on(ToolUsageFinishedEvent)
        def on_tool_usage_finished(_: Any, event: Any) -> None:
            if len(self.messages) == 0:
                logging.warning("Direct tool usage without agent invocation")
                return
            last_message = self.messages[-1]
            if not isinstance(last_message, AIMessage):
                logging.warning(
                    "Tool call must be preceded by an AIMessage somewhere in the conversation."
                )
                return
            if not last_message.tool_calls:
                logging.warning("No previous tool calls found")
                return
            self.messages.append(ToolMessage(content=event.output))