Instrument the Application

Attach the Splunk Agent Observability Callback

3 minutes

The agent runs its LangGraph workflow asynchronously, so you’ll attach Splunk Agent Observability’s async callback handler. Because the callback is passed at the graph level, it propagates to every node automatically, with no per-tool instrumentation required.

Exercise Add the callback to the agent
1

Add the imports

We’ve already added the following imports to the ~/workshop/healthcare-assistant/2-app-with-instrumentation/agent.py file, which are required to collect traces:

python 
import os
from galileo import galileo_context
from galileo.handlers.langchain import GalileoAsyncCallback

Note about the SDK

This workshop was built using the Galileo LangChain callback handler, GalileoAsyncCallback. For new deployments, we recommend using the SplunkAOAsyncCallback package instead. Refer to the LangChain and LangGraph document for details about this newer SDK.
2

Wrap the graph invocation in a Galileo context

The base version of _process_query_async invokes the graph with no tracing:

python 
    async def _process_query_async(self, messages: List[Dict[str, str]]) -> str:
        if not self.tools:
            self.load_tools()
        self.graph = self._build_graph()

        langchain_messages: List[BaseMessage] = []
        for msg in messages:
            if msg["role"] == "user":
                langchain_messages.append(HumanMessage(content=msg["content"]))
            elif msg["role"] == "assistant":
                langchain_messages.append(AIMessage(content=msg["content"]))

        result = await self.graph.ainvoke(
            {"messages": langchain_messages},
            self.langgraph_config,
        )
        if result["messages"]:
            return result["messages"][-1].content
        return "No response generated"

We’ve updated the ~/workshop/healthcare-assistant/2-app-with-instrumentation/agent.py file to update this function to open a galileo_context, start a session keyed to the agent’s session_id, and attach a fresh GalileoAsyncCallback to the run config:

python 
    async def _process_query_async(self, messages: List[Dict[str, str]]) -> str:
        if not self.tools:
            self.load_tools()
        self.graph = self._build_graph()

        langchain_messages: List[BaseMessage] = []
        for msg in messages:
            if msg["role"] == "user":
                langchain_messages.append(HumanMessage(content=msg["content"]))
            elif msg["role"] == "assistant":
                langchain_messages.append(AIMessage(content=msg["content"]))

        with galileo_context(
            project=os.getenv("GALILEO_PROJECT"),
            log_stream=os.getenv("GALILEO_LOG_STREAM"),
        ):
            galileo_context.start_session(external_id=self.session_id)

            # One callback per request keeps each user turn in its own trace.
            callback = GalileoAsyncCallback()
            run_config = {**self.langgraph_config, "callbacks": [callback]}

            result = await self.graph.ainvoke(
                {"messages": langchain_messages},
                run_config,
            )
        if result["messages"]:
            return result["messages"][-1].content
        return "No response generated"

Why a single callback per request?

Creating one GalileoAsyncCallback per call to _process_query_async keeps each user turn in its own trace. Because it’s attached to the LangGraph run config, every node’s LLM and tool call becomes a nested span under that same trace, giving you the end-to-end view of a turn instead of a pile of disconnected spans.

Why does this app use GalileoAsyncCallback rather than GalileoCallback?

Click here to see the answer
Because the agent streams/invokes the graph asynchronously (self.graph.ainvoke(...)). The async callback matches the async run. A synchronous app that called invoke(...) would use GalileoCallback instead.