LangChain & LangGraph Observability

import os
import json

from langgraph.graph import StateGraph, MessagesState, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_google_community import GoogleSearchAPIWrapper

from fi_instrumentation import register
from fi_instrumentation.fi_types import (
    ProjectType,
    EvalName,
    EvalTag,
    EvalTagType,
    EvalSpanKind,
    ModelChoices,
)
# the PyPI distribution is traceai-langchain; the import path uses an underscore
from traceai_langchain import LangChainInstrumentor

You should see no import errors. If traceai_langchain fails to import, the package isn’t installed: run pip install traceai-langchain.

Each EvalTag scores one span kind against one metric. This agent has an AGENT span (the reasoning step) and a TOOL span (the Google Search call), so tag both.

eval_tags = [
    # each tag pins one eval to one span kind, so the platform knows which spans to score
    EvalTag(
        type=EvalTagType.OBSERVATION_SPAN,
        value=EvalSpanKind.AGENT,
        eval_name=EvalName.COMPLETENESS,
        config={},
        mapping={"input": "raw.input", "output": "raw.output"},
        custom_eval_name="Completeness",
        model=ModelChoices.TURING_LARGE,
    ),
    EvalTag(
        type=EvalTagType.OBSERVATION_SPAN,
        value=EvalSpanKind.AGENT,
        eval_name=EvalName.GROUNDEDNESS,
        config={},
        mapping={"input": "raw.input", "output": "raw.output"},
        custom_eval_name="Groundedness",
        model=ModelChoices.TURING_LARGE,
    ),
    # EVALUATE_FUNCTION_CALLING is the only eval here that needs a TOOL span kind
    EvalTag(
        type=EvalTagType.OBSERVATION_SPAN,
        value=EvalSpanKind.TOOL,
        eval_name=EvalName.EVALUATE_FUNCTION_CALLING,
        config={},
        mapping={"input": "raw.input", "output": "raw.output"},
        custom_eval_name="Tool_Calling",
        model=ModelChoices.TURING_LARGE,
    ),
    EvalTag(
        type=EvalTagType.OBSERVATION_SPAN,
        value=EvalSpanKind.AGENT,
        eval_name=EvalName.DETECT_HALLUCINATION,
        config={},
        mapping={"input": "raw.input", "output": "raw.output"},
        custom_eval_name="Hallucination",
        model=ModelChoices.TURING_LARGE,
    ),
]

mapping points each eval’s required inputs at span attributes; raw.input and raw.output are the span’s recorded input and output. Browse the full list of built-in evals for other metrics you can substitute in.

You should see this list build with no exception. EvalTag validates its eval name, model, config, and mapping keys in __post_init__, so a typo in any of them raises a ValueError here, not later when the trace is scored.

# register() must run first: instrument() needs the tracer_provider it returns
trace_provider = register(
    project_type=ProjectType.EXPERIMENT,
    project_name="LangGraph-Google-Search-App",
    project_version_name="v1",
    eval_tags=eval_tags,
)

LangChainInstrumentor().instrument(tracer_provider=trace_provider)

register() wires the eval tags into an OpenTelemetry trace provider. LangChainInstrumentor().instrument() patches LangChain and LangGraph so every node call emits a span through that provider. You should see no output here: instrumentation is silent until something runs.

Prove the pipeline end to end before adding any graph or tool complexity.

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm.invoke("Reply with the single word: ready")

Open app.futureagi.com, go to the LangGraph-Google-Search-App project, and confirm one trace has arrived. If it has, the keys, register(), and instrument() are all wired correctly, and the rest of this tutorial only adds the graph and tools around this same call.

