Log intermediate actions within a task.

marketing_strategy/src/marketing_posts/crew.py

@@ -38,85 +38,100 @@ class Copy(BaseModel):
 	title: str = Field(..., description="Title of the copy")
 	body: str = Field(..., description="Body of the copy")
 
+
+def make_step_callback(span):
+	"""Logs intermediate actions within a task as events in the trace."""
+	def step_callback(step_output):
+		if isinstance(step_output, list):
+			for action, output in step_output:
+				span.event(name="action", input=action, output=output)
+		else:
+			# step_output is an `AgentFinish` instance.
+			span.event(name=f"action_finish", output=step_output.return_values)
+	return step_callback
+
+
 @CrewBase
 class MarketingPostsCrew():
 	"""MarketingPosts crew"""
 	agents_config = 'config/agents.yaml'
 	tasks_config = 'config/tasks.yaml'
 
 	def __init__(self):
-		self.trace = langfuse.trace(
-			id="marketing-posts-crew",
-			name="Marketing Posts Generation"
-		)
-		self.handler = self.trace.get_langchain_handler()
+		self.trace = langfuse.trace(name="Marketing Posts Generation")
 
 	@agent
 	def lead_market_analyst(self) -> Agent:
+		span = self.trace.span(name=f"lead_market_analyst")
 		return Agent(
 			config=self.agents_config['lead_market_analyst'],
 			tools=[SerperDevTool(), ScrapeWebsiteTool()],
 			verbose=True,
 			memory=False,
-			callbacks=[self.handler],
+			step_callback=make_step_callback(span),
+			callbacks=[span.get_langchain_handler()],
 		)
 
 	@agent
 	def chief_marketing_strategist(self) -> Agent:
+		span = self.trace.span(name=f"chief_marketing_strategist")
 		return Agent(
 			config=self.agents_config['chief_marketing_strategist'],
 			tools=[SerperDevTool(), ScrapeWebsiteTool()],
-				verbose=True,
-				memory=False,
-				callbacks=[self.handler],
+			verbose=True,
+			memory=False,
+			step_callback=make_step_callback(span),
+			callbacks=[span.get_langchain_handler()],
 		)
 
 	@agent
 	def creative_content_creator(self) -> Agent:
+		span = self.trace.span(name=f"creative_content_creator")
 		return Agent(
 			config=self.agents_config['creative_content_creator'],
 			verbose=True,
 			memory=False,
-			callbacks=[self.handler],
+			step_callback=make_step_callback(span),
+			callbacks=[span.get_langchain_handler()],
 		)
 
 	@task
 	def research_task(self) -> Task:
 		return Task(
 			config=self.tasks_config['research_task'],
-			agent=self.lead_market_analyst()
+			agent=self.lead_market_analyst(),
 		)
 
 	@task
 	def project_understanding_task(self) -> Task:
 		return Task(
 			config=self.tasks_config['project_understanding_task'],
-			agent=self.chief_marketing_strategist()
+			agent=self.chief_marketing_strategist(),
 		)
 
 	@task
 	def marketing_strategy_task(self) -> Task:
 		return Task(
 			config=self.tasks_config['marketing_strategy_task'],
 			agent=self.chief_marketing_strategist(),
-			output_json=MarketStrategy
+			output_json=MarketStrategy,
 		)
 
 	@task
 	def campaign_idea_task(self) -> Task:
 		return Task(
 			config=self.tasks_config['campaign_idea_task'],
 			agent=self.creative_content_creator(),
-   		output_json=CampaignIdea
+   			output_json=CampaignIdea,
 		)
 
 	@task
 	def copy_creation_task(self) -> Task:
 		return Task(
 			config=self.tasks_config['copy_creation_task'],
 			agent=self.creative_content_creator(),
-   		context=[self.marketing_strategy_task(), self.campaign_idea_task()],
-			output_json=Copy
+   			context=[self.marketing_strategy_task(), self.campaign_idea_task()],
+			output_json=Copy,
 		)
 
 	@crew