feat(langfuse): Parallel traces in langfuse + obvious step instructions (#1908)

1. Langfuse SDK version bumped
2. Parallel threads will now log under the same langfuse trace
2. Adds the instruction to the root cause obvious call

![CleanShot 2025-02-11 at 16 09
17](https://github.com/user-attachments/assets/35abda72-2bec-46f0-9294-f7eb31c21ea6)
![CleanShot 2025-02-11 at 16 08
44](https://github.com/user-attachments/assets/ab6e4739-ebf9-4f3c-b8a4-ccbf0bf44946)

Author: jennmueng

pyproject.toml

@@ -17,7 +17,7 @@
     mypy_path = []
     plugins = ["pydantic.mypy"]
     files = ["src/"]
-    disable_error_code = ["no-untyped-def"]
+    disable_error_code = ["no-untyped-def", "unused-ignore"]
     # minimal strictness settings
     no_implicit_reexport = true
     warn_unreachable = true

requirements-constraints.txt

@@ -100,7 +100,7 @@ google-cloud-storage==2.*
 google-cloud-aiplatform==1.*
 google-cloud-secret-manager==2.*
 anthropic[vertex]==0.45.*
-langfuse @ git+https://github.com/jennmueng/langfuse-python.git@9d9350de1e4e84fa548fe84f82c1b826be17956e
+langfuse @ git+https://github.com/jennmueng/langfuse-python.git@d7c0127682ddb20f73c5cf4fbb396cdfa8961fc3
 watchdog
 stumpy==1.13.0
 pytest_alembic==0.11.1

requirements.txt

@@ -110,7 +110,7 @@ convertdate==2.4.0
     # via -r requirements-constraints.txt
 covdefaults==2.3.0
     # via -r requirements-constraints.txt
-coverage==7.6.11
+coverage==7.6.12
     # via
     #   covdefaults
     #   pytest-cov
@@ -165,7 +165,7 @@ flask-sqlalchemy==3.1.1
     #   -r requirements-constraints.txt
     #   flask-migrate
     #   types-flask-migrate
-flatbuffers==25.1.24
+flatbuffers==25.2.10
     # via onnxruntime
 fonttools==4.43.0
     # via
@@ -200,7 +200,7 @@ google-auth==2.38.0
     #   google-cloud-secret-manager
     #   google-cloud-storage
     #   google-genai
-google-cloud-aiplatform==1.79.0
+google-cloud-aiplatform==1.80.0
     # via -r requirements-constraints.txt
 google-cloud-bigquery==3.29.0
     # via google-cloud-aiplatform
@@ -339,7 +339,7 @@ kiwisolver==1.4.5
     #   matplotlib
 kombu==5.4.2
     # via celery
-langfuse @ git+https://github.com/jennmueng/langfuse-python.git@9d9350de1e4e84fa548fe84f82c1b826be17956e
+langfuse @ git+https://github.com/jennmueng/langfuse-python.git@d7c0127682ddb20f73c5cf4fbb396cdfa8961fc3
     # via -r requirements-constraints.txt
 lazy-object-proxy==1.10.0
     # via openapi-spec-validator
@@ -440,7 +440,7 @@ optimum==1.16.2
     # via -r requirements-constraints.txt
 overrides==7.7.0
     # via chromadb
-packaging==23.2
+packaging==24.2
     # via
     #   -r requirements-constraints.txt
     #   build
@@ -487,7 +487,7 @@ pip-tools==7.4.1
     # via -r requirements-constraints.txt
 pluggy==1.5.0
     # via pytest
-posthog==3.11.0
+posthog==3.12.1
     # via chromadb
 prompt-toolkit==3.0.50
     # via click-repl
@@ -644,6 +644,7 @@ requests==2.32.2
     #   huggingface-hub
     #   jsonschema-path
     #   jsonschema-spec
+    #   langfuse
     #   posthog
     #   pygithub
     #   transformers

src/seer/automation/autofix/autofix_agent.py

@@ -3,6 +3,9 @@
 from concurrent.futures import Executor, Future, ThreadPoolExecutor
 from typing import Callable, Optional
 
+from langfuse.decorators import langfuse_context, observe
+from sentry_sdk.ai.monitoring import ai_track
+
 from seer.automation.agent.agent import AgentConfig, LlmAgent, RunConfig
 from seer.automation.agent.models import (
     LlmGenerateTextResponse,
@@ -167,6 +170,9 @@ def run_iteration(self, run_config: RunConfig):
             and not cur.request.options.disable_interactivity
             and completion.message.tool_calls  # only if the run is in progress
         ):
+            trace_id = langfuse_context.get_current_trace_id()
+            observation_id = langfuse_context.get_current_observation_id()
+
             text_before_tag = completion.message.content.split("<")[0]
             text = text_before_tag
             if text:
@@ -178,6 +184,8 @@ def run_iteration(self, run_config: RunConfig):
                         cur_step_idx,
                         self.context.state,
                         len(self.memory) - 1,
+                        langfuse_parent_trace_id=trace_id,  # type: ignore
+                        langfuse_parent_observation_id=observation_id,  # type: ignore
                     )
                 )
 
@@ -226,6 +234,8 @@ def use_user_messages(self):
             self.queued_user_messages = []
             self.context.event_manager.add_log("Thanks for the input. Thinking through it now...")
 
+    @observe(name="Share Insights in parallel")
+    @ai_track(description="Share Insights in parallel")
     def share_insights(
         self,
         text: str,

src/seer/automation/autofix/components/coding/component.py

@@ -4,7 +4,7 @@
 import textwrap
 
 import sentry_sdk
-from langfuse.decorators import observe
+from langfuse.decorators import langfuse_context, observe
 from openai import BadRequestError as OpenAiBadRequestError
 from openai import LengthFinishReasonError as OpenAiLengthFinishReasonError
 from pydantic import BaseModel
@@ -290,6 +290,8 @@ def invoke(self, request: CodingRequest) -> CodingOutput | None:
         # Resolve LlmClient once in the main thread
         resolved_llm_client = self._get_llm_client()
 
+        @observe(name="Process Change Task")
+        @ai_track(description="Process Change Task")
         def process_task(task, llm_client):
             repo_client = self.context.get_repo_client(task.repo_name)
             if task.type == "file_change":
@@ -349,8 +351,17 @@ def process_task(task, llm_client):
 
         # apply change tasks in parallel
         with concurrent.futures.ThreadPoolExecutor() as executor:
+            trace_id = langfuse_context.get_current_trace_id()
+            observation_id = langfuse_context.get_current_observation_id()
+
             futures = [
-                executor.submit(process_task, task, resolved_llm_client)
+                executor.submit(
+                    process_task,
+                    task,
+                    resolved_llm_client,
+                    langfuse_parent_trace_id=trace_id,  # type: ignore
+                    langfuse_parent_observation_id=observation_id,  # type: ignore
+                )
                 for task in code_changes_output.tasks
             ]
             for future in concurrent.futures.as_completed(futures):

src/seer/automation/autofix/components/is_root_cause_obvious.py

@@ -5,13 +5,15 @@
 
 from seer.automation.agent.client import GeminiProvider, LlmClient
 from seer.automation.autofix.autofix_context import AutofixContext
+from seer.automation.autofix.prompts import format_instruction
 from seer.automation.component import BaseComponent, BaseComponentOutput, BaseComponentRequest
 from seer.automation.models import EventDetails
 from seer.dependency_injection import inject, injected
 
 
 class IsRootCauseObviousRequest(BaseComponentRequest):
     event_details: EventDetails
+    instruction: str | None = None
 
 
 class IsRootCauseObviousOutput(BaseComponentOutput):
@@ -22,16 +24,21 @@ class IsRootCauseObviousPrompts:
     @staticmethod
     def format_default_msg(
         event_details: EventDetails,
+        instruction: str | None = None,
     ):
         return (
             textwrap.dedent(
                 """\
             You are an exceptional principal engineer that is amazing at finding the root cause of any issue. We have an issue in our codebase described below. Is the true, deepest root cause of the issue clear from the details below? Or does it require searching for more information around the codebase?
 
-            {event_details}"""
+            <issue_details>
+            {event_details}
+            </issue_details>
+            {instruction_str}"""
             )
             .format(
                 event_details=event_details.format_event(),
+                instruction_str=format_instruction(instruction),
             )
             .strip()
         )
@@ -51,6 +58,7 @@ def invoke(
         output = llm_client.generate_structured(
             prompt=IsRootCauseObviousPrompts.format_default_msg(
                 event_details=request.event_details,
+                instruction=request.instruction,
             ),
             model=GeminiProvider.model("gemini-2.0-flash-001"),
             response_format=IsRootCauseObviousOutput,

src/seer/automation/autofix/components/root_cause/component.py

@@ -35,7 +35,10 @@ def invoke(
     ) -> RootCauseAnalysisOutput:
         is_obvious = (
             IsRootCauseObviousComponent(self.context).invoke(
-                IsRootCauseObviousRequest(event_details=request.event_details)
+                IsRootCauseObviousRequest(
+                    event_details=request.event_details,
+                    instruction=request.instruction,
+                )
             )
             if not request.initial_memory
             else None