GPT Research MCP Server

--- title: OpenAI (Python) sidebarTitle: OpenAI (Python) seoTitle: Open Source Observability for OpenAI (Python) description: Drop-in replacement of OpenAI SDK (Python) to get full observability in Langfuse by changing only the import. logo: /images/integrations/openai_icon.svg --- # Observability for OpenAI SDK (Python) <Callout type="info"> Looking for the JS/TS version? [Check it out here](/integrations/model-providers/openai-js). </Callout> If you use the OpenAI Python SDK, you can use the Langfuse **drop-in replacement** to get full logging by changing only the import. This works with OpenAI and Azure OpenAI. ```diff - import openai + from langfuse.openai import openai Alternative imports: + from langfuse.openai import OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI ``` Langfuse automatically tracks: - All prompts/completions with support for streaming, async and functions - Latencies - [API Errors](#error-tracking) - Model usage (tokens) and cost (USD) ([learn more](/docs/model-usage-and-cost)) ## How it works <Steps> ### Install Langfuse SDK The integration is compatible with OpenAI SDK versions `>=0.27.8`. It supports async functions and streaming for OpenAI SDK versions `>=1.0.0`. ```sh pip install langfuse openai ``` ### Switch to Langfuse Wrapped OpenAI SDK <Tabs items={["Environment variables", "Attributes"]}> <Tab> Add Langfuse credentials to your environment variables ```bash filename=".env" LANGFUSE_SECRET_KEY = "sk-lf-..." LANGFUSE_PUBLIC_KEY = "pk-lf-..." LANGFUSE_BASE_URL = "https://cloud.langfuse.com" # 🇪🇺 EU region # LANGFUSE_BASE_URL = "https://us.cloud.langfuse.com" # 🇺🇸 US region ``` Change import ```diff - import openai + from langfuse.openai import openai Alternative imports: + from langfuse.openai import OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI ``` </Tab> <Tab> Change import ```diff - import openai + from langfuse.openai import openai Alternative imports: + from langfuse.openai import OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI ``` Add Langfuse credentials to your code ```python openai.langfuse_public_key = "pk-lf-..." openai.langfuse_secret_key = "sk-lf-..." openai.langfuse_enabled = True # Default is True, set to False to disable Langfuse openai.LANGFUSE_BASE_URL = "https://cloud.langfuse.com" # 🇪🇺 EU region # openai.LANGFUSE_BASE_URL = "https://us.cloud.langfuse.com" # 🇺🇸 US region # Set openai key via attribute openai.api_key = "sk-..." ``` </Tab> </Tabs> Optional, checks the SDK connection with the server. Not recommended for production usage. ```python from langfuse import get_client get_client().auth_check() ``` ### Use OpenAI SDK as usual _No changes required._ Check out the notebook for end-to-end examples of the integration: import { FileCode } from "lucide-react"; <Cards> <Card title="Example notebook" href="/guides/cookbook/integration_openai_sdk" icon={<FileCode />} /> </Cards> </Steps> ## Troubleshooting ### Queuing and batching of events The Langfuse SDKs queue and batches events in the background to reduce the number of network requests and improve overall performance. In a long-running application, this works without any additional configuration. If you are running a short-lived application, you need to flush Langfuse to ensure that all events are flushed before the application exits. ```python from langfuse import get_client from langfuse.openai import openai # Flush via global client langfuse = get_client() langfuse.flush() ``` Learn more about queuing and batching of events [here](/docs/tracing). ### Assistants API Tracing of the assistants api is not supported by this integration as OpenAI Assistants have server-side state that cannot easily be captured without additional api requests. We added some more information on how to best track usage of the assistants api in this [FAQ](/faq/all/openai-assistant-api). ### Debug mode If you are having issues with the integration, you can enable debug mode to get more information about the requests and responses. ```python from langfuse import Langfuse from langfuse.openai import openai # Enable debug via global client langfuse = Langfuse(debug=True) ``` Alternatively, you can set the environment variable: ```sh export LANGFUSE_DEBUG=true ``` ### Sampling [Sampling](/docs/tracing-features/sampling) can be used to control the volume of traces collected by the Langfuse server. ```python from langfuse import Langfuse from langfuse.openai import openai # Set sampling via global client (default is 1.0) langfuse = Langfuse(sample_rate=0.1) ``` Alternatively, you can set the environment variable: ```sh export LANGFUSE_SAMPLE_RATE=0.1 ``` ### Disable tracing You may disable sending traces to Langfuse by setting the appropriate flag. ```python from langfuse import Langfuse from langfuse.openai import openai # Disable via global client langfuse = Langfuse(tracing_enabled=False) ``` Alternatively, you can set the environment variable: ```sh export LANGFUSE_TRACING_ENABLED=false ``` ## Advanced usage ### Custom trace properties You can add the following properties to the openai method: | Property | Description | | ----------------------- | ---------------------------------------------------------------------------- | | `name` | Set `name` to identify a specific type of generation. | | `metadata` | Set `metadata` with additional information that you want to see in Langfuse. | | `trace_id` | See "Interoperability with Langfuse Python SDK" (below) for more details. | | `parent_observation_id` | See "Interoperability with Langfuse Python SDK" (below) for more details. | **Setting trace attributes (`session_id`, `user_id`, `tags`):** You have two options: **Option 1: Via metadata (simplest approach):** ```python from langfuse.openai import openai result = openai.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "You are a very accurate calculator."}, {"role": "user", "content": "1 + 1 = "} ], name="test-chat", metadata={ "langfuse_session_id": "session_123", "langfuse_user_id": "user_456", "langfuse_tags": ["calculator"], "someMetadataKey": "someValue" # Regular metadata still works } ) ``` **Option 2: Via enclosing span (for more control):** ```python from langfuse import get_client from langfuse.openai import openai langfuse = get_client() with langfuse.start_as_current_observation(as_type="span", name="calculator-request") as span: span.update_trace( session_id="session_123", user_id="user_456", tags=["calculator"] ) result = openai.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "You are a very accurate calculator."}, {"role": "user", "content": "1 + 1 = "} ], name="test-chat", metadata={"someMetadataKey": "someValue"}, ) ``` ### Use Traces [Langfuse Tracing](/docs/tracing) groups multiple observations (can be any LLM or non-LLM call) into a single trace. This integration by default creates a single trace for each openai call. - Add non-OpenAI related observations to the trace. - Group multiple OpenAI calls into a single trace while customizing the trace. - Have more control over the trace structure. - Use all Langfuse Tracing features. <Callout type="info"> New to Langfuse Tracing? Checkout this [introduction](/docs/tracing) to the basic concepts. </Callout> You can use any of the following options: 1. [Python `@observe()` decorator](/docs/sdk/python/decorators) - works with both v2 and v3 2. Use explicit span management - differs between v3 and v2 **Option 1: Python Decorator** ```python from langfuse import observe from langfuse.openai import openai @observe() def capital_poem_generator(country): capital = openai.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "What is the capital of the country?"}, {"role": "user", "content": country}], name="get-capital", ).choices[0].message.content poem = openai.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "You are a poet. Create a poem about this city."}, {"role": "user", "content": capital}], name="generate-poem", ).choices[0].message.content return poem capital_poem_generator("Bulgaria") ``` **Option 2: Context Managers** ```python from langfuse import get_client, propagate_attributes from langfuse.openai import openai langfuse = get_client() with langfuse.start_as_current_observation(as_type="span", name="capital-poem-generator") as span: # Propagate trace attributes to all child observations with propagate_attributes( user_id="user_123", session_id="session_456", tags=["poetry", "capital"] ): capital = openai.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "What is the capital of the country?"}, {"role": "user", "content": "Bulgaria"}], name="get-capital", ).choices[0].message.content poem = openai.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "You are a poet. Create a poem about this city."}, {"role": "user", "content": capital}], name="generate-poem", ).choices[0].message.content ``` ### OpenAI token usage on streamed responses OpenAI returns the token usage on streamed responses only when in `stream_options` the `include_usage` parameter is set to `True`. If you would like to benefit from OpenAI's directly provided token usage, you can set `{"include_usage": True} in the `stream_options` argument. <Callout type="info"> When using streaming responses with `include_usage=True`, OpenAI returns token usage information in a final chunk that has an empty `choices` list. Make sure your application properly handles these empty `choices` chunks to ensure accurate token usage tracking by not trying to access some index in the `choices` list without checking if it is non-empty. </Callout> ```python /stream_options={"include_usage": True}/ from langfuse import get_client from langfuse.openai import openai client = openai.OpenAI() stream = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "How are you?"}], stream=True, stream_options={"include_usage": True}, ) result = "" for chunk in stream: # Check if chunk choices are not empty. OpenAI returns token usage in a final chunk with an empty choices list. if chunk.choices: result += chunk.choices[0].delta.content or "" # Flush via global client get_client().flush() ``` ### OpenAI Beta APIs Since OpenAI beta APIs are changing frequently across versions, we fully support only the stable APIs in the OpenAI SDK. If you are using a beta API, you can still use the Langfuse SDK by wrapping the OpenAI SDK manually with the `@observe()` [decorator](/docs/sdk/python/decorators). #### Structured Output For **structured output parsing**, please use the `response_format` argument to `openai.chat.completions.create()` instead of the Beta API. This will allow you to set Langfuse attributes and metadata. If you rely on parsing Pydantic defintions for your `response_format`, you may leverage the `type_to_response_format_param` utility function from the OpenAI Python SDK to convert the Pydantic definition to a `response_format` dictionary. This is the same function the OpenAI Beta API uses to convert Pydantic definitions to `response_format` dictionaries. ```python from langfuse import get_client from langfuse.openai import openai from openai.lib._parsing._completions import type_to_response_format_param from pydantic import BaseModel class CalendarEvent(BaseModel): name: str date: str participants: list[str] completion = openai.chat.completions.create( model="gpt-4o-2024-08-06", messages=[ {"role": "system", "content": "Extract the event information."}, { "role": "user", "content": "Alice and Bob are going to a science fair on Friday.", }, ], response_format=type_to_response_format_param(CalendarEvent), ) print(completion) # Flush via global client get_client().flush() ``` #### Assistants API Tracing of the assistants api is not supported by this integration as OpenAI Assistants have server-side state that cannot easily be captured without additional api requests. Check out this [notebook](/integrations/model-providers/openai-assistants-api) for an end-to-end example on how to best track usage of the assistants api in Langfuse. ## Tracking of OpenAI API Errors [#error-tracking] Langfuse automatically tracks and monitors OpenAI API errors if you use the native integration. They are captured via the `level` and `statusMessage` fields (see [docs](/docs/tracing-features/log-levels)). Learn more about how to get started [here](/integrations/model-providers/openai-py). ```diff - import openai + from langfuse.openai import openai ``` ```python # Cause an error by attempting to use a host that does not exist. openai.base_url = "https://example.com" country = openai.chat.completions.create( name="will-error", model="gpt-3.5-turbo", messages=[ {"role": "user", "content": "How are you?"}], ) ``` Throws error 👆  ## FAQ import { FaqPreview } from "@/components/faq/FaqPreview"; <FaqPreview tags={["integration-openai"]} /> ## GitHub Discussions import { GhDiscussionsPreview } from "@/components/gh-discussions/GhDiscussionsPreview"; <GhDiscussionsPreview labels={["integration-openai"]} />