Chain Tracing

import os
from dotenv import load_dotenv
import noveum_trace
from noveum_trace import NoveumTraceCallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

load_dotenv()

# Initialize Noveum Trace
noveum_trace.init(
    api_key=os.getenv("NOVEUM_API_KEY"),
    project="demo-chatbot",
    environment="development"
)

def create_document_processing_chain():
    """Create a document processing chain with tracing."""
    # Initialize callback handler
    callback_handler = NoveumTraceCallbackHandler()
    
    # Create LLM (callbacks will be passed at chain level)
    llm = ChatOpenAI(
        model="gpt-4",
        temperature=0.7
    )
    
    # Define prompts for each step
    summarize_prompt = ChatPromptTemplate.from_template("""
    Summarize the following document in 2-3 sentences:
    
    Document: {document}
    
    Summary:
    """)
    
    analyze_prompt = ChatPromptTemplate.from_template("""
    Analyze the following summary and identify key themes:
    
    Summary: {summary}
    
    Key themes:
    """)
    
    format_prompt = ChatPromptTemplate.from_template("""
    Format the following analysis into a structured report:
    
    Analysis: {analysis}
    
    Structured Report:
    """)
    
    # Create the chain with three stages: summarize -> analyze -> format
    chain = (
        {"document": RunnablePassthrough()} |
        summarize_prompt | llm | StrOutputParser() |
        {"summary": RunnablePassthrough()} |
        analyze_prompt | llm | StrOutputParser() |
        {"analysis": RunnablePassthrough()} |
        format_prompt | llm | StrOutputParser()
    )
    
    # Return both chain and handler for config-based usage
    return chain, callback_handler

def run_document_processing():
    """Run the document processing chain with tracing."""
    print("=== Document Processing Chain Tracing ===")
    
    # Create the chain
    chain, handler = create_document_processing_chain()
    
    # Process a document
    document = """
    Artificial Intelligence (AI) is transforming industries across the globe. 
    From healthcare to finance, AI technologies are enabling new capabilities 
    and improving efficiency. Machine learning algorithms can process vast 
    amounts of data to identify patterns and make predictions. However, 
    challenges remain in areas like bias, transparency, and ethical considerations.
    """
    
    # Pass callbacks via config
    result = chain.invoke(
        document,
        config={"callbacks": [handler]}
    )
    
    print(f"Final result: {result}")
    return result

if __name__ == "__main__":
    run_document_processing()
    noveum_trace.flush()

pip install noveum-trace langchain-openai python-dotenv

export NOVEUM_API_KEY="your-noveum-api-key"
export OPENAI_API_KEY="your-openai-api-key"

from noveum_trace import NoveumTraceCallbackHandler

# Create callback handler
handler = NoveumTraceCallbackHandler()

# Manually start a trace
handler.start_trace("my-custom-trace")

# Your LangChain operations here
llm = ChatOpenAI()
response = llm.invoke(
    "Hello world",
    config={"callbacks": [handler]}
)

# Manually end the trace
handler.end_trace()

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Create chain using LCEL
prompt = ChatPromptTemplate.from_template("Process this input: {input}")
chain = prompt | ChatOpenAI() | StrOutputParser()

# Pass callbacks and metadata via config
result = chain.invoke(
    {"input": "Your input"},
    config={
        "callbacks": [handler],
        "metadata": {
            "noveum": {
                "name": "parent_llm",
                "parent_name": "parent_span"
            }
        }
    }
)

metadata = {
    "noveum": {
        "name": "custom_span_name",        # Custom name for this span
        "parent_name": "parent_span_name"  # Name of parent span to attach to
    }
}

# Enable LangChain parent ID resolution
handler = NoveumTraceCallbackHandler(use_langchain_assigned_parent=True)

# LangChain will automatically resolve parent relationships
# based on parent_run_id in the callback events

from langgraph.graph import StateGraph, END
from langchain_core.runnables import RunnableConfig

def route_function(state, config):
    """Routing function that emits routing events."""
    
    # Make routing decision
    decision = "next_node" if state["count"] < 5 else "finish"
    
    # Emit routing event (if callbacks available)
    if config and config.get("callbacks"):
        callbacks = config["callbacks"]
        
        # Normalize callbacks into an iterable
        if not isinstance(callbacks, (list, tuple)):
            callbacks = [callbacks]
        
        # Iterate over each callback handler
        for handler in callbacks:
            if hasattr(handler, 'on_custom_event'):
                handler.on_custom_event(
                    "langgraph.routing_decision",
                    {
                        "source_node": "current_node",
                        "target_node": decision,
                        "decision": decision,
                        "reason": f"Count {state['count']} {'< 5' if state['count'] < 5 else '>= 5'}",
                        "confidence": 0.9,
                        "state_snapshot": state,
                    }
                )
    
    return decision

# Create graph with routing
workflow = StateGraph(State)
workflow.add_node("process", process_node)
workflow.add_node("finish", finish_node)
workflow.add_conditional_edges(
    "process",
    route_function,
    {"next_node": "process", "finish": "finish"}
)

# Run with callback handler
app = workflow.compile()
result = app.invoke(
    {"count": 0},
    config={"callbacks": [handler]}
)

from langchain_core.runnables import RunnableBranch

def create_conditional_chain():
    """Create a chain with conditional logic."""
    callback_handler = NoveumTraceCallbackHandler()
    llm = ChatOpenAI()
    
    # Different processing for different document types
    technical_prompt = ChatPromptTemplate.from_template("""
    Provide a technical analysis of: {document}
    """)
    
    financial_prompt = ChatPromptTemplate.from_template("""
    Provide a financial analysis of: {document}
    """)
    
    general_prompt = ChatPromptTemplate.from_template("""
    Provide a general analysis of: {document}
    """)
    
    # Create conditional chain using RunnableBranch
    # RunnableBranch evaluates conditions and selects one branch
    chain = (
        {"document": RunnablePassthrough()} |
        RunnableBranch(
            (lambda x: "technical" in x["document"].lower(), technical_prompt | llm | StrOutputParser()),
            (lambda x: "financial" in x["document"].lower(), financial_prompt | llm | StrOutputParser()),
            general_prompt | llm | StrOutputParser()  # default branch
        )
    )
    
    # Use chain with config
    # result = chain.invoke(
    #     {"document": "Your document"},
    #     config={"callbacks": [callback_handler]}
    # )
    
    return chain