Skip to content

Instantly share code, notes, and snippets.

@iddogino
Last active January 20, 2026 22:10
Show Gist options
  • Select an option

  • Save iddogino/ccddf21638e607a1b5b00ec557489999 to your computer and use it in GitHub Desktop.

Select an option

Save iddogino/ccddf21638e607a1b5b00ec557489999 to your computer and use it in GitHub Desktop.
LangSmith CSV Trace Export
#!/usr/bin/env python3
"""
LangSmith Trace Exporter
========================
Export LangSmith traces to CSV format with comprehensive metadata.
This script exports LLM call data from a LangSmith project to a CSV file with
21 columns including: messages (OpenAI format), tokens, costs, latency, and more.
REQUIRED ENVIRONMENT VARIABLES:
LANGSMITH_API_KEY Your LangSmith API key (required)
Get it from: https://smith.langchain.com/settings
OPTIONAL ENVIRONMENT VARIABLES:
LANGSMITH_ENDPOINT LangSmith API endpoint
Default: https://api.smith.langchain.com
USAGE:
python export_traces_to_csv.py
Or with uv:
uv run --with langsmith export_traces_to_csv.py
OUTPUT:
Creates langsmith_traces.csv with 21 columns per LLM call:
- Core: trace_id, run_id, parent_run_id, input_messages, output_message
- Model: model, provider, temperature, finish_reason
- Status: status, error
- Tokens: total_tokens, prompt_tokens, completion_tokens
- Cost: total_cost, prompt_cost, completion_cost
- Performance: latency_ms, start_time, end_time
- Metadata: tags
"""
import os
import sys
import json
import csv
from datetime import datetime
try:
from langsmith import Client
except ImportError:
print("Error: langsmith package not found.")
print("Install it with: pip install langsmith")
sys.exit(1)
# Configuration
PROJECT_NAME = "Demo Project - Traces" # Change this to your LangSmith project name
OUTPUT_FILE = "langsmith_traces.csv"
# Validate environment variables
if not os.environ.get("LANGSMITH_API_KEY"):
print("Error: LANGSMITH_API_KEY environment variable is required.")
print("Get your API key from: https://smith.langchain.com/settings")
sys.exit(1)
# Initialize LangSmith client
client = Client(
api_key=os.environ.get("LANGSMITH_API_KEY"),
api_url=os.environ.get("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com")
)
def parse_langchain_message(msg):
"""
Parse a LangChain serialized message to OpenAI format.
Converts LangChain's internal message format to standard OpenAI format:
{"role": "user|assistant|system", "content": "...", "tool_calls": [...]}
"""
role_map = {
"human": "user",
"ai": "assistant",
"system": "system",
"HumanMessage": "user",
"AIMessage": "assistant",
"SystemMessage": "system"
}
# Handle serialized LangChain format (with 'kwargs' and 'id')
if isinstance(msg, dict):
if "kwargs" in msg:
kwargs = msg["kwargs"]
msg_type = kwargs.get("type", "")
# Check if it's in the 'id' field
if "id" in msg and isinstance(msg["id"], list):
msg_class = msg["id"][-1] if msg["id"] else ""
role = role_map.get(msg_class, "assistant")
else:
role = role_map.get(msg_type, msg_type)
content = kwargs.get("content", "")
# Check for tool calls
tool_calls = kwargs.get("tool_calls", [])
if tool_calls:
return {
"role": role,
"content": content,
"tool_calls": tool_calls
}
return {
"role": role,
"content": content
}
# Handle simple dict format
elif "type" in msg:
role = role_map.get(msg.get("type", ""), msg.get("type", ""))
content = msg.get("content", "")
tool_calls = msg.get("tool_calls", [])
if tool_calls:
return {
"role": role,
"content": content,
"tool_calls": tool_calls
}
return {
"role": role,
"content": content
}
return {"role": "user", "content": str(msg)}
def format_messages_to_oai(messages):
"""Convert messages to OpenAI format."""
if not messages:
return "[]"
# If messages is a list of lists (batch format), take the first batch
if isinstance(messages, list) and messages and isinstance(messages[0], list):
messages = messages[0]
formatted = []
for msg in messages:
parsed_msg = parse_langchain_message(msg)
formatted.append(parsed_msg)
return json.dumps(formatted, ensure_ascii=False)
def format_output_message(output):
"""Format output message including tool calls."""
if not output:
return ""
# Handle LangSmith LLM output format with generations
if isinstance(output, dict) and "generations" in output:
generations = output["generations"]
if generations and len(generations) > 0 and len(generations[0]) > 0:
gen = generations[0][0]
if "message" in gen:
message = gen["message"]
# Parse the LangChain message format
parsed = parse_langchain_message(message)
return json.dumps(parsed, ensure_ascii=False)
# Handle direct message format
if isinstance(output, dict):
parsed = parse_langchain_message(output)
return json.dumps(parsed, ensure_ascii=False)
return str(output)
def extract_llm_calls_from_run(run, trace_id):
"""Extract LLM calls from a run and its children."""
llm_calls = []
# Check if this run is an LLM call
if run.run_type == "llm":
# Get input messages
input_messages = []
if run.inputs:
if "messages" in run.inputs:
input_messages = run.inputs["messages"]
elif "prompts" in run.inputs:
# Handle prompt format
input_messages = [{"role": "user", "content": run.inputs["prompts"][0]}]
# Get output message and finish reason
output_message = ""
finish_reason = ""
if run.outputs:
if "generations" in run.outputs:
# Standard LLM output format
generations = run.outputs["generations"]
if generations and len(generations) > 0 and len(generations[0]) > 0:
gen = generations[0][0]
if "message" in gen:
output_message = gen["message"]
elif "text" in gen:
output_message = {"role": "assistant", "content": gen["text"]}
# Get finish reason
if "generation_info" in gen:
finish_reason = gen["generation_info"].get("finish_reason", "")
elif "content" in run.outputs:
output_message = run.outputs
# Calculate latency
latency_ms = ""
if run.start_time and run.end_time:
latency_ms = int((run.end_time - run.start_time).total_seconds() * 1000)
# Get metadata
metadata = run.extra.get("metadata", {}) if run.extra else {}
invocation_params = run.extra.get("invocation_params", {}) if run.extra else {}
llm_calls.append({
"trace_id": trace_id,
"run_id": str(run.id),
"parent_run_id": str(run.parent_run_id) if run.parent_run_id else "",
"input_messages": format_messages_to_oai(input_messages),
"output_message": format_output_message(output_message),
"model": metadata.get("ls_model_name", ""),
"provider": metadata.get("ls_provider", ""),
"temperature": metadata.get("ls_temperature", ""),
"finish_reason": finish_reason,
"status": run.status or "",
"error": run.error or "",
"total_tokens": run.total_tokens or 0,
"prompt_tokens": run.prompt_tokens or 0,
"completion_tokens": run.completion_tokens or 0,
"total_cost": run.total_cost or 0,
"prompt_cost": run.prompt_cost or 0,
"completion_cost": run.completion_cost or 0,
"latency_ms": latency_ms,
"tags": json.dumps(run.tags) if run.tags else "[]",
"start_time": run.start_time.isoformat() if run.start_time else "",
"end_time": run.end_time.isoformat() if run.end_time else ""
})
return llm_calls
def export_traces_to_csv(output_file=OUTPUT_FILE, project_name=PROJECT_NAME):
"""Export all traces from the project to CSV with progress indicators."""
print(f"\n{'='*70}")
print(f" LangSmith Trace Exporter")
print(f"{'='*70}")
print(f"\nProject: {project_name}")
print(f"Output: {output_file}")
# Step 1: Fetch runs
print(f"\n[1/3] Fetching runs from LangSmith...", end="", flush=True)
try:
runs = list(client.list_runs(project_name=project_name))
print(f" ✓ ({len(runs)} runs)")
except Exception as e:
print(f" ✗")
print(f"Error: Failed to fetch runs from project '{project_name}'")
print(f"Details: {e}")
print(f"\nMake sure:")
print(f" - The project name is correct")
print(f" - Your API key has access to this project")
print(f" - You can view the project at: https://smith.langchain.com/")
sys.exit(1)
if not runs:
print(f"\nNo runs found in project '{project_name}'")
print(f"Please check the project name or create some traces first.")
sys.exit(1)
# Step 2: Group by trace and extract LLM calls
print(f"\n[2/3] Processing traces...")
all_llm_calls = []
# Group runs by trace_id
traces = {}
for run in runs:
trace_id = str(run.trace_id) if run.trace_id else str(run.id)
if trace_id not in traces:
traces[trace_id] = []
traces[trace_id].append(run)
total_traces = len(traces)
print(f" Found {total_traces} unique traces")
# Process each trace with progress indicator
for i, (trace_id, trace_runs) in enumerate(traces.items(), 1):
# Progress bar
progress = int(50 * i / total_traces)
bar = '█' * progress + '░' * (50 - progress)
percent = 100 * i / total_traces
print(f"\r [{bar}] {percent:5.1f}% ({i}/{total_traces})", end="", flush=True)
# Extract LLM calls from all runs in this trace
for run in trace_runs:
llm_calls = extract_llm_calls_from_run(run, trace_id)
all_llm_calls.extend(llm_calls)
print(f"\r [{'█'*50}] 100.0% ({total_traces}/{total_traces}) ✓")
print(f" Extracted {len(all_llm_calls)} LLM calls")
# Step 3: Write to CSV
if not all_llm_calls:
print(f"\n⚠ No LLM calls found in the traces")
print(f" The project may only contain tool/chain runs without LLM calls.")
sys.exit(1)
print(f"\n[3/3] Writing to CSV...", end="", flush=True)
fieldnames = [
"trace_id", "run_id", "parent_run_id",
"input_messages", "output_message",
"model", "provider", "temperature", "finish_reason",
"status", "error",
"total_tokens", "prompt_tokens", "completion_tokens",
"total_cost", "prompt_cost", "completion_cost",
"latency_ms", "tags",
"start_time", "end_time"
]
try:
with open(output_file, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(all_llm_calls)
print(f" ✓")
except Exception as e:
print(f" ✗")
print(f"Error: Failed to write CSV file")
print(f"Details: {e}")
sys.exit(1)
# Calculate statistics
total_tokens = sum(call.get("total_tokens", 0) for call in all_llm_calls)
total_cost = sum(call.get("total_cost", 0) for call in all_llm_calls)
latencies = [call.get("latency_ms", 0) for call in all_llm_calls if call.get("latency_ms")]
avg_latency = sum(latencies) / len(latencies) if latencies else 0
# Count finish reasons
finish_reasons = {}
for call in all_llm_calls:
reason = call.get("finish_reason", "unknown")
finish_reasons[reason] = finish_reasons.get(reason, 0) + 1
# Print summary
print(f"\n{'='*70}")
print(f" Export Complete!")
print(f"{'='*70}")
print(f"\n📊 Summary:")
print(f" • File: {output_file}")
print(f" • Traces: {len(traces)}")
print(f" • LLM calls: {len(all_llm_calls)}")
print(f"\n💰 Usage & Cost:")
print(f" • Total tokens: {total_tokens:,}")
print(f" • Total cost: ${total_cost:.6f}")
print(f"\n⚡ Performance:")
print(f" • Avg latency: {avg_latency:.0f}ms")
if finish_reasons:
print(f"\n🎯 Finish Reasons:")
for reason, count in sorted(finish_reasons.items(), key=lambda x: -x[1]):
percent = 100 * count / len(all_llm_calls)
print(f" • {reason:15s} {count:4d} ({percent:5.1f}%)")
print(f"\n✓ Data exported successfully!")
print(f" View the CSV file: {output_file}")
print()
return output_file
def main():
"""Main entry point for the script."""
try:
export_traces_to_csv()
except KeyboardInterrupt:
print("\n\n⚠ Export cancelled by user")
sys.exit(130)
except Exception as e:
print(f"\n\n✗ Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment