langfuse-core-workflow-a

# Langfuse Core Workflow A: Tracing LLM Calls ## Overview Primary workflow for Langfuse: end-to-end tracing of LLM calls, chains, and agents. ## Prerequisites - Completed `langfuse-install-auth` setup - Understanding of Langfuse trace hierarchy - LLM provider SDK (OpenAI, Anthropic, etc.) ## Instructions ### Step 1: Set Up Automatic OpenAI Tracing ```typescript import { observeOpenAI } from "langfuse"; import OpenAI from "openai"; // Wrap OpenAI client for automatic tracing const openai = observeOpenAI(new OpenAI()); // All calls are now automatically traced const response = await openai.chat.completions.create({ model: "gpt-4", messages: [ { role: "system", content: "You are a helpful assistant." }, { role: "user", content: "What is Langfuse?" }, ], }); ``` ### Step 2: Manual Tracing for Complex Flows ```typescript import { Langfuse } from "langfuse"; const langfuse = new Langfuse(); async function ragPipeline(query: string) { // Create top-level trace const trace = langfuse.trace({ name: "rag-pipeline", input: { query }, metadata: { pipeline: "rag-v1" }, }); // Span: Query embedding const embedSpan = trace.span({ name: "embed-query", input: { text: query }, }); const queryEmbedding = await embedText(query); embedSpan.end({ output: { dimensions: queryEmbedding.length }, metadata: { model: "text-embedding-ada-002" }, }); // Span: Vector search const searchSpan = trace.span({ name: "vector-search", input: { embedding: "vector[1536]" }, }); const documents = await searchVectorDB(queryEmbedding); searchSpan.end({ output: { documentCount: documents.length, topScore: documents[0]?.score, }, }); // Generation: LLM call with context const generation = trace.generation({ name: "generate-answer", model: "gpt-4", modelParameters: { temperature: 0.7, maxTokens: 500 }, input: { query, context: documents.map((d) => d.content), }, }); const answer = await generateAnswer(query, documents); generation.end({ output: answer.content, usage: { promptTokens: answer.usage.prompt_tokens, completionTokens: answer.usage.completion_tokens, totalTokens: answer.usage.total_tokens, }, }); // Update trace with final output trace.update({ output: { answer: answer.content }, }); return answer.content; } ``` ### Step 3: Trace Streaming Responses ```typescript async function streamingChat(messages: ChatMessage[]) { const trace = langfuse.trace({ name: "streaming-chat", input: { messages }, }); const generation = trace.generation({ name: "stream-response", model: "gpt-4", input: messages, }); const stream = await openai.chat.completions.create({ model: "gpt-4", messages, stream: true, }); let fullContent = ""; let usage = { promptTokens: 0, completionTokens: 0 }; for await (const chunk of stream) { const content = chunk.choices[0]?.delta?.content || ""; fullContent += content; // Track usage from final chunk if (chunk.usage) { usage = { promptTokens: chunk.usage.prompt_tokens, completionTokens: chunk.usage.completion_tokens, }; } yield content; } // End generation after stream completes generation.end({ output: fullContent, usage, }); trace.update({ output: { response: fullContent } }); } ``` ### Step 4: Link Traces to Parent Operations ```typescript // API endpoint that creates parent trace async function handleChatRequest(req: Request) { const trace = langfuse.trace({ name: "api/chat", userId: req.userId, sessionId: req.sessionId, input: req.body, }); // Pass trace to child operations const response = await processChat(trace, req.body.message); return response; } // Child function receives parent trace async function processChat( parentTrace: ReturnType, message: string ) { // Create child span under parent trace const span = parentTrace.span({ name: "process-chat", input: { message }, }); // OpenAI call linked to parent const response = await openai.chat.completions.create( { model: "gpt-4", messages: [{ role: "user", content: message }], }, { langfuseParent: span } // Link generation to span ); span.end({ output: response.choices[0].message }); return response.choices[0].message.content; } ``` ## Output - Automatic OpenAI tracing with zero code changes - Manual tracing for RAG and complex pipelines - Streaming response tracking - Linked parent-child trace hierarchy ## Error Handling | Issue | Cause | Solution | |-------|-------|----------| | Missing generations | Wrapper not applied | Use `observeOpenAI()` wrapper | | Orphaned spans | Missing `.end()` call | Always end spans in finally block | | No token usage | Stream without usage | Use `stream_options: {include_usage: true}` | | Broken hierarchy | Missing parent link | Pass `langfuseParent` option | ## Examples ### Anthropic Claude Tracing ```typescript import Anthropic from "@anthropic-ai/sdk"; import { Langfuse } from "langfuse"; const anthropic = new Anthropic(); const langfuse = new Langfuse(); async function callClaude(prompt: string) { const trace = langfuse.trace({ name: "claude-call", input: { prompt } }); const generation = trace.generation({ name: "claude-response", model: "claude-3-sonnet-20240229", input: [{ role: "user", content: prompt }], }); const response = await anthropic.messages.create({ model: "claude-3-sonnet-20240229", max_tokens: 1024, messages: [{ role: "user", content: prompt }], }); generation.end({ output: response.content[0].text, usage: { promptTokens: response.usage.input_tokens, completionTokens: response.usage.output_tokens, }, }); return response.content[0].text; } ``` ### LangChain Integration ```python from langchain.callbacks import LangfuseCallbackHandler from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage # Create callback handler langfuse_handler = LangfuseCallbackHandler() # Pass to LLM llm = ChatOpenAI(callbacks=[langfuse_handler]) # All LangChain operations are now traced response = llm([HumanMessage(content="Hello!")]) ``` ### Tool/Function Calling Tracing ```typescript async function tracedToolCall(trace: Trace, toolName: string, args: any) { const span = trace.span({ name: `tool/${toolName}`, input: args, metadata: { type: "tool-call" }, }); try { const result = await executeTool(toolName, args); span.end({ output: result }); return result; } catch (error) { span.end({ level: "ERROR", statusMessage: String(error), }); throw error; } } ``` ## Resources - [Langfuse Tracing Guide](https://langfuse.com/docs/tracing) - [OpenAI Integration](https://langfuse.com/docs/integrations/openai) - [LangChain Integration](https://langfuse.com/docs/integrations/langchain) - [Streaming Guide](https://langfuse.com/docs/tracing-features/sessions) ## Next Steps For evaluation and scoring workflows, see `langfuse-core-workflow-b`.