karenina.adapters.langchain.trace

trace

Trace conversion and harmonization utilities for LangChain messages.

This module provides utilities for: 1. Converting LangChain message history to standardized trace formats 2. Harmonizing LangGraph agent responses into trace strings 3. Extracting final AI messages from agent responses

Two output formats

• raw_trace: String with "--- Message Type ---" delimiters (legacy/database)
• trace_messages: List of dicts matching TypeScript TraceMessage interface

Example

from langchain_core.messages import AIMessage, ToolMessage messages = [ ... AIMessage(content="Let me search for that."), ... ToolMessage(content="results...", tool_call_id="call_abc"), ... AIMessage(content="The answer is 42."), ... ] raw = langchain_messages_to_raw_trace(messages) structured = langchain_messages_to_trace_messages(messages)

Or harmonize agent response directly

trace = harmonize_agent_response({"messages": messages})

Functions

extract_final_ai_message_from_response

extract_final_ai_message_from_response(
    response: Any,
) -> tuple[str | None, str | None]

Extract only the final AI text response from an agent response.

This function works with the original agent response before harmonization, checking message types directly rather than parsing strings.

Parameters:

Name	Type	Description	Default
response	Any	Response from a LangGraph agent (messages list, dict with 'messages', or state dict)	required

Returns:

Type	Description
str | None	Tuple of (extracted_message, error_message) where:
str | None	extracted_message: The final AI text content, or None if extraction failed
tuple[str | None, str | None]	error_message: Error description if extraction failed, or None if successful

Error cases

• Empty or no messages
• Last message is not an AIMessage
• Final AIMessage has no text content (only tool calls)

Example

from langchain_core.messages import AIMessage messages = [AIMessage(content="Final answer")] message, error = extract_final_ai_message_from_response(messages) print(message) # "Final answer"

Source code in src/karenina/adapters/langchain/trace.py

def extract_final_ai_message_from_response(response: Any) -> tuple[str | None, str | None]:
    """Extract only the final AI text response from an agent response.

    This function works with the original agent response before harmonization,
    checking message types directly rather than parsing strings.

    Args:
        response: Response from a LangGraph agent (messages list, dict with
            'messages', or state dict)

    Returns:
        Tuple of (extracted_message, error_message) where:
        - extracted_message: The final AI text content, or None if extraction failed
        - error_message: Error description if extraction failed, or None if successful

    Error cases:
        - Empty or no messages
        - Last message is not an AIMessage
        - Final AIMessage has no text content (only tool calls)

    Example:
        >>> from langchain_core.messages import AIMessage
        >>> messages = [AIMessage(content="Final answer")]
        >>> message, error = extract_final_ai_message_from_response(messages)
        >>> print(message)  # "Final answer"
    """
    try:
        from langchain_core.messages import AIMessage
    except ImportError:
        return None, "langchain_core not available"

    # Extract messages list from various response formats
    messages = None

    if response is None:
        return None, "Empty response"

    # Handle nested agent state dict: {'agent': {'messages': [...]}}
    if isinstance(response, dict):
        if "agent" in response and isinstance(response["agent"], dict) and "messages" in response["agent"]:
            messages = response["agent"]["messages"]
        elif "messages" in response:
            messages = response["messages"]
    # Handle list of messages directly
    elif isinstance(response, list):
        messages = response
    # Handle single message
    elif isinstance(response, AIMessage):
        messages = [response]

    if not messages or len(messages) == 0:
        return None, "No messages found in response"

    # Get the last message
    last_message = messages[-1]

    # Check if it's an AIMessage
    if not isinstance(last_message, AIMessage):
        return None, "Last message is not an AIMessage"

    # Extract content
    content = last_message.content if last_message.content else ""

    # Handle list content (convert to string)
    if isinstance(content, list):
        text_parts = []
        for item in content:
            if isinstance(item, str):
                text_parts.append(item)
            elif isinstance(item, dict) and "text" in item:
                text_parts.append(str(item["text"]))
        content = " ".join(text_parts) if text_parts else ""

    # Check if content is empty
    if not content or not content.strip():
        # Check if there are tool calls but no text content
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            return None, "Final AI message has no text content (only tool calls)"
        return None, "Final AI message has no content"

    return content.strip(), None

harmonize_agent_response

harmonize_agent_response(response: Any) -> str

Harmonize LangGraph agent response into a single trace string.

LangGraph agents return multiple messages instead of a single response. This function extracts AI and Tool messages (excluding system and human messages) and returns the complete agent execution trace.

Parameters:

Name	Type	Description	Default
response	Any	Response from a LangGraph agent, which may be: - A single message with content attribute - List of messages - Dict with 'messages' key (from ainvoke) - Nested dict {'agent': {'messages': [...]}} (from astream)	required