search = GoogleSearchAPIWrapper()
google_tool = Tool(
    name="google_search",
    description="Use this to search Google for current events or factual knowledge.",
    func=search.run,
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([google_tool])

bind_tools gives the model the option to call google_search, not the obligation. The router in the next step reads that decision off the model’s response. GoogleSearchAPIWrapper() reads GOOGLE_API_KEY and GOOGLE_CSE_ID from the environment and raises immediately if either is missing, so a bad key surfaces here rather than mid-graph.

class AgentState(MessagesState):
    # LangGraph rejects writes to keys outside the state schema, so this has to be
    # declared here even though only tool_node writes to it
    intermediate_steps: list


def agent_node(state: AgentState) -> AgentState:
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": messages + [response], "intermediate_steps": state.get("intermediate_steps", [])}


def tool_node(state: AgentState) -> AgentState:
    messages = state["messages"]
    tool_call = messages[-1].tool_calls[0]
    args = tool_call.get("args") or json.loads(tool_call.get("arguments", "{}"))
    result = google_tool.invoke(args)
    # appending a ToolMessage to messages is the scratchpad bind_tools expects back
    tool_msg = ToolMessage(tool_call_id=tool_call["id"], content=str(result))
    return {
        "messages": messages + [tool_msg],
        "intermediate_steps": state.get("intermediate_steps", []) + [(messages[-1], tool_msg)],
    }


def router(state: AgentState) -> str:
    msg = state["messages"][-1]
    if getattr(msg, "tool_calls", None):
        return "tool"
    return "final"

The agent node decides whether to answer or call the tool. The tool node runs the Google Search and appends the result. router reads tool_calls off the model’s last message to pick the next node. You should see all three functions and AgentState define with no output; nothing runs until the graph is compiled and invoked.

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tool", tool_node)

graph.set_entry_point("agent")
graph.add_conditional_edges("agent", router, {"tool": "tool", "final": END})
graph.add_edge("tool", "agent")

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

example_queries = [
    "What is the current status of the AWS us-east-1 region?",
    "What is Stripe's current per-transaction fee for US card payments?",
    "What is Zendesk's current refund policy for annual plan cancellations?",
]

for i, query in enumerate(example_queries):
    print(f"\nQUERY {i + 1}: {query}\n")
    # each conversation needs its own thread_id, or MemorySaver replays the previous run's messages
    config = {"configurable": {"thread_id": f"multi-tool-agent-{i}"}}
    output = app.invoke({"messages": [HumanMessage(content=query)]}, config)
    output["messages"][-1].pretty_print()

You should see three answers printed, each preceded by its query. Every node call in each run generated a span under the LangGraph-Google-Search-App project.

The tool’s description is what makes the model decide to search. Bind a deliberately vague one and rerun a query that a live search should answer:

weak_tool = Tool(
    name="google_search",
    description="Use this to search Google.",
    func=search.run,
)

# reassigning the module-level `llm` is enough: agent_node reads it by name on every call
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([weak_tool])

failure_query = "What is Stripe's current per-transaction fee for US card payments?"
config = {"configurable": {"thread_id": "failure-demo-weak"}}
output = app.invoke({"messages": [HumanMessage(content=failure_query)]}, config)
output["messages"][-1].pretty_print()

With the vague description, the model tends to answer from training data instead of calling the tool. Open this trace: there is a single agent span and no tool span, and Completeness and Groundedness on that span score low, because the answer isn’t grounded in anything the trace actually retrieved.

Rebind the original, specific tool description and rerun the same query on a new thread:

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([google_tool])

config = {"configurable": {"thread_id": "failure-demo-fixed"}}
output = app.invoke({"messages": [HumanMessage(content=failure_query)]}, config)
output["messages"][-1].pretty_print()

This time the trace shows agent → tool → agent: Tool_Calling on the tool span passes, and Completeness and Groundedness on the final agent span score noticeably higher than the first run, because the answer is now grounded in the search result. The tool’s description, not just its presence, decides whether the model uses it.

Open app.futureagi.com and go to the LangGraph-Google-Search-App project. Each of the three queries from Step 6 is a separate trace: an agent span shows the model’s reasoning, a tool span shows the google_search call and its result, and a second agent span shows the model turning that result into an answer.

Open a trace and check the eval scores attached to each span: Completeness and Groundedness on the agent spans, Tool_Calling on the tool span, Hallucination on the last agent span.