Skip to content

langdb/pylangdb

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

32 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

LangDB - Multi-Framework AI Agent Tracing and Client

PyPI version License: MIT

Key Features

LangDB exposes two complementary capabilities:

  1. Chat Completions Client – Call LLMs using the LangDb Python client. This works as a drop-in replacement for openai.ChatCompletion while adding automatic usage, cost and latency reporting.
  2. Agent Tracing – Instrument your existing AI framework (ADK, LangChain, CrewAI, etc.) with a single init() call. All calls are routed through the LangDB collector and are enriched with additional metadata regarding the framework is visible on the LangDB dashboard.

⚡ Quick Start (Chat Completions)

pip install pylangdb[client]
from pylangdb.client import LangDb

# Initialize LangDB client
client = LangDb(api_key="your_api_key", project_id="your_project_id")

# Simple chat completion
resp = client.chat.completions.create(
    model="openai/gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}]
)
print(resp.choices[0].message.content)

🚀 Agent Tracing Quick Start

# Install the package with Google ADK support
pip install pylangdb[adk]
# Import and initialize LangDB tracing
# First initialize LangDB before defining any agents
from pylangdb.adk import init
init()

import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    if city.lower() != "new york":
        return {"status": "error", "error_message": f"Weather information for '{city}' is not available."}
    return {"status": "success", "report": "The weather in New York is sunny with a temperature of 25 degrees Celsius (77 degrees Fahrenheit)."}

def get_current_time(city: str) -> dict:
    if city.lower() != "new york":
        return {"status": "error", "error_message": f"Sorry, I don't have timezone information for {city}."}
    tz = ZoneInfo("America/New_York")
    now = datetime.datetime.now(tz)
    return {"status": "success", "report": f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'}

root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash",
    description=("Agent to answer questions about the time and weather in a city." ),
    instruction=("You are a helpful agent who can answer user questions about the time and weather in a city."),
    tools=[get_weather, get_current_time],
)

Note: Always initialize LangDB before importing any framework-specific classes to ensure proper instrumentation.

Example Trace Screenshot

Google ADK Trace Example

🛠️ Supported Frameworks (Tracing)

Framework Installation Import Pattern Key Features
Google ADK pip install pylangdb[adk] from pylangdb.adk import init Automatic sub-agent discovery
OpenAI pip install pylangdb[openai] from pylangdb.openai import init Custom model provider support
LangChain pip install pylangdb[langchain] from pylangdb.langchain import init Automatic chain tracing
CrewAI pip install pylangdb[crewai] from pylangdb.crewai import init Multi-agent crew tracing
Agno pip install pylangdb[agno] from pylangdb.agno import init Tool usage tracing, model interactions

🔧 How It Works

LangDB uses intelligent monkey patching to instrument your AI frameworks at runtime:

👉 Click to see technical details for each framework

Google ADK

  • Patches Agent.__init__ to inject callbacks
  • Tracks agent hierarchies and tool usage
  • Maintains thread context across invocations

OpenAI

  • Intercepts HTTP requests via AsyncOpenAI.post
  • Propagates trace context via headers
  • Correlates spans across agent interactions

LangChain

  • Modifies httpx.Client.send for request tracing
  • Automatically tracks chains and agents
  • Injects trace headers into all requests

CrewAI

  • Intercepts litellm.completion for LLM calls
  • Tracks crew members and task delegation
  • Propagates context through LiteLLM headers

Agno

  • Patches LangDB.invoke and client parameters
  • Traces workflows and model interactions
  • Maintains consistent session context

📦 Installation

# For client library functionality (chat completions, analytics, etc.)
pip install pylangdb[client]

# For framework tracing - install specific framework extras
pip install pylangdb[adk]      # Google ADK tracing
pip install pylangdb[openai]   # OpenAI agents tracing
pip install pylangdb[langchain] # LangChain tracing
pip install pylangdb[crewai]   # CrewAI tracing
pip install pylangdb[agno]     # Agno tracing

🔑 Configuration

Set your credentials (or pass them directly to the init() function):

export LANGDB_API_KEY="your-api-key"
export LANGDB_PROJECT_ID="your-project-id"

Client Usage (Chat Completions)

Initialize LangDb Client

from pylangdb import LangDb

# Initialize with API key and project ID
client = LangDb(api_key="your_api_key", project_id="your_project_id")

Chat Completions

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Say hello!"}
]

response = client.completion(
    model="gemini-1.5-pro-latest",
    messages=messages,
    temperature=0.7,
    max_tokens=100
)

Thread Operations

Get Messages

Retrieve messages from a specific thread:

messages = client.get_messages(thread_id="your_thread_id")

# Access message details
for message in messages:
    print(f"Type: {message.type}")
    print(f"Content: {message.content}")
    if message.tool_calls:
        for tool_call in message.tool_calls:
            print(f"Tool: {tool_call.function.name}")

Get Thread Cost

Get cost and token usage information for a thread:

usage = client.get_usage(thread_id="your_thread_id")
print(f"Total cost: ${usage.total_cost:.4f}")
print(f"Input tokens: {usage.total_input_tokens}")
print(f"Output tokens: {usage.total_output_tokens}")

Analytics

Get analytics data for specific tags:

# Get raw analytics data
analytics = client.get_analytics(
    tags="model1,model2",
    start_time_us=None,  # Optional: defaults to 24 hours ago
    end_time_us=None     # Optional: defaults to current time
)

# Get analytics as a pandas DataFrame
df = client.get_analytics_dataframe(
    tags="model1,model2",
    start_time_us=None,
    end_time_us=None
)

Evaluate Multiple Threads

df = client.create_evaluation_df(thread_ids=["thread1", "thread2"])
print(df.head())

List Available Models

models = client.list_models()
print(models)

🧩 Framework-Specific Examples (Tracing)

Google ADK

from pylangdb.adk import init

# Monkey-patch the client for tracing
init()

# Import your agents after initializing tracing
from google.adk.agents import Agent
from travel_concierge.sub_agents.booking.agent import booking_agent
from travel_concierge.sub_agents.in_trip.agent import in_trip_agent
from travel_concierge.sub_agents.inspiration.agent import inspiration_agent
from travel_concierge.sub_agents.planning.agent import planning_agent
from travel_concierge.sub_agents.post_trip.agent import post_trip_agent
from travel_concierge.sub_agents.pre_trip.agent import pre_trip_agent
from travel_concierge.tools.memory import _load_precreated_itinerary


root_agent = Agent(
    model="openai/gpt-4.1",
    name="root_agent",
    description="A Travel Conceirge using the services of multiple sub-agents",
    instruction="Instruct the travel concierge to plan a trip for the user.",
    sub_agents=[
        inspiration_agent,
        planning_agent,
        booking_agent,
        pre_trip_agent,
        in_trip_agent,
        post_trip_agent,
    ],
    before_agent_callback=_load_precreated_itinerary,
)

OpenAI

import uuid
import os

# Import LangDB tracing
from pylangdb.openai import init

# Initialize tracing
init()

# Import agent components
from agents import (
    Agent,
    Runner,
    set_default_openai_client,
    RunConfig,
    ModelProvider,
    Model,
    OpenAIChatCompletionsModel
)

# Configure OpenAI client with environment variables
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ.get("LANGDB_API_KEY"),
    base_url=os.environ.get("LANGDB_API_BASE_URL"),
    default_headers={
        "x-project-id": os.environ.get("LANGDB_PROJECT_ID")
    }
)
set_default_openai_client(client)

# Create a custom model provider
class CustomModelProvider(ModelProvider):
    def get_model(self, model_name: str | None) -> Model:
        return OpenAIChatCompletionsModel(model=model_name, openai_client=client)

CUSTOM_MODEL_PROVIDER = CustomModelProvider()

agent = Agent(
    name="Math Tutor",
    model="gpt-4.1",
    instruction="You are a math tutor who can help students with their math homework.",
)

group_id = str(uuid.uuid4())
# Use the model provider with a unique group_id for tracing
async def run_agent():
    response = await Runner.run(
        triage_agent,
        input="Hello World",
        run_config=RunConfig(
            model_provider=CUSTOM_MODEL_PROVIDER,  # Inject custom model provider
            group_id=group_id                      # Link all steps to the same trace
        )
    )
    print(response.final_output)

# Run the async function with asyncio
asyncio.run(run_agent())

LangChain

import os
from pylangdb.langchain import init

init()

# Get environment variables for configuration
api_base = os.getenv("LANGDB_API_BASE_URL")
api_key = os.getenv("LANGDB_API_KEY")
if not api_key:
    raise ValueError("Please set the LANGDB_API_KEY environment variable")

project_id = os.getenv("LANGDB_PROJECT_ID")

# Default headers for API requests
default_headers: dict[str, str] = {
    "x-project-id": project-id
}

# Your existing LangChain code works with proper configuration
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage

# Initialize OpenAI LLM with proper configuration
llm = ChatOpenAI(
    model_name="gpt-4",
    temperature=0.3,
    openai_api_base=api_base,
    openai_api_key=api_key,
    default_headers=default_headers,
)
result = llm.invoke([HumanMessage(content="Hello, LangChain!")])

CrewAI

import os
from crewai import Agent, Task, Crew, LLM
from dotenv import load_dotenv

load_dotenv()

# Import and initialize LangDB tracing
from pylangdb.crewai import init

# Initialize tracing before importing or creating any agents
init()

# Initialize API credentials
api_key = os.environ.get("LANGDB_API_KEY")
api_base = os.environ.get("LANGDB_API_BASE_URL")
project_id = os.environ.get("LANGDB_PROJECT_ID")

# Create LLM with proper headers
llm = LLM(
    model="gpt-4",
    api_key=api_key,
    base_url=api_base,
    extra_headers={
        "x-project-id": project_id
    }
)

# Create and use your CrewAI components as usual
# They will be automatically traced by LangDB
researcher = Agent(
    role="researcher",
    goal="Research the topic thoroughly",
    backstory="You are an expert researcher",
    llm=llm,
    verbose=True
)

task = Task(
    description="Research the given topic",
    agent=researcher
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()

Agno

import os
from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools

# Import and initialize LangDB tracing
from pylangdb.agno import init
init()

# Import LangDB model after initializing tracing
from agno.models.langdb import LangDB

# Create agent with LangDB model
agent = Agent(
    name="Web Agent",
    role="Search the web for information",
    model=LangDB(
        id="openai/gpt-4",
        base_url=os.getenv("LANGDB_API_BASE_URL") + '/' + os.getenv("LANGDB_PROJECT_ID") + '/v1',
        api_key=os.getenv("LANGDB_API_KEY"),
        project_id=os.getenv("LANGDB_PROJECT_ID"),
    ),
    tools=[DuckDuckGoTools()],
    instructions="Answer questions using web search",
    show_tool_calls=True,
    markdown=True,
)

# Use the agent
response = agent.run("What is LangDB?")

⚙️ Advanced Configuration

Environment Variables

Variable Description Default
LANGDB_API_KEY Your LangDB API key Required
LANGDB_PROJECT_ID Your LangDB project ID Required
LANGDB_API_BASE_URL LangDB API base URL https://api.us-east-1.langdb.ai
LANGDB_TRACING_BASE_URL Tracing collector endpoint https://api.us-east-1.langdb.ai:4317
LANGDB_TRACING Enable/disable tracing true
LANGDB_TRACING_EXPORTERS Comma-separated list of exporters otlp, console

Custom Configuration

All init() functions accept the same optional parameters:

from langdb.openai import init

init(
    collector_endpoint='https://api.us-east-1.langdb.ai:4317',
    api_key="langdb-api-key",
    project_id="langdb-project-id"
)

🔬 Technical Details

Session and Thread Management

  • Thread ID: Maintains consistent session identifiers across agent calls
  • Run ID: Unique identifier for each execution trace
  • Invocation Tracking: Tracks the sequence of agent invocations
  • State Persistence: Maintains context across callbacks and sub-agent interactions

Distributed Tracing

  • OpenTelemetry Integration: Uses OpenTelemetry for standardized tracing
  • Attribute Propagation: Automatically propagates LangDB-specific attributes
  • Span Correlation: Links related spans across different agents and frameworks
  • Custom Exporters: Supports multiple export formats (OTLP, Console)

API Reference

Initialization Functions

Each framework has a simple init() function that handles all necessary setup:

  • langdb.adk.init(): Patches Google ADK Agent class with LangDB callbacks
  • langdb.openai.init(): Initializes OpenAI agents tracing
  • langdb.langchain.init(): Initializes LangChain tracing
  • langdb.crewai.init(): Initializes CrewAI tracing
  • langdb.agno.init(): Initializes Agno tracing

All init functions accept optional parameters for custom configuration (collector_endpoint, api_key, project_id)

🛟 Troubleshooting

Common Issues

  1. Missing API Key: Ensure LANGDB_API_KEY and LANGDB_PROJECT_ID are set
  2. Tracing Not Working: Check that initialization functions are called before creating agents
  3. Network Issues: Verify collector endpoint is accessible
  4. Framework Conflicts: Initialize LangDB integration before other instrumentation

Debug Mode

Enable console output for debugging:

export LANGDB_TRACING_EXPORTERS="otlp,console"

Disable tracing entirely:

export LANGDB_TRACING="false"

Development

Setting up the environment

  1. Clone the repository
  2. Create a .env file with your credentials:
LANGDB_API_KEY="your_api_key"
LANGDB_PROJECT_ID="your_project_id"

Running Tests

python -m unittest tests/client.py -v

Publishing

poetry build
poetry publish

📋 Requirements

  • Python >= 3.10
  • Framework-specific dependencies (installed automatically)
  • OpenTelemetry libraries (installed automatically)

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🤝 Support

About

No description, website, or topics provided.

Resources

License

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published

Contributors 3

  •  
  •  
  •  

Languages