Returns:

Type	Description
str	A single string containing the complete agent trace with reasoning
str	and tool usage in the "--- Message Type ---" format.

Example

from langchain_core.messages import AIMessage, ToolMessage messages = [ ... AIMessage(content="I need to search for information."), ... ToolMessage(content="Search results: ...", tool_call_id="call_123"), ... AIMessage(content="Based on the search, the answer is 42.") ... ] trace = harmonize_agent_response({"messages": messages}) print("Full agent trace with reasoning and tool usage")

Source code in src/karenina/adapters/langchain/trace.py

def harmonize_agent_response(response: Any) -> str:
    """Harmonize LangGraph agent response into a single trace string.

    LangGraph agents return multiple messages instead of a single response.
    This function extracts AI and Tool messages (excluding system and human
    messages) and returns the complete agent execution trace.

    Args:
        response: Response from a LangGraph agent, which may be:
            - A single message with content attribute
            - List of messages
            - Dict with 'messages' key (from ainvoke)
            - Nested dict {'agent': {'messages': [...]}} (from astream)

    Returns:
        A single string containing the complete agent trace with reasoning
        and tool usage in the "--- Message Type ---" format.

    Example:
        >>> from langchain_core.messages import AIMessage, ToolMessage
        >>> messages = [
        ...     AIMessage(content="I need to search for information."),
        ...     ToolMessage(content="Search results: ...", tool_call_id="call_123"),
        ...     AIMessage(content="Based on the search, the answer is 42.")
        ... ]
        >>> trace = harmonize_agent_response({"messages": messages})
        >>> print("Full agent trace with reasoning and tool usage")
    """
    if response is None:
        return ""

    # Handle single message with content attribute
    if hasattr(response, "content"):
        return str(response.content)

    # Handle nested agent state dict (from astream): {'agent': {'messages': [...]}}
    if isinstance(response, dict):
        if "agent" in response and isinstance(response["agent"], dict) and "messages" in response["agent"]:
            messages = response["agent"]["messages"]
            return _extract_agent_trace(messages)
        # Handle flat state dict with 'messages' key (from ainvoke)
        elif "messages" in response:
            messages = response["messages"]
            return _extract_agent_trace(messages)

    # Handle list of messages directly
    if isinstance(response, list):
        return _extract_agent_trace(response)

    # Fallback: convert to string
    return str(response)

langchain_messages_to_raw_trace

langchain_messages_to_raw_trace(
    messages: list[BaseMessage],
    include_system: bool = False,
) -> str

Convert LangChain messages to raw trace string format.

This produces the legacy format with "--- Message Type ---" delimiters that is used for: - Database storage (raw_llm_response TEXT column) - Regex-based highlighting in the frontend - Rubric evaluation on raw text

Parameters:

Name	Type	Description	Default
messages	list[BaseMessage]	List of LangChain BaseMessage instances	required
include_system	bool	If True, include SystemMessage blocks in output	False

Returns:

Type	Description
str	Trace string with message blocks separated by double newlines

Format

--- AI Message --- I'll search for that information.

Tool Calls: search (call_abc123) Call ID: abc123 Args: {"query": "..."}

--- Tool Message (call_id: call_abc123) --- {"result": "search results..."}

--- AI Message --- Based on the results, the answer is 42.

Source code in src/karenina/adapters/langchain/trace.py

def langchain_messages_to_raw_trace(
    messages: list[BaseMessage],
    include_system: bool = False,
) -> str:
    """Convert LangChain messages to raw trace string format.

    This produces the legacy format with "--- Message Type ---" delimiters that
    is used for:
    - Database storage (raw_llm_response TEXT column)
    - Regex-based highlighting in the frontend
    - Rubric evaluation on raw text

    Args:
        messages: List of LangChain BaseMessage instances
        include_system: If True, include SystemMessage blocks in output

    Returns:
        Trace string with message blocks separated by double newlines

    Format:
        --- AI Message ---
        I'll search for that information.

        Tool Calls:
          search (call_abc123)
           Call ID: abc123
           Args: {"query": "..."}

        --- Tool Message (call_id: call_abc123) ---
        {"result": "search results..."}

        --- AI Message ---
        Based on the results, the answer is 42.
    """
    try:
        from langchain_core.messages import SystemMessage
    except ImportError:
        # Fallback if langchain_core not available
        return _fallback_to_raw_trace(messages, include_system)

    trace_parts: list[str] = []

    for msg in messages:
        # Skip system messages unless requested
        if isinstance(msg, SystemMessage) and not include_system:
            continue

        formatted = _format_langchain_message(msg)
        if formatted.strip():
            trace_parts.append(formatted)

    return "\n\n".join(trace_parts)

