I would like to get detailed info from callback function of a crew

Crews
, , ,
1

I am going to build an app using CrewAI in python
I already built a crew and then add agents list, task list and step_callback function.

I would like to get detailed agent info(role, …) and detailed task info from callback function whenever event(ToolResult, AgentAction, TaskOutput) has happened

2

First off, welcome aboard.

Your interest seems to be related to event listeners. We now have official documentation on this topic, check it out here.

We’ve also discussed listening to events within our community forum; take a look at these threads here, here, and here.

3

Just want to know what info I can get from step_callback or task_callback function of a crew.

def step_callback(self, event, **kwargs):
event_type = event.class.name (“ToolResult”, “AgentAction”, …)

for example, we can get event_type from the above code(this code is working as well)

I would like to get all info from callback function, not only event type.

4

Well, the argument passed to the task_callback function is an object of the TaskOutput class. The attributes of this class are documented here.

As for the argument passed to the step_callback function, it will vary depending on the step that was completed. It can be an object of the ToolResult, AgentAction, or AgentFinish class. That’s why I suggested earlier that a well-defined event listener might be better suited for your use case.

In any case, to learn more about the argument your callback is receiving, here’s a snippet:

from crewai import LLM, Agent, Task, Crew, Process
from typing import Any

# [Define your Agents, Tasks, Tools, etc.]

def log_task_output(task_output: Any):
    print("\n--- Task Callback Received ---")
    if task_output is not None:
        print(f"Object type: {type(task_output).__name__}")
        attributes = getattr(
            task_output, "__dict__", "N/A (object may lack __dict__)"
        )
        print(f"Instance attributes: {attributes}")
    else:
        print("Received None object.")

def log_step_output(step_output: Any):
    print("\n--- Step Callback Received ---")
    if step_output is not None:
        print(f"Object type: {type(step_output).__name__}")
        attributes = getattr(
            step_output, "__dict__", "N/A (object may lack __dict__)"
        )
        print(f"Instance attributes: {attributes}")
    else:
        print("Received None object.")

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    task_callback=log_task_output,
    step_callback=log_step_output,
    verbose=False
)

my_crew.kickoff()
1 Like
5

This topic was automatically closed 24 hours after the last reply. New replies are no longer allowed.