diff --git a/python/tracing/langchain/00_quickstart.py b/python/tracing/langchain/00_quickstart.py index afbf250..7f57632 100644 --- a/python/tracing/langchain/00_quickstart.py +++ b/python/tracing/langchain/00_quickstart.py @@ -9,48 +9,21 @@ from __future__ import annotations -import os - -from dotenv import find_dotenv, load_dotenv +from _shared import init_telemetry, tracing_config from langchain_core.language_models.fake_chat_models import FakeListChatModel from langchain_core.messages import HumanMessage, SystemMessage -from respan_instrumentation_langchain import add_respan_callback -from respan_tracing import RespanTelemetry - -load_dotenv(find_dotenv(), override=False) def langchain_instrumentation_quickstart() -> None: - api_key = os.getenv("RESPAN_API_KEY") - telemetry: RespanTelemetry | None = None - - if api_key: - telemetry = RespanTelemetry( - app_name="langchain-quickstart", - api_key=api_key, - base_url=os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api"), - is_auto_instrument=False, - is_batching_enabled=False, - is_enabled=True, - ) - else: - print("RESPAN_API_KEY is not set; running locally without exporting spans.") + init_telemetry("langchain-quickstart") model = FakeListChatModel(responses=["Hello from a traced LangChain run."]) - config = { - "run_name": "hello_world", - "tags": ["respan-langchain-example", "quickstart"], - "metadata": {"example": "quickstart"}, - } - if telemetry: - config = add_respan_callback(config) - response = model.invoke( [ SystemMessage(content="Reply in one short sentence."), HumanMessage(content="Say hello to Respan tracing."), ], - config=config, + config=tracing_config("hello_world"), ) print(response.content) diff --git a/python/tracing/langchain/06_chat_model_astream.py b/python/tracing/langchain/06_chat_model_astream.py index bcfb900..9547352 100644 --- a/python/tracing/langchain/06_chat_model_astream.py +++ b/python/tracing/langchain/06_chat_model_astream.py @@ -2,14 +2,13 @@ import asyncio -from langchain_core.language_models.fake_chat_models import FakeChatModel - from _shared import init_telemetry, message_text, tracing_config +from langchain_core.language_models.fake_chat_models import FakeListChatModel async def chat_model_astream() -> None: - telemetry = init_telemetry("langchain-chat-model-astream") - model = FakeChatModel() + init_telemetry("langchain-chat-model-astream") + model = FakeListChatModel(responses=["Asynchronous streaming chat output."]) chunks = [] async for chunk in model.astream( "Stream asynchronously.", @@ -17,5 +16,7 @@ async def chat_model_astream() -> None: ): chunks.append(message_text(chunk)) print("".join(chunks)) + + if __name__ == "__main__": asyncio.run(chat_model_astream()) diff --git a/python/tracing/langchain/09_chat_model_astream_events.py b/python/tracing/langchain/09_chat_model_astream_events.py index 0e9f390..74968bb 100644 --- a/python/tracing/langchain/09_chat_model_astream_events.py +++ b/python/tracing/langchain/09_chat_model_astream_events.py @@ -2,14 +2,13 @@ import asyncio -from langchain_core.language_models.fake_chat_models import FakeChatModel - from _shared import init_telemetry, tracing_config +from langchain_core.language_models.fake_chat_models import FakeListChatModel async def chat_model_astream_events() -> None: - telemetry = init_telemetry("langchain-chat-model-astream-events") - model = FakeChatModel() + init_telemetry("langchain-chat-model-astream-events") + model = FakeListChatModel(responses=["Semantic stream events."]) events = [] async for event in model.astream_events( "Emit semantic stream events.", @@ -18,5 +17,7 @@ async def chat_model_astream_events() -> None: ): events.append(event["event"]) print(events) + + if __name__ == "__main__": asyncio.run(chat_model_astream_events()) diff --git a/python/tracing/langchain/11_llm_stream.py b/python/tracing/langchain/11_llm_stream.py index 73608af..3a35390 100644 --- a/python/tracing/langchain/11_llm_stream.py +++ b/python/tracing/langchain/11_llm_stream.py @@ -1,13 +1,49 @@ """Legacy string LLM stream.""" -from langchain_core.language_models.fake import FakeStreamingListLLM +from collections.abc import Iterator +from typing import Any from _shared import init_telemetry, tracing_config +from langchain_core.callbacks import CallbackManagerForLLMRun +from langchain_core.language_models.llms import LLM +from langchain_core.outputs import GenerationChunk + + +class CallbackStreamingLLM(LLM): + """Deterministic LLM that exercises LangChain's token callback contract.""" + + response: str + + @property + def _llm_type(self) -> str: + return "callback-streaming-list" + + def _call( + self, + prompt: str, + stop: list[str] | None = None, + run_manager: CallbackManagerForLLMRun | None = None, + **kwargs: Any, + ) -> str: + return self.response + + def _stream( + self, + prompt: str, + stop: list[str] | None = None, + run_manager: CallbackManagerForLLMRun | None = None, + **kwargs: Any, + ) -> Iterator[GenerationChunk]: + for token in self.response.splitlines(keepends=True): + chunk = GenerationChunk(text=token) + if run_manager is not None: + run_manager.on_llm_new_token(token, chunk=chunk) + yield chunk def llm_stream() -> None: - telemetry = init_telemetry("langchain-llm-stream") - llm = FakeStreamingListLLM(responses=["tokenized completion"]) + init_telemetry("langchain-llm-stream") + llm = CallbackStreamingLLM(response="tokenized completion\nwith callbacks") chunks = list( llm.stream( "Stream this completion.", @@ -15,5 +51,7 @@ def llm_stream() -> None: ) ) print("".join(chunks)) + + if __name__ == "__main__": llm_stream() diff --git a/python/tracing/langchain/README.md b/python/tracing/langchain/README.md index 5d2ee62..e7c2033 100644 --- a/python/tracing/langchain/README.md +++ b/python/tracing/langchain/README.md @@ -41,6 +41,8 @@ Run one example: python 00_quickstart.py ``` +Run the complete bounded set with `python run_all_examples.py`. + ## Examples | Script | LangChain function or behavior | diff --git a/python/tracing/langchain/_shared.py b/python/tracing/langchain/_shared.py index 732c348..aaa54d8 100644 --- a/python/tracing/langchain/_shared.py +++ b/python/tracing/langchain/_shared.py @@ -2,21 +2,41 @@ from __future__ import annotations +import atexit import os +from pathlib import Path from typing import Any -from dotenv import find_dotenv, load_dotenv +from dotenv import load_dotenv from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel from langchain_core.messages import AIMessage from langchain_core.tools import tool from respan_instrumentation_langchain import add_respan_callback from respan_tracing import RespanTelemetry -load_dotenv(find_dotenv(), override=False) +ROOT_DIR = Path(__file__).resolve().parents[3] +load_dotenv(ROOT_DIR / ".env", override=True) + +RUN_ID = os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or "langchain-local" +_ACTIVE_TELEMETRY: list[RespanTelemetry] = [] class NoopTelemetry: - pass + def flush(self) -> None: + return None + + +def _flush_telemetry() -> None: + """Flush every example telemetry instance on normal and exceptional exits.""" + while _ACTIVE_TELEMETRY: + telemetry = _ACTIVE_TELEMETRY.pop() + try: + telemetry.flush() + except Exception: # noqa: BLE001,S110 - process-exit flush is best-effort + pass + + +atexit.register(_flush_telemetry) def init_telemetry(app_name: str) -> RespanTelemetry | NoopTelemetry: @@ -25,7 +45,7 @@ def init_telemetry(app_name: str) -> RespanTelemetry | NoopTelemetry: if not api_key: return NoopTelemetry() - return RespanTelemetry( + telemetry = RespanTelemetry( app_name=app_name, api_key=api_key, base_url=os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api"), @@ -33,6 +53,8 @@ def init_telemetry(app_name: str) -> RespanTelemetry | NoopTelemetry: is_batching_enabled=False, is_enabled=True, ) + _ACTIVE_TELEMETRY.append(telemetry) + return telemetry def tracing_config(name: str, metadata: dict[str, Any] | None = None) -> dict[str, Any]: @@ -40,7 +62,19 @@ def tracing_config(name: str, metadata: dict[str, Any] | None = None) -> dict[st { "run_name": name, "tags": ["respan-langchain-example", name], - "metadata": {"example": name, **(metadata or {})}, + "metadata": { + "example": name, + **(metadata or {}), + "respan_params": { + "trace_group_identifier": f"langchain_{name}.workflow", + "custom_identifier": f"{RUN_ID}:{name}", + "metadata": { + "example": "langchain", + "example_run_id": RUN_ID, + "workflow_name": f"langchain_{name}.workflow", + }, + }, + }, } ) @@ -66,7 +100,7 @@ def bind_tools( self, tools: Any, **kwargs: Any, - ) -> "ToolCallingFakeMessagesListChatModel": + ) -> ToolCallingFakeMessagesListChatModel: return self @@ -107,7 +141,12 @@ def make_openai_chat_model(model_name: str = "gpt-4o-mini") -> Any | None: "api_key": api_key, "temperature": 0, } - base_url = os.getenv("OPENAI_BASE_URL") or os.getenv("RESPAN_OPENAI_BASE_URL") + base_url = ( + os.getenv("OPENAI_BASE_URL") + or os.getenv("RESPAN_OPENAI_BASE_URL") + or os.getenv("RESPAN_GATEWAY_BASE_URL") + or os.getenv("RESPAN_BASE_URL") + ) if base_url: kwargs["base_url"] = base_url return ChatOpenAI(**kwargs) diff --git a/python/tracing/langchain/langchain_agent.py b/python/tracing/langchain/langchain_agent.py deleted file mode 100644 index 8b8687b..0000000 --- a/python/tracing/langchain/langchain_agent.py +++ /dev/null @@ -1,228 +0,0 @@ -""" -LangGraph + LangChain Agent Example with Respan Tracing - -This example demonstrates how to use LangGraph with LangChain tools and agents, -all traced automatically by Respan. - -Shows: -- LangGraph's agent workflow -- Custom LangChain tools -- Automatic Respan tracing - -Prerequisites: -1. Install dependencies: poetry install -2. Set up environment variables in .env: - - RESPAN_API_KEY=your_api_key (required for tracing) - - OPENAI_API_KEY=your_openai_key (required for LLM calls) - - RESPAN_BASE_URL=https://api.respan.ai/api (optional) - -Run: - python langchain_agent.py -""" - -from dotenv import load_dotenv - -load_dotenv(override=True) - -import os -from typing import Annotated, TypedDict, Literal -from langchain_openai import ChatOpenAI -from langchain_core.tools import tool -from langchain_core.messages import HumanMessage, AIMessage -from langgraph.graph import StateGraph, START, END, MessagesState -from langgraph.prebuilt import ToolNode -from respan_tracing import RespanTelemetry, workflow -from respan_tracing.instruments import Instruments - - -# Initialize Respan tracing for LangGraph + LangChain -telemetry = RespanTelemetry( - app_name="langgraph-agent-example", - api_key=os.getenv("RESPAN_API_KEY"), - base_url=os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api"), - instruments={Instruments.LANGCHAIN, Instruments.OPENAI}, -) - -print("āœ“ LangGraph + LangChain Agent tracing enabled via RespanTelemetry\n") - - -# Define custom tools for the agent -@tool -def get_weather(city: str) -> str: - """Get the current weather for a city.""" - # In a real app, this would call a weather API - return f"The weather in {city} is sunny with a temperature of 72Ā°F." - - -@tool -def calculate(expression: str) -> str: - """Calculate a mathematical expression. Input should be a valid Python expression.""" - try: - result = eval(expression) - return f"The result is: {result}" - except Exception as e: - return f"Error calculating: {str(e)}" - - -@tool -def search_wiki(query: str) -> str: - """Search for information. Use this for general knowledge questions.""" - # In a real app, this would search Wikipedia or another knowledge base - return f"Here's what I found about '{query}': [Simulated search result - in production, this would return actual data]" - - -# Create the tools list -tools = [get_weather, calculate, search_wiki] - - -# Initialize the LLM (using Respan proxy for tracing) -llm = ChatOpenAI( - model="gpt-4o-mini", - openai_api_key=os.getenv("OPENAI_API_KEY"), - openai_api_base=os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api/chat/completions"), - default_headers={ - "Authorization": f"Bearer {os.getenv('RESPAN_API_KEY')}", - }, - temperature=0, -) - -# Bind tools to the LLM -llm_with_tools = llm.bind_tools(tools) - - -# Define the agent graph using LangGraph -def should_continue(state: MessagesState) -> Literal["tools", "__end__"]: - """Decide whether to continue with tools or end""" - messages = state["messages"] - last_message = messages[-1] - - # If there are tool calls, continue to tools node - if hasattr(last_message, "tool_calls") and last_message.tool_calls: - return "tools" - # Otherwise, end - return "__end__" - - -def call_model(state: MessagesState): - """Call the LLM with the current state""" - messages = state["messages"] - response = llm_with_tools.invoke(messages) - return {"messages": [response]} - - -# Build the agent graph with LangGraph -graph_builder = StateGraph(MessagesState) - -# Add nodes -graph_builder.add_node("agent", call_model) -graph_builder.add_node("tools", ToolNode(tools)) - -# Add edges -graph_builder.add_edge(START, "agent") -graph_builder.add_conditional_edges( - "agent", - should_continue, - { - "tools": "tools", - "__end__": END, - } -) -graph_builder.add_edge("tools", "agent") - -# Compile the graph -agent_graph = graph_builder.compile() - - -@workflow(name="weather_query") -def weather_query(): - """Simple weather query workflow using LangGraph""" - print("=== Weather Query Example ===\n") - result = agent_graph.invoke({ - "messages": [HumanMessage(content="What's the weather like in San Francisco?")] - }) - final_message = result["messages"][-1] - print(f"\nFinal Answer: {final_message.content}\n") - return result - - -@workflow(name="calculation_query") -def calculation_query(): - """Math calculation workflow using LangGraph""" - print("=== Calculation Example ===\n") - result = agent_graph.invoke({ - "messages": [HumanMessage(content="What is 15 multiplied by 23, plus 100?")] - }) - final_message = result["messages"][-1] - print(f"\nFinal Answer: {final_message.content}\n") - return result - - -@workflow(name="multi_tool_query") -def multi_tool_query(): - """Query that requires multiple tools using LangGraph""" - print("=== Multi-Tool Example ===\n") - result = agent_graph.invoke({ - "messages": [HumanMessage(content="What's the weather in Tokyo, and then calculate 25 * 4?")] - }) - final_message = result["messages"][-1] - print(f"\nFinal Answer: {final_message.content}\n") - return result - - -@workflow(name="interactive_agent") -def interactive_agent(): - """Interactive agent chat using LangGraph""" - print("=== Interactive Agent (type 'quit' to exit) ===\n") - - conversation_history = [] - - while True: - try: - user_input = input("You: ") - if user_input.lower() in ["quit", "exit", "q"]: - print("Goodbye!") - break - - # Add user message to history - conversation_history.append(HumanMessage(content=user_input)) - - # Invoke the agent graph - result = agent_graph.invoke({"messages": conversation_history}) - - # Update conversation history with all messages from the result - conversation_history = result["messages"] - - # Print the final response - final_message = result["messages"][-1] - print(f"\nAgent: {final_message.content}\n") - - except KeyboardInterrupt: - print("\nGoodbye!") - break - except Exception as e: - print(f"Error: {e}\n") - - -if __name__ == "__main__": - # Run examples - print("Running LangGraph + LangChain Agent examples with Respan tracing...\n") - print("=" * 60) - - # Example 1: Weather query - weather_query() - - print("=" * 60) - - # Example 2: Calculation - calculation_query() - - print("=" * 60) - - # Example 3: Multi-tool query - multi_tool_query() - - print("=" * 60) - - # Example 4: Interactive mode - interactive_agent() - diff --git a/python/tracing/langchain/run_all_examples.py b/python/tracing/langchain/run_all_examples.py new file mode 100644 index 0000000..03607e7 --- /dev/null +++ b/python/tracing/langchain/run_all_examples.py @@ -0,0 +1,55 @@ +"""Run every bounded LangChain example in a fresh process.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +EXAMPLES = tuple( + f"{index:02d}_{name}.py" + for index, name in enumerate( + ( + "quickstart", + "chat_model_invoke", + "chat_model_stream", + "chat_model_batch", + "chat_model_batch_as_completed", + "chat_model_ainvoke", + "chat_model_astream", + "chat_model_abatch", + "chat_model_abatch_as_completed", + "chat_model_astream_events", + "llm_invoke", + "llm_stream", + "model_bind_tools", + "model_with_structured_output", + "tool_invoke", + "tool_ainvoke", + "prompt_chain_invoke", + "runnable_parallel_invoke", + "retriever_invoke", + "agent_invoke", + "agent_stream_updates", + "agent_stream_messages", + "agent_stream_custom", + "agent_structured_output", + "custom_event", + "runnable_with_retry", + "chain_error", + "tool_error", + "retriever_error", + ) + ) +) + + +def main() -> None: + base_dir = Path(__file__).resolve().parent + for script_name in EXAMPLES: + print(f"\n=== {script_name} ===", flush=True) + subprocess.run([sys.executable, str(base_dir / script_name)], check=True) + + +if __name__ == "__main__": + main() diff --git a/python/tracing/langfuse/README.md b/python/tracing/langfuse/README.md index 3daab06..1f193e7 100644 --- a/python/tracing/langfuse/README.md +++ b/python/tracing/langfuse/README.md @@ -1,179 +1,20 @@ -# Langfuse to Respan Integration +# Langfuse tracing with Respan -This example demonstrates how to use the [Langfuse Python SDK](https://python.reference.langfuse.com/) to send traces directly to Respan +This bounded example uses the current Langfuse Python SDK and the linked +`respan-instrumentation-langfuse` package. It creates two deterministic traces: -## Overview +- `langfuse_simple.workflow`: workflow to generation +- `langfuse_research.workflow`: workflow to two tools and one generation -The Langfuse SDK provides a convenient API for tracing LLM applications. By simply pointing the SDK to Respan's API endpoint, you can use all of Langfuse's tracing features while sending data to Respan. +The generation spans include model, prompt/completion content, and exact token +usage. Dummy Langfuse credentials are used only to enable local SDK span +creation; the Respan instrumentor intercepts Langfuse's exporter before any +request is sent to Langfuse. -## Installation +Install the requirements and the local package, then run: -This project uses Poetry for dependency management. Make sure you have Poetry installed: + RESPAN_EXAMPLE_RUN_ID=your-marker python langfuse_simple_example.py -```bash -# Install Poetry if you haven't already -curl -sSL https://install.python-poetry.org | python3 - - -# Navigate to the python scripts directory -cd example_scripts/python - -# Install dependencies -poetry install -``` - -## Configuration - -1. Copy `.env.example` to `.env` in the langfuse directory: - ```bash - cp langfuse/.env.example langfuse/.env - ``` - -2. Add your Respan API key to `langfuse/.env`: - ```env - RESPAN_API_KEY=your_respan_api_key_here - RESPAN_BASE_URL=https://api.respan.ai/api - - # Optional: Set Langfuse credentials if you want to use actual Langfuse - LANGFUSE_PUBLIC_KEY= - LANGFUSE_SECRET_KEY= - LANGFUSE_BASE_URL= - ``` - -3. Get your API key from [Respan Platform](https://platform.respan.ai/platform/api/api-keys) - -## Usage - -### Basic Integration with Decorators - -The example uses Langfuse's `@observe()` decorator for automatic tracing: - -```python -import os -from langfuse import observe, get_client - -# Set environment variables -os.environ["LANGFUSE_PUBLIC_KEY"] = os.getenv("LANGFUSE_PUBLIC_KEY", "") -os.environ["LANGFUSE_SECRET_KEY"] = os.getenv("LANGFUSE_SECRET_KEY", "") -os.environ["LANGFUSE_BASE_URL"] = os.getenv("RESPAN_BASE_URL", "") - -langfuse = get_client() - -# Use @observe decorator to automatically trace functions -@observe(as_type="generation") -def chat_completion(user_message: str): - response = f"Response to: {user_message}" - return response - -# Function calls are automatically traced -result = chat_completion("Hello!") - -# Flush to send data -langfuse.flush() -``` - -## Running the Example - -```bash -# Run the example script -poetry run python langfuse/langfuse_simple_example.py -``` - -This will: -1. Create two example traces demonstrating different patterns -2. Example 1: Simple trace with LLM generation -3. Example 2: Deep research workflow with multi-level nested spans -4. Send all traces to Respan - -## Key Features - -### Automatic Tracing with Decorators -Use `@observe()` to automatically trace function calls: -```python -@observe() -def my_function(input_data): - # Function inputs and outputs are automatically captured - result = process_data(input_data) - return result -``` - -### LLM Generations -Mark functions as generations to track LLM calls: -```python -@observe(as_type="generation") -def chat_completion(user_message: str, model: str = "gpt-4o-mini"): - # Automatically captures input, output, and model info - response = call_llm(user_message, model) - return response -``` - -### Nested Spans -Create deep trace trees by calling decorated functions: -```python -@observe() -def parent_function(): - # This creates a parent span - result1 = child_function_1() # Creates child span - result2 = child_function_2() # Creates another child span - return combine_results(result1, result2) - -@observe() -def child_function_1(): - return "result 1" - -@observe() -def child_function_2(): - return "result 2" -``` - -### Multi-Level Workflows -The example demonstrates a deep research workflow with: -- 4 levels of nesting -- 3 parallel branches (Wikipedia, ArXiv, Google Scholar) -- 13 total spans showing complex trace trees - -## Documentation - -- [Langfuse Python SDK Reference](https://python.reference.langfuse.com/) -- [Langfuse Low-Level SDK Guide](https://langfuse.com/docs/sdk/python/low-level-sdk) -- [Respan Documentation](https://docs.respan.ai/) - -## How It Works - -This integration uses a monkey-patched OpenTelemetry exporter to: -1. Intercept traces created by Langfuse's `@observe()` decorators -2. Transform OpenTelemetry span format to Respan's log format -3. Send traces to `https://api.respan.ai/api/v1/traces/ingest` - -### Key Points -- Uses Langfuse's decorator-based API (`@observe()`) -- Automatically captures function inputs and outputs -- Creates nested span trees for complex workflows -- No need for Langfuse credentials - uses your Respan API key -- All traces appear in your Respan dashboard at https://platform.respan.ai/ - -## Expected Output - -When you run the example, you'll see: -``` -šŸš€ Initializing Langfuse with Respan base_url... - -============================================================ -Example 1: Simple Trace with LLM Generation -============================================================ -šŸ“ Created trace: simple-chat -šŸ¤– Output: Response to: Hello, how are you? - -============================================================ -Example 2: Deep Research Workflow (Multi-Level Tree) -============================================================ -šŸš€ Starting deep research workflow for: 'What is quantum computing?' - šŸ“š Gathering research from multiple sources... - šŸ” Searching Wikipedia... - šŸ“„ Extracting info from Wikipedia... - āœ“ Validating Wikipedia... - ... -āœ… Research complete! - -āœ… All traces flushed! -šŸ“Š Check your Respan dashboard -``` +The script loads the repository root `.env`, never prints credentials, flushes +both SDKs in `finally`, and fails unless all six expected Langfuse spans pass +through the local instrumentation package. diff --git a/python/tracing/langfuse/langfuse_simple_example.py b/python/tracing/langfuse/langfuse_simple_example.py index c269913..a270385 100644 --- a/python/tracing/langfuse/langfuse_simple_example.py +++ b/python/tracing/langfuse/langfuse_simple_example.py @@ -1,266 +1,117 @@ -"""Langfuse to Respan Integration using @observe decorators.""" +"""Exercise current Langfuse SDK spans through the linked Respan instrumentor.""" + +from __future__ import annotations import os from pathlib import Path -from langfuse import observe, Langfuse -from dotenv import load_dotenv - -env_path = Path(__file__).parent / '.env' -load_dotenv(dotenv_path=env_path) - -import opentelemetry.exporter.otlp.proto.http.trace_exporter as otlp_module -import requests -import json -from datetime import datetime, timezone - -original_export = otlp_module.OTLPSpanExporter.export - -def patched_export(self, spans): - """Transform and export spans to Respan.""" - respan_endpoint = "https://api.respan.ai/api/v1/traces/ingest" - api_key = os.getenv("RESPAN_API_KEY") - - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json", - "Accept": "application/json" - } - - batch_logs = [] - - for span in spans: - attributes = dict(span.attributes) if span.attributes else {} - - langfuse_type = attributes.get("langfuse.observation.type", "span") - log_type_mapping = { - "span": "workflow" if not span.parent else "tool", - "generation": "generation" - } - log_type = log_type_mapping.get(langfuse_type, "custom") - - start_time_ns = span.start_time - end_time_ns = span.end_time - start_time_iso = datetime.fromtimestamp(start_time_ns / 1e9, tz=timezone.utc).isoformat() - timestamp_iso = datetime.fromtimestamp(end_time_ns / 1e9, tz=timezone.utc).isoformat() - latency = (end_time_ns - start_time_ns) / 1e9 - - payload = { - "trace_unique_id": format(span.context.trace_id, '032x'), - "span_unique_id": format(span.context.span_id, '016x'), - "span_parent_id": format(span.parent.span_id, '016x') if span.parent else None, - "span_name": span.name, - "span_workflow_name": attributes.get("langfuse.trace.name", span.name), - "log_type": log_type, - "customer_identifier": attributes.get("user.id"), - "timestamp": timestamp_iso, - "start_time": start_time_iso, - "latency": latency, - } - - if "langfuse.observation.input" in attributes: - input_str = attributes["langfuse.observation.input"] - payload["input"] = input_str if isinstance(input_str, str) else json.dumps(input_str) - - if "langfuse.observation.output" in attributes: - output_str = attributes["langfuse.observation.output"] - payload["output"] = output_str if isinstance(output_str, str) else json.dumps(output_str) - - if "langfuse.observation.model.name" in attributes: - payload["model"] = attributes["langfuse.observation.model.name"] - - if "langfuse.observation.usage_details" in attributes: - try: - usage = json.loads(attributes["langfuse.observation.usage_details"]) - payload["usage"] = { - "prompt_tokens": usage.get("prompt_tokens", 0), - "completion_tokens": usage.get("completion_tokens", 0), - "total_tokens": usage.get("total_tokens", 0) - } - except: - pass - - batch_logs.append(payload) - - if batch_logs: - try: - response = requests.post(respan_endpoint, json=batch_logs, headers=headers, timeout=10) - response.raise_for_status() - except Exception as e: - print(f"Warning: Failed to send batch to Respan. Error: {e}") - - from opentelemetry.sdk.trace.export import SpanExportResult - return SpanExportResult.SUCCESS - -otlp_module.OTLPSpanExporter.export = patched_export - -respan_api_key = os.getenv("RESPAN_API_KEY") -respan_base_url = os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api") - -if not respan_api_key: - raise ValueError("RESPAN_API_KEY environment variable is required") - -langfuse_public_key = os.getenv("LANGFUSE_PUBLIC_KEY", "") -langfuse_secret_key = os.getenv("LANGFUSE_SECRET_KEY", "") - -langfuse = Langfuse( - public_key=langfuse_public_key, - secret_key=langfuse_secret_key, - base_url=respan_base_url -) - - -@observe(as_type="generation") -def chat_completion(user_message: str, model: str = "gpt-4o-mini"): - """Simulate chat completion.""" - response = f"Response to: {user_message}" - return response - - -@observe() -def search_web(query: str, source: str): - """Search web from source.""" - print(f" šŸ” Searching {source}...") - return { - "source": source, - "results": f"Search results for '{query}' from {source}" - } - - -@observe() -def extract_information(search_result: dict): - """Extract information from results.""" - source = search_result["source"] - print(f" šŸ“„ Extracting info from {source}...") - return { - "source": source, - "extracted_info": f"Key facts from {source}", - "relevance_score": 0.85 - } - - -@observe() -def validate_source(extracted_info: dict): - """Validate source reliability.""" - source = extracted_info["source"] - print(f" āœ“ Validating {source}...") - return { - "source": source, - "is_valid": True, - "confidence": 0.9 - } - - -@observe() -def research_topic(topic: str, source: str): - """Research topic from source.""" - search_result = search_web(topic, source) - extracted = extract_information(search_result) - validated = validate_source(extracted) - return validated - - -@observe() -def gather_research(query: str): - """Gather research from multiple sources.""" - print(f" šŸ“š Gathering research from multiple sources...") - sources = ["Wikipedia", "ArXiv", "Google Scholar"] - results = [] - - for source in sources: - result = research_topic(query, source) - results.append(result) - - return results +from dotenv import load_dotenv -@observe(as_type="generation") -def synthesize_answer(query: str, research_results: list): - """Synthesize answer from research.""" - print(f" šŸ§  Synthesizing answer...") - valid_sources = [r["source"] for r in research_results if r["is_valid"]] - answer = f"Based on research from {', '.join(valid_sources)}, here's the answer about {query}..." +ROOT_DIR = Path(__file__).resolve().parents[3] +load_dotenv(ROOT_DIR / ".env", override=True) + +os.environ.setdefault("LANGFUSE_PUBLIC_KEY", "pk-lf-respan-example") +os.environ.setdefault("LANGFUSE_SECRET_KEY", "sk-lf-respan-example") +os.environ.setdefault("LANGFUSE_BASE_URL", "https://cloud.langfuse.com") + +from langfuse import get_client, observe +from respan import Respan +from respan_instrumentation_langfuse import LangfuseInstrumentor + +RUN_ID = os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() or "langfuse-local" +MODEL = os.getenv("RESPAN_LANGFUSE_MODEL", "gpt-4o-mini") +EXPECTED_SPANS = 6 + + +def _mark_trace(workflow_name: str, input_value: object) -> None: + get_client().update_current_trace( + name=workflow_name, + user_id="langfuse-example-user", + session_id=f"{RUN_ID}:session", + input=input_value, + metadata={ + "example": "langfuse", + "example_run_id": RUN_ID, + "workflow_name": workflow_name, + }, + ) + + +@observe(as_type="generation", name="answer-question") +def answer_question(question: str) -> str: + answer = f"A concise answer about: {question}" + get_client().update_current_generation( + input=[{"role": "user", "content": question}], + output=[{"role": "assistant", "content": answer}], + model=MODEL, + usage_details={ + "prompt_tokens": 9, + "completion_tokens": 7, + "total_tokens": 16, + }, + metadata={"example_run_id": RUN_ID}, + ) return answer -@observe(as_type="generation") -def evaluate_answer(query: str, answer: str): - """Evaluate answer quality.""" - print(f" āš–ļø Evaluating answer quality...") - return { - "quality_score": 0.92, - "completeness": 0.88, - "accuracy": 0.95, - "feedback": "High quality answer with good coverage" - } - - -@observe() -def multi_step_workflow(query: str): - """Multi-level research workflow.""" - print(f"\nšŸš€ Starting deep research workflow for: '{query}'\n") - - research_results = gather_research(query) - answer = synthesize_answer(query, research_results) - evaluation = evaluate_answer(query, answer) - - print(f"\nāœ… Research complete!") - print(f" Quality Score: {evaluation['quality_score']}") - - return { - "answer": answer, - "evaluation": evaluation, - "sources_used": len(research_results) - } - - -@observe() -def simple_chat_example(): - """Simple chat example.""" - result = chat_completion("Hello, how are you?") - print(f"\nšŸ“ Created trace: simple-chat") - print(f"šŸ¤– Output: {result}\n") +@observe(as_type="tool", name="search-source") +def search_source(source: str, query: str) -> dict[str, str]: + result = {"source": source, "result": f"{source} evidence for {query}"} + get_client().update_current_span( + output=result, + metadata={"example_run_id": RUN_ID}, + ) return result -def main(): - """Run Langfuse to Respan integration examples.""" - print("šŸš€ Initializing Langfuse with Respan base_url...\n") - print(f"āœ… Langfuse initialized with base_url: {respan_base_url}") - print(f"šŸ”‘ Using API Key: {respan_api_key[:10]}...\n") - - print("=" * 60) - print("Example 1: Simple Trace with LLM Generation") - print("=" * 60) - simple_chat_example() - - print("=" * 60) - print("Example 2: Deep Research Workflow (Multi-Level Tree)") - print("=" * 60) - print("\nThis creates a deep trace tree with:") - print(" - 3 parallel research branches (Wikipedia, ArXiv, Google Scholar)") - print(" - Each branch has 3 levels: Search ā†’ Extract ā†’ Validate") - print(" - Plus synthesis and evaluation steps") - print(" - Total: ~13 spans across 4 levels\n") - - result2 = multi_step_workflow("What is quantum computing?") - print(f"\nšŸ“ Created deep trace: multi-step-workflow") - print(f"šŸ¤– Answer preview: {result2['answer'][:80]}...") - print(f" Sources: {result2['sources_used']}, Quality: {result2['evaluation']['quality_score']}\n") - - print("=" * 60) - print("Flushing traces to Respan...") - print("=" * 60) - langfuse.flush() - - print("\nāœ… All traces flushed!") - print("\nšŸ“Š Check your Respan dashboard:") - print(" https://platform.respan.ai/") +@observe(name="langfuse_simple.workflow") +def simple_workflow() -> str: + _mark_trace("langfuse_simple.workflow", {"question": "What is tracing?"}) + return answer_question("What is tracing?") + + +@observe(name="langfuse_research.workflow") +def research_workflow() -> str: + query = "OpenTelemetry" + _mark_trace("langfuse_research.workflow", {"query": query}) + evidence = [ + search_source("docs", query), + search_source("examples", query), + ] + return answer_question( + f"Summarize {query} using {', '.join(item['source'] for item in evidence)}" + ) + + +def main() -> None: + api_key = os.environ["RESPAN_API_KEY"] + respan = Respan( + api_key=api_key, + base_url=os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api"), + app_name="langfuse-current-sdk", + instrumentations=[], + is_batching_enabled=False, + ) + instrumentor = LangfuseInstrumentor() + instrumentor.instrument() + client = get_client() + + try: + print(simple_workflow()) + print(research_workflow()) + client.flush() + respan.flush() + if instrumentor.exported_span_count != EXPECTED_SPANS: + raise RuntimeError( + "Langfuse exported " + f"{instrumentor.exported_span_count} spans; expected {EXPECTED_SPANS}" + ) + print(f"Langfuse exported {EXPECTED_SPANS} canonical spans.") + finally: + client.flush() + instrumentor.uninstrument() + respan.shutdown() if __name__ == "__main__": - try: - main() - except Exception as e: - print(f"āŒ Error: {e}") - import traceback - traceback.print_exc() + main() diff --git a/python/tracing/langfuse/requirements.txt b/python/tracing/langfuse/requirements.txt index 14e267f..cc72bef 100644 --- a/python/tracing/langfuse/requirements.txt +++ b/python/tracing/langfuse/requirements.txt @@ -1,3 +1,4 @@ -langfuse>=2.0.0 -requests>=2.31.0 +langfuse>=3.12.0 +respan-ai +respan-instrumentation-langfuse python-dotenv>=1.0.0 diff --git a/python/tracing/litellm/01_basic_completion.py b/python/tracing/litellm/01_basic_completion.py index 4f9ccf5..390d067 100644 --- a/python/tracing/litellm/01_basic_completion.py +++ b/python/tracing/litellm/01_basic_completion.py @@ -1,6 +1,4 @@ import litellm -from respan import workflow - from _shared import ( GATEWAY_API_KEY, GATEWAY_BASE_URL, @@ -8,6 +6,7 @@ create_respan, run_with_example_attributes, ) +from respan import workflow WORKFLOW_NAME = "litellm_basic_completion.workflow" @@ -32,11 +31,16 @@ def litellm_basic_completion() -> str: def main() -> None: respan = create_respan("litellm-basic-completion") - output = run_with_example_attributes( - respan, - workflow_name=WORKFLOW_NAME, - action=litellm_basic_completion, - ) - print(output) + try: + output = run_with_example_attributes( + respan, + workflow_name=WORKFLOW_NAME, + action=litellm_basic_completion, + ) + print(output) + finally: + respan.shutdown() + + if __name__ == "__main__": main() diff --git a/python/tracing/litellm/02_streaming_completion.py b/python/tracing/litellm/02_streaming_completion.py index f7440fb..da497bc 100644 --- a/python/tracing/litellm/02_streaming_completion.py +++ b/python/tracing/litellm/02_streaming_completion.py @@ -1,6 +1,4 @@ import litellm -from respan import workflow - from _shared import ( GATEWAY_API_KEY, GATEWAY_BASE_URL, @@ -8,6 +6,7 @@ create_respan, run_with_example_attributes, ) +from respan import workflow WORKFLOW_NAME = "litellm_streaming_completion.workflow" @@ -44,11 +43,16 @@ def litellm_streaming_completion() -> str: def main() -> None: respan = create_respan("litellm-streaming-completion") - output = run_with_example_attributes( - respan, - workflow_name=WORKFLOW_NAME, - action=litellm_streaming_completion, - ) - print(output) + try: + output = run_with_example_attributes( + respan, + workflow_name=WORKFLOW_NAME, + action=litellm_streaming_completion, + ) + print(output) + finally: + respan.shutdown() + + if __name__ == "__main__": main() diff --git a/python/tracing/litellm/03_respan_attributes.py b/python/tracing/litellm/03_respan_attributes.py index 3bd4671..4860c27 100644 --- a/python/tracing/litellm/03_respan_attributes.py +++ b/python/tracing/litellm/03_respan_attributes.py @@ -1,7 +1,4 @@ import litellm -from respan import propagate_attributes -from respan import workflow - from _shared import ( GATEWAY_API_KEY, GATEWAY_BASE_URL, @@ -9,6 +6,7 @@ create_respan, run_with_example_attributes, ) +from respan import propagate_attributes, workflow WORKFLOW_NAME = "litellm_respan_attributes.workflow" @@ -38,11 +36,16 @@ def litellm_respan_attributes() -> str: def main() -> None: respan = create_respan("litellm-respan-attributes") - output = run_with_example_attributes( - respan, - workflow_name=WORKFLOW_NAME, - action=litellm_respan_attributes, - ) - print(output) + try: + output = run_with_example_attributes( + respan, + workflow_name=WORKFLOW_NAME, + action=litellm_respan_attributes, + ) + print(output) + finally: + respan.shutdown() + + if __name__ == "__main__": main() diff --git a/python/tracing/litellm/04_tool_calling.py b/python/tracing/litellm/04_tool_calling.py new file mode 100644 index 0000000..6051de5 --- /dev/null +++ b/python/tracing/litellm/04_tool_calling.py @@ -0,0 +1,65 @@ +import json + +import litellm +from _shared import ( + GATEWAY_API_KEY, + GATEWAY_BASE_URL, + MODEL, + create_respan, + run_with_example_attributes, +) +from respan import workflow + +WORKFLOW_NAME = "litellm_tool_calling.workflow" +TOOLS = [ + { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get deterministic weather for a city.", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + }, + } +] + + +@workflow(name=WORKFLOW_NAME) +def litellm_tool_calling() -> list[dict[str, object]]: + response = litellm.completion( + api_key=GATEWAY_API_KEY, + api_base=GATEWAY_BASE_URL, + model=MODEL, + messages=[{"role": "user", "content": "What is the weather in Paris?"}], + tools=TOOLS, + tool_choice={"type": "function", "function": {"name": "get_weather"}}, + temperature=0, + max_tokens=80, + ) + calls = response.choices[0].message.tool_calls or [] + if not calls: + raise RuntimeError("The provider returned no tool call.") + return [ + call.model_dump() if hasattr(call, "model_dump") else dict(call) + for call in calls + ] + + +def main() -> None: + respan = create_respan("litellm-tool-calling") + try: + output = run_with_example_attributes( + respan, + workflow_name=WORKFLOW_NAME, + action=litellm_tool_calling, + ) + print(json.dumps(output, default=str)) + finally: + respan.shutdown() + + +if __name__ == "__main__": + main() diff --git a/python/tracing/litellm/05_expected_error.py b/python/tracing/litellm/05_expected_error.py new file mode 100644 index 0000000..3cca1f3 --- /dev/null +++ b/python/tracing/litellm/05_expected_error.py @@ -0,0 +1,43 @@ +import litellm +from _shared import ( + GATEWAY_BASE_URL, + create_respan, + run_with_example_attributes, +) +from respan import workflow + +WORKFLOW_NAME = "litellm_expected_error.workflow" + + +@workflow(name=WORKFLOW_NAME) +def litellm_expected_error() -> str: + try: + litellm.completion( + api_key="respan-example-intentionally-invalid", + api_base=GATEWAY_BASE_URL, + model="openai/gpt-4o-mini", + messages=[{"role": "user", "content": "This request should fail."}], + max_tokens=20, + ) + except litellm.APIError as exc: + return f"caught expected {type(exc).__name__}" + raise RuntimeError( + "The controlled invalid-credential request unexpectedly succeeded." + ) + + +def main() -> None: + respan = create_respan("litellm-expected-error") + try: + output = run_with_example_attributes( + respan, + workflow_name=WORKFLOW_NAME, + action=litellm_expected_error, + ) + print(output) + finally: + respan.shutdown() + + +if __name__ == "__main__": + main() diff --git a/python/tracing/litellm/06_async_completion.py b/python/tracing/litellm/06_async_completion.py new file mode 100644 index 0000000..a7b8739 --- /dev/null +++ b/python/tracing/litellm/06_async_completion.py @@ -0,0 +1,46 @@ +import asyncio + +import litellm +from _shared import ( + GATEWAY_API_KEY, + GATEWAY_BASE_URL, + MODEL, + create_respan, + run_async_with_example_attributes, +) +from respan import workflow + +WORKFLOW_NAME = "litellm_async_completion.workflow" + + +@workflow(name=WORKFLOW_NAME) +async def litellm_async_completion() -> str: + response = await litellm.acompletion( + api_key=GATEWAY_API_KEY, + api_base=GATEWAY_BASE_URL, + model=MODEL, + messages=[{"role": "user", "content": "Reply with: async LiteLLM works."}], + temperature=0, + max_tokens=30, + ) + return response.choices[0].message.content + + +def main() -> None: + respan = create_respan("litellm-async-completion") + try: + print( + asyncio.run( + run_async_with_example_attributes( + respan, + workflow_name=WORKFLOW_NAME, + action=litellm_async_completion, + ) + ) + ) + finally: + respan.shutdown() + + +if __name__ == "__main__": + main() diff --git a/python/tracing/litellm/07_async_streaming.py b/python/tracing/litellm/07_async_streaming.py new file mode 100644 index 0000000..ea61dca --- /dev/null +++ b/python/tracing/litellm/07_async_streaming.py @@ -0,0 +1,55 @@ +import asyncio + +import litellm +from _shared import ( + GATEWAY_API_KEY, + GATEWAY_BASE_URL, + MODEL, + create_respan, + run_async_with_example_attributes, +) +from respan import workflow + +WORKFLOW_NAME = "litellm_async_streaming.workflow" + + +@workflow(name=WORKFLOW_NAME) +async def litellm_async_streaming() -> str: + stream = await litellm.acompletion( + api_key=GATEWAY_API_KEY, + api_base=GATEWAY_BASE_URL, + model=MODEL, + messages=[{"role": "user", "content": "Reply with: async stream works."}], + stream=True, + stream_options={"include_usage": True}, + temperature=0, + max_tokens=30, + ) + chunks = [] + async for chunk in stream: + choices = getattr(chunk, "choices", None) or [] + if choices: + content = getattr(getattr(choices[0], "delta", None), "content", None) + if content: + chunks.append(content) + return "".join(chunks).strip() + + +def main() -> None: + respan = create_respan("litellm-async-streaming") + try: + print( + asyncio.run( + run_async_with_example_attributes( + respan, + workflow_name=WORKFLOW_NAME, + action=litellm_async_streaming, + ) + ) + ) + finally: + respan.shutdown() + + +if __name__ == "__main__": + main() diff --git a/python/tracing/litellm/README.md b/python/tracing/litellm/README.md index 0b6648c..5fbc9a8 100644 --- a/python/tracing/litellm/README.md +++ b/python/tracing/litellm/README.md @@ -26,6 +26,10 @@ The scripts use the root `.env` file. Required values: python python/tracing/litellm/01_basic_completion.py python python/tracing/litellm/02_streaming_completion.py python python/tracing/litellm/03_respan_attributes.py +python python/tracing/litellm/04_tool_calling.py +python python/tracing/litellm/05_expected_error.py +python python/tracing/litellm/06_async_completion.py +python python/tracing/litellm/07_async_streaming.py ``` Or run all examples: @@ -41,3 +45,11 @@ The examples are searchable in Respan by these workflow names: - `litellm_basic_completion.workflow` - `litellm_streaming_completion.workflow` - `litellm_respan_attributes.workflow` +- `litellm_tool_calling.workflow` +- `litellm_expected_error.workflow` +- `litellm_async_completion.workflow` +- `litellm_async_streaming.workflow` + +Every script shuts down Respan in `finally`. The set covers sync/async, +stream/non-stream, canonical tool schemas/calls, propagated attributes, and a +controlled expected provider error. diff --git a/python/tracing/litellm/_shared.py b/python/tracing/litellm/_shared.py index c517fb9..15e8dba 100644 --- a/python/tracing/litellm/_shared.py +++ b/python/tracing/litellm/_shared.py @@ -1,7 +1,8 @@ import os import time +from collections.abc import Awaitable, Callable from pathlib import Path -from typing import Callable, TypeVar +from typing import TypeVar from dotenv import load_dotenv from respan import Respan @@ -22,7 +23,7 @@ def _example_run_id() -> str: configured = os.getenv("RESPAN_EXAMPLE_RUN_ID", "").strip() - if configured and "".join(("co", "dex")) not in configured.lower(): + if configured and "codex" not in configured.lower(): return configured return f"litellm-{int(time.time())}" @@ -58,3 +59,21 @@ def run_with_example_attributes( }, ): return action() + + +async def run_async_with_example_attributes( + respan: Respan, + *, + workflow_name: str, + action: Callable[[], Awaitable[T]], +) -> T: + with respan.propagate_attributes( + trace_group_identifier=workflow_name, + custom_identifier=f"{RUN_ID}:{workflow_name}", + metadata={ + "example": "litellm", + "example_run_id": RUN_ID, + "workflow_name": workflow_name, + }, + ): + return await action() diff --git a/python/tracing/litellm/run_all_examples.py b/python/tracing/litellm/run_all_examples.py index 620fef0..ace70f2 100644 --- a/python/tracing/litellm/run_all_examples.py +++ b/python/tracing/litellm/run_all_examples.py @@ -6,6 +6,10 @@ "01_basic_completion.py", "02_streaming_completion.py", "03_respan_attributes.py", + "04_tool_calling.py", + "05_expected_error.py", + "06_async_completion.py", + "07_async_streaming.py", )