langchain_messages_to_trace_messages

langchain_messages_to_trace_messages(
    messages: list[BaseMessage],
    include_system: bool = False,
) -> list[dict[str, Any]]

Convert LangChain messages to structured TraceMessage list.

This produces the new structured format that matches the TypeScript TraceMessage interface used by the frontend structured view.

Parameters:

Name	Type	Description	Default
messages	list[BaseMessage]	List of LangChain BaseMessage instances	required
include_system	bool	If True, include system messages in output	False

Returns:

Type	Description
list[dict[str, Any]]	List of TraceMessage dicts with keys: - role: "system" | "user" | "assistant" | "tool" - content: str (message text) - block_index: int (for navigation) - tool_calls?: list[ToolCall] (on assistant messages with tools) - tool_result?: ToolResultMeta (on tool messages)

Example output

[ { "role": "assistant", "content": "Let me search for that.", "block_index": 0, "tool_calls": [ {"id": "call_abc", "name": "search", "input": {"query": "..."}} ] }, { "role": "tool", "content": "results...", "block_index": 1, "tool_result": {"tool_use_id": "call_abc", "is_error": False} }, { "role": "assistant", "content": "The answer is 42.", "block_index": 2 } ]

Source code in src/karenina/adapters/langchain/trace.py

def langchain_messages_to_trace_messages(
    messages: list[BaseMessage],
    include_system: bool = False,
) -> list[dict[str, Any]]:
    """Convert LangChain messages to structured TraceMessage list.

    This produces the new structured format that matches the TypeScript
    TraceMessage interface used by the frontend structured view.

    Args:
        messages: List of LangChain BaseMessage instances
        include_system: If True, include system messages in output

    Returns:
        List of TraceMessage dicts with keys:
            - role: "system" | "user" | "assistant" | "tool"
            - content: str (message text)
            - block_index: int (for navigation)
            - tool_calls?: list[ToolCall] (on assistant messages with tools)
            - tool_result?: ToolResultMeta (on tool messages)

    Example output:
        [
            {
                "role": "assistant",
                "content": "Let me search for that.",
                "block_index": 0,
                "tool_calls": [
                    {"id": "call_abc", "name": "search", "input": {"query": "..."}}
                ]
            },
            {
                "role": "tool",
                "content": "results...",
                "block_index": 1,
                "tool_result": {"tool_use_id": "call_abc", "is_error": False}
            },
            {
                "role": "assistant",
                "content": "The answer is 42.",
                "block_index": 2
            }
        ]
    """
    try:
        from langchain_core.messages import (
            AIMessage,
            HumanMessage,
            SystemMessage,
            ToolMessage,
        )
    except ImportError:
        # Fallback if langchain_core not available
        return _fallback_to_trace_messages(messages, include_system)

    result: list[dict[str, Any]] = []
    block_index = 0

    for msg in messages:
        # Skip system messages unless requested
        if isinstance(msg, SystemMessage) and not include_system:
            continue

        if isinstance(msg, SystemMessage):
            result.append(
                {
                    "role": "system",
                    "content": str(msg.content) if msg.content else "",
                    "block_index": block_index,
                }
            )
            block_index += 1

        elif isinstance(msg, HumanMessage):
            result.append(
                {
                    "role": "user",
                    "content": str(msg.content) if msg.content else "",
                    "block_index": block_index,
                }
            )
            block_index += 1

        elif isinstance(msg, AIMessage):
            trace_msg: dict[str, Any] = {
                "role": "assistant",
                "content": str(msg.content) if msg.content else "",
                "block_index": block_index,
            }

            # Add tool_calls if present
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                tool_calls = []
                for tc in msg.tool_calls:
                    if isinstance(tc, dict):
                        tool_calls.append(
                            {
                                "id": tc.get("id") or "",
                                "name": tc.get("name") or "",
                                "input": tc.get("args") or {},
                            }
                        )
                    else:
                        # Object-style tool call
                        tool_calls.append(
                            {
                                "id": getattr(tc, "id", "") or "",
                                "name": getattr(tc, "name", "") or "",
                                "input": getattr(tc, "args", {}) or {},
                            }
                        )
                if tool_calls:
                    trace_msg["tool_calls"] = tool_calls

            result.append(trace_msg)
            block_index += 1

        elif isinstance(msg, ToolMessage):
            content = str(msg.content) if msg.content else ""
            tool_call_id = getattr(msg, "tool_call_id", "") or ""

            result.append(
                {
                    "role": "tool",
                    "content": content,
                    "block_index": block_index,
                    "tool_result": {
                        "tool_use_id": tool_call_id,
                        "is_error": _detect_tool_error(content),
                    },
                }
            )
            block_index += 1

    return result