Cookbook：OpenAI 集成（Python）

%pip install langfuse openai --upgrade

import os
 
# Get keys for your project from the project settings page: https://litefuse.cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..." 
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..." 
os.environ["LANGFUSE_BASE_URL"] = "https://litefuse.cloud"
 
# Your openai key
os.environ["OPENAI_API_KEY"] = "sk-proj-..."

# instead of: import openai
from langfuse.openai import openai

completion = openai.chat.completions.create(
  name="test-chat",
  model="gpt-4o",
  messages=[
      {"role": "system", "content": "You are a very accurate calculator. You output only the result of the calculation."},
      {"role": "user", "content": "1 + 1 = "}],
  temperature=0,
  metadata={"someMetadataKey": "someValue"},
)

completion = openai.chat.completions.create(
  name="test-url-image",
  model="gpt-4o-mini", # GPT-4o, GPT-4o mini, and GPT-4 Turbo have vision capabilities
  messages=[
      {"role": "system", "content": "You are an AI trained to describe and interpret images. Describe the main objects and actions in the image."},
      {"role": "user", "content": [
        {"type": "text", "text": "What’s depicted in this image?"},
        {
          "type": "image_url",
          "image_url": {
            "url": "https://static.langfuse.com/langfuse-dev/langfuse-example-image.jpeg",
          },
        },
      ],
    }
  ],
  temperature=0,
  metadata={"someMetadataKey": "someValue"},
)

completion = openai.chat.completions.create(
  name="test-chat",
  model="gpt-4o",
  messages=[
      {"role": "system", "content": "You are a professional comedian."},
      {"role": "user", "content": "Tell me a joke."}],
  temperature=0,
  metadata={"someMetadataKey": "someValue"},
  stream=True
)
 
for chunk in completion:
  print(chunk.choices[0].delta.content, end="")

from langfuse.openai import AsyncOpenAI
 
async_client = AsyncOpenAI()

completion = await async_client.chat.completions.create(
  name="test-chat",
  model="gpt-4o",
  messages=[
      {"role": "system", "content": "You are a very accurate calculator. You output only the result of the calculation."},
      {"role": "user", "content": "1 + 100 = "}],
  temperature=0,
  metadata={"someMetadataKey": "someValue"},
)

%pip install pydantic --upgrade

from typing import List
from pydantic import BaseModel
 
class StepByStepAIResponse(BaseModel):
    title: str
    steps: List[str]
schema = StepByStepAIResponse.model_json_schema() # returns a dict like JSON schema

from openai import OpenAI
import json
 
client = OpenAI()
 
response = client.chat.completions.create(
    model="gpt-5",
    messages=[
        {"role": "user", "content": "Explain how to assemble a PC"}
    ],
    tools=[
        {
            "type": "function",
            "function": {
                "name": "get_answer_for_user_query",
                "description": "Get user answer in a series of steps",
                "parameters": StepByStepAIResponse.model_json_schema()
            }
        }
    ],
    tool_choice={"type": "function", "function": {"name": "get_answer_for_user_query"}}
)
 
tool_call = response.choices[0].message.tool_calls[0]
args = json.loads(tool_call.function.arguments)

from langfuse.openai import openai
 
completion = openai.chat.completions.create(
  name="test-chat",
  model="gpt-3.5-turbo",
  messages=[
    {"role": "system", "content": "You are a calculator."},
    {"role": "user", "content": "1 + 1 = "}],
  temperature=0,
  metadata={
    "langfuse_tags": ["tag-1", "tag-2"],
    "langfuse_user_id": "user-123",
    "langfuse_session_id": "session-123",
    "langfuse_metadata": {"key": "value"}
  }
)

AZURE_OPENAI_KEY=""
AZURE_ENDPOINT=""
AZURE_DEPLOYMENT_NAME="cookbook-gpt-4o-mini" # example deployment name

# instead of: from openai import AzureOpenAI
from langfuse.openai import AzureOpenAI

client = AzureOpenAI(
    api_key=AZURE_OPENAI_KEY,  
    api_version="2023-03-15-preview",
    azure_endpoint=AZURE_ENDPOINT
)

client.chat.completions.create(
  name="test-chat-azure-openai",
  model=AZURE_DEPLOYMENT_NAME, # deployment name
  messages=[
      {"role": "system", "content": "You are a very accurate calculator. You output only the result of the calculation."},
      {"role": "user", "content": "1 + 1 = "}],
  temperature=0,
)

from langfuse.openai import openai
from langfuse import observe
 
@observe() # decorator to automatically create trace and nest generations
def main(country: str, user_id: str, **kwargs) -> str:
    # nested generation 1: use openai to get capital of country
    capital = openai.chat.completions.create(
      name="geography-teacher",
      model="gpt-4o",
      messages=[
          {"role": "system", "content": "You are a Geography teacher helping students learn the capitals of countries. Output only the capital when being asked."},
          {"role": "user", "content": country}],
      temperature=0,
    ).choices[0].message.content
 
    # nested generation 2: use openai to write poem on capital
    poem = openai.chat.completions.create(
      name="poet",
      model="gpt-4o",
      messages=[
          {"role": "system", "content": "You are a poet. Create a poem about a city."},
          {"role": "user", "content": capital}],
      temperature=1,
      max_tokens=200,
    ).choices[0].message.content
 
    return poem
 
# run main function and let Langfuse decorator do the rest
print(main("Bulgaria", "admin"))

from langfuse.openai import openai
from langfuse import observe, get_client, propagate_attributes
langfuse = get_client()
 
@observe() # decorator to automatically create trace and nest generations
def main(country: str, user_id: str, **kwargs) -> str:
 
    # Propagate attributes to all child observations
    with propagate_attributes(
        session_id="1234",
        user_id=user_id,
        tags=["tag1", "tag2"],
        metadata = {"env": "development"}
    ):
 
      # nested generation 1: use openai to get capital of country
      capital = openai.chat.completions.create(
        name="geography-teacher",
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a Geography teacher helping students learn the capitals of countries. Output only the capital when being asked."},
            {"role": "user", "content": country}],
        temperature=0,
      ).choices[0].message.content
 
      # nested generation 2: use openai to write poem on capital
      poem = openai.chat.completions.create(
        name="poet",
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a poet. Create a poem about a city."},
            {"role": "user", "content": capital}],
        temperature=1,
        max_tokens=200,
      ).choices[0].message.content
 
    return poem
 
# create random trace_id, could also use existing id from your application, e.g. conversation id
trace_id = langfuse.create_trace_id()
 
# run main function, set your own id, and let Langfuse decorator do the rest
print(main("Bulgaria", "admin", langfuse_observation_id=trace_id))

from langfuse import observe, get_client
langfuse = get_client()
 
@observe() # decorator to automatically create trace and nest generations
def main():
    # get trace_id of current trace
    trace_id = langfuse.get_current_trace_id()
 
    # rest of your application ...
 
    return "res", trace_id
 
# execute the main function to generate a trace
_, trace_id = main()
 
# Score the trace from outside the trace context
langfuse.create_score(
    trace_id=trace_id,
    name="my-score-name",
    value=1
)