View the complete cookbook example on GitHub: honeyhiveai/cookbook/langgraph-cookbook
This guide demonstrates how to build a sophisticated code generation system using LangGraph and HoneyHive tracing. The system combines Retrieval Augmented Generation (RAG) with self-correction capabilities to generate reliable code solutions. HoneyHive’s tracing capabilities provide comprehensive visibility into the entire process, making debugging and optimization easier than ever.

​
Overview

The system consists of several key components:
  • Documentation loading and processing (traced with HoneyHive)
  • LLM setup with structured output (monitored for performance)
  • A LangGraph workflow for code generation and validation (with detailed tracing)
  • HoneyHive tracing for monitoring and debugging (real-time insights)

​
Prerequisites

Before running this code, ensure you have the following:
  • Python 3.x
  • Required API keys:
    • HoneyHive API key (for comprehensive tracing)
    • OpenAI API key
    • Anthropic API key
  • Required packages: 
    pip install langchain langchain-community langgraph beautifulsoup4 honeyhive langchain-openai langchain-anthropic

​
Code Implementation

​
Environment Setup and Imports

import getpass
import os
import sys
from bs4 import BeautifulSoup as Soup
from langchain_community.document_loaders.recursive_url_loader import RecursiveUrlLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field
from typing import List
from typing_extensions import TypedDict, Annotated
from langgraph.graph import END, StateGraph, START
from honeyhive import HoneyHiveTracer, trace

# Set up environment variables
# Delete the below if it is already set in the environment
os.environ["HONEYHIVE_API_KEY"] = "your honeyhive api key"
os.environ["HONEYHIVE_PROJECT"] = "your honeyhive project"
os.environ["HONEYHIVE_SOURCE"] = "your honeyhive source"
os.environ["OPENAI_API_KEY"] = "your openai api key"
os.environ["ANTHROPIC_API_KEY"] = "your anthropic api key"

​
HoneyHive Tracing Setup

HoneyHive’s tracing setup provides comprehensive monitoring capabilities: 
# Initialize HoneyHive tracer with detailed configuration
HoneyHiveTracer.init(
    api_key=os.environ.get("HONEYHIVE_API_KEY", "your honeyhive api key"),
    project=os.environ.get("HONEYHIVE_PROJECT", "your honeyhive project"),
    source="development",
    session_name="LangGraph Code Generation"
)
With this setup, you get:
  • Real-time monitoring of all traced functions
  • Detailed performance metrics
  • Error tracking and debugging
  • Session-based analytics
  • Custom metadata support

​
Documentation Loading

The system uses RecursiveUrlLoader with HoneyHive tracing to monitor documentation loading: 
@trace
def load_documentation(url):
    """Load documentation from a URL"""
    print("---LOADING DOCUMENTATION---")
    loader = RecursiveUrlLoader(
        url=url, max_depth=20, extractor=lambda x: Soup(x, "html.parser").text
    )
    docs = loader.load()

    # Sort and concatenate documentation
    d_sorted = sorted(docs, key=lambda x: x.metadata["source"])
    d_reversed = list(reversed(d_sorted))
    concatenated_content = "\n\n\n --- \n\n\n".join(
        [doc.page_content for doc in d_reversed]
    )
    print("---DOCUMENTATION LOADED---")
    return concatenated_content
HoneyHive tracing here provides:
  • Loading time metrics
  • Document count tracking
  • Error handling for failed loads
  • Memory usage monitoring
  • URL accessibility tracking

​
Data Model

The system uses Pydantic with HoneyHive tracing for structured output validation: 
class code(BaseModel):
    """Schema for code solutions to questions about HoneyHive."""
    prefix: str = Field(description="Description of the problem and approach")
    imports: str = Field(description="Code block import statements")
    code: str = Field(description="Code block not including import statements")
HoneyHive helps track:
  • Model validation success rates
  • Field completion rates
  • Schema compliance
  • Data quality metrics

​
LLM Setup

The system uses Claude with HoneyHive tracing for comprehensive LLM monitoring: 
code_gen_prompt_claude = ChatPromptTemplate.from_messages([
    (
        "system",
        """<instructions> You are a coding assistant with expertise in HoneyHive. \n 
        Here is the LCEL documentation:  \n ------- \n  {context} \n ------- \n Answer the user  question based on the \n 
        above provided documentation. Ensure any code you provide can be executed with all required imports and variables \n
        defined. Structure your answer: 1) a prefix describing the code solution, 2) the imports, 3) the functioning code block. \n
        Invoke the code tool to structure the output correctly. </instructions> \n Here is the user question:""",
    ),
    ("placeholder", "{messages}"),
])

@trace
def setup_llm():
    """Set up the LLM with structured output"""
    expt_llm_claude = "claude-3-7-sonnet-latest"
    llm_claude = ChatAnthropic(
        model=expt_llm_claude,
        default_headers={"anthropic-beta": "tools-2024-04-04"},
    )
    structured_llm_claude = llm_claude.with_structured_output(code, include_raw=True)
    return structured_llm_claude

llm = setup_llm()
HoneyHive provides:
  • LLM response time tracking
  • Token usage monitoring
  • Error rate tracking
  • Model performance analytics
  • Cost tracking per request

​
Documentation Loading and Chain Setup

First, load the documentation and set up the code generation chain: 
# Load HoneyHive documentation
documentation = load_documentation("https://docs.honeyhive.ai/introduction/quickstart")

# Helper function for Claude output processing
@trace
def parse_output(solution):
    """Parse the structured output from Claude"""
    if "parsed" in solution:
        return solution["parsed"]
    return solution

# Set up the code generation chain
code_gen_chain = code_gen_prompt_claude | llm | parse_output

​
LangGraph Implementation

The system uses LangGraph with HoneyHive tracing for workflow monitoring:
  1. State Definition:
class GraphState(TypedDict):
    """
    Represents the state of our graph.

    Attributes:
        error : Binary flag for control flow to indicate whether test error was tripped
        messages : With user question, error messages, reasoning
        generation : str with code solution
        iterations : Number of tries
    """
    error: str
    messages: List
    generation: str
    iterations: int
  1. Graph Nodes with HoneyHive tracing:
Generate Node: 
@trace
def generate(state: GraphState):
    """
    Generate a code solution

    Args:
        state (dict): The current graph state

    Returns:
        state (dict): New key added to state, generation
    """
    print("---GENERATING CODE SOLUTION---")
    messages = state["messages"]
    
    # Generate the code solution
    generation = code_gen_chain.invoke(
        {"messages": messages, "context": documentation}
    )
    
    print("---CODE SOLUTION GENERATED---")
    return {"generation": generation, "iterations": state["iterations"] + 1}
Code Check Node: 
@trace
def code_check(state: GraphState):
    """
    Verify that the code solution works by:
    1. Checking that imports don't error
    2. Checking that code execution doesn't error
    
    Args:
        state (dict): The current graph state with the code solution

    Returns:
        state (dict): State with updated error flag and messages
    """
    print("---CHECKING CODE SOLUTION---")
    generation = state["generation"]
    
    # Extract imports and code
    imports = generation.imports
    code_block = generation.code
    
    # Check imports
    error_msg = None
    try:
        print("---CHECKING IMPORTS---")
        exec(imports)
        print("Imports OK!")
    except Exception as e:
        error_msg = f"Import error: {str(e)}"
        print(f"Import error: {e}")
    
    # If imports okay, check code execution
    if not error_msg:
        try:
            print("---CHECKING CODE EXECUTION---")
            # Only syntax check (don't execute the code for safety)
            compile(code_block, "<string>", "exec")
            print("Code syntax OK!")
        except Exception as e:
            error_msg = f"Code execution error: {str(e)}"
            print(f"Code execution error: {e}")
    
    # Update state based on checks
    has_error = error_msg is not None
    if has_error:
        messages = state["messages"] + [
            (
                "assistant",
                f"There was an error with the code: {error_msg}. Let me fix it.",
            )
        ]
        return {"error": "yes", "messages": messages}
    else:
        return {"error": "no"}
Reflect Node: 
@trace
def reflect(state: GraphState):
    """
    Reflect on the code solution and improve it
    
    Args:
        state (dict): The current graph state

    Returns:
        state (dict): State with updated messages for reflection
    """
    print("---REFLECTING ON SOLUTION---")
    
    # Add a reflection step to messages for the next iteration
    messages = state["messages"] + [
        (
            "assistant", 
            "Let me review the code once more to make sure it's correct and follows best practices."
        )
    ]
    
    return {"messages": messages}
Decision Node: 
@trace
def decide_to_finish(state: GraphState):
    """
    Decide whether to finish or try again
    
    Args:
        state (dict): The current graph state

    Returns:
        str: "reflect", "finish", or "generate"
    """
    error = state["error"]
    iterations = state["iterations"]
    max_iterations = 3
    
    # If there's an error and we haven't reached max iterations, generate again
    if error == "yes" and iterations < max_iterations:
        print(f"---ERROR DETECTED, REGENERATING (Iteration {iterations}/{max_iterations})---")
        return "generate"
    
    # If no error but want to reflect before finishing (optional)
    # Change flag to "reflect" to enable this branch
    flag = "do not reflect"  # Change to "reflect" to enable reflection
    if error == "no" and flag == "reflect":
        print("---NO ERROR, REFLECTING BEFORE FINISHING---")
        return "reflect"
    
    # Otherwise, finish
    print("---FINISHING---")
    return "finish"
  1. Graph Construction with comprehensive tracing:
@trace
def build_graph():
    """Build the LangGraph for code generation"""
    # Create a graph
    graph_builder = StateGraph(GraphState)
    
    # Add nodes
    graph_builder.add_node("generate", generate)
    graph_builder.add_node("code_check", code_check)
    graph_builder.add_node("reflect", reflect)
    
    # Add edges
    graph_builder.add_edge(START, "generate")
    graph_builder.add_edge("generate", "code_check")
    
    # Add conditional edges
    graph_builder.add_conditional_edges(
        "code_check",
        decide_to_finish,
        {
            "generate": "generate",
            "reflect": "reflect",
            "finish": END,
        },
    )
    graph_builder.add_edge("reflect", "generate")
    
    # Compile the graph
    return graph_builder.compile()

# Create the graph
graph = build_graph()

​
Main Execution Function

The main function to run the graph with a question: 
# Function to run the graph with a question
@trace
def solve_coding_question(question):
    """Run the graph to solve a coding question"""
    # Initialize the state
    state = {
        "error": "no",
        "messages": [("human", question)],
        "generation": None,
        "iterations": 0,
    }
    
    # Execute the graph
    result = graph.invoke(state)
    
    # Return the generated code solution
    return result["generation"]
HoneyHive tracing provides:
  • Node execution time tracking
  • Edge traversal monitoring
  • State transition tracking
  • Error propagation analysis
  • Performance bottlenecks identification

​
Usage

To use the code generation system with HoneyHive monitoring: 
# Example usage
if __name__ == "__main__":
    question = "How can I use HoneyHive tracing with LangGraph?"
    solution = solve_coding_question(question)
    
    print("\n=== FINAL SOLUTION ===")
    print(f"\n{solution.prefix}\n")
    print(f"IMPORTS:\n{solution.imports}\n")
    print(f"CODE:\n{solution.code}")
    
    # This will end the current session in HoneyHive
    # For a new session, call HoneyHiveTracer.init() again

​
Key Features

  1. RAG Integration: HoneyHive traces document retrieval and processing
  2. Self-Correction: Monitors validation and improvement cycles
  3. Structured Output: Tracks schema compliance and data quality
  4. HoneyHive Tracing: Provides comprehensive monitoring and debugging
  5. Maximum Iterations: Tracks iteration counts and success rates

​
Best Practices

  1. Always set up proper environment variables before running
  2. Monitor the HoneyHive dashboard for:
    • Performance metrics
    • Error rates
    • Cost analysis
    • Usage patterns
  3. Adjust the max_depth parameter in RecursiveUrlLoader based on your needs
  4. Customize the reflection step based on your specific use case
  5. Implement proper error handling for production use

​
HoneyHive Dashboard Insights

The HoneyHive dashboard provides valuable insights:
  1. Performance Metrics:
    • Response times
    • Throughput
    • Resource usage
  2. Error Tracking:
    • Error rates
    • Stack traces
    • Error patterns
  3. Cost Analysis:
    • API usage costs
    • Resource consumption
    • Cost optimization opportunities
  4. Usage Patterns:
    • Peak usage times
    • Common operations
    • User behavior
Langgraph Trace

​
Conclusion

This implementation provides a robust foundation for code generation with self-correction capabilities. The combination of LangGraph and HoneyHive tracing ensures reliable and monitored code generation processes. HoneyHive’s comprehensive tracing capabilities make it easier to:
  • Debug issues
  • Optimize performance
  • Track costs
  • Monitor quality
  • Scale the system