Last active
January 20, 2026 22:10
-
-
Save iddogino/ccddf21638e607a1b5b00ec557489999 to your computer and use it in GitHub Desktop.
LangSmith CSV Trace Export
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/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