Arize Phoenix で実現する LLM アプリケーションのトレース

services:
  phoenix:
    image: arizephoenix/phoenix:latest
    ports:
      - "6006:6006"  # UI and OTLP HTTP collector
      - "4317:4317"  # OTLP gRPC collector
  llm-server:
    build:
      context: .
      dockerfile: Docker/Dockerfile
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      - phoenix

OPENAI_API_KEY="YOUR_OPENAI_API_KEY"
COLLECTOR_ENDPOINT="http://phoenix:6006/v1/traces"
PROD_CORS_ORIGIN="http://phoenix:3000"

import os
from phoenix.otel import register
from openinference.instrumentation.openai import OpenAIInstrumentor

# Initialize tracer provider and tracer
TRACER_PROVIDER = register(
    project_name="phoenix-run-on-docker",
    endpoint=os.getenv("COLLECTOR_ENDPOINT"),
    set_global_tracer_provider=False
)

# Setup OpenAI instrumentation
OpenAIInstrumentor().instrument(tracer_provider=TRACER_PROVIDER)

# Get tracer instance
TRACER = TRACER_PROVIDER.get_tracer(__name__)

import uuid
import json
import chainlit as cl
from dotenv import load_dotenv
from openinference.instrumentation import using_session
from openinference.semconv.trace import SpanAttributes
from phoenix_config import TRACER
from llm_service import get_response
from tools import (
    fetch_db_info,
    auth_user
)

load_dotenv()

MESSAGE_HISTORY = []
session_id = str(uuid.uuid4())


@cl.on_chat_start
async def on_chat_start():
    cl.Message(content="Ask me anything!").send()


# エラー分析のため、一部挙動パターンに対応していない不完全な関数となっています
@cl.on_message
async def on_message(message: cl.Message):
    if not message.content:
        await cl.Message(content="Feel free to ask me anything!").send()

    with TRACER.start_as_current_span(name="agent", attributes={SpanAttributes.OPENINFERENCE_SPAN_KIND: "agent"}) as span:
        with using_session(session_id):
            span.add_event("チャットストリームを開始")
            span.set_attribute(SpanAttributes.INPUT_VALUE, message.content)
            
            MESSAGE_HISTORY.append({"role": "user", "content": message.content})
            response = get_response(MESSAGE_HISTORY)
            if response.choices[0].message.tool_calls:
                span.add_event("function calling の実施")
                tool_call = response.choices[0].message.tool_calls[0]
                args = json.loads(tool_call.function.arguments)
                tool_name = tool_call.function.name
                try:
                    if tool_name == "fetch_db_info":
                        result = fetch_db_info()
                    elif tool_name == "auth_user":
                        result = auth_user(args["username"], args["password"])
                    else:
                        raise ValueError(f"Unknown tool name: {tool_name}")
                except Exception as e:
                    raise e

                MESSAGE_HISTORY.append(response.choices[0].message)
                MESSAGE_HISTORY.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result)
                })
                response = get_response(MESSAGE_HISTORY)
                MESSAGE_HISTORY.append(response.choices[0].message)

                response_content = response.choices[0].message.content
            else:
                response_content = response.choices[0].message.content
                MESSAGE_HISTORY.append({"role": "assistant", "content": response_content})

        span.set_attribute(SpanAttributes.OUTPUT_VALUE, response_content)
        await cl.Message(content=response_content).send()

import random
import time
from phoenix_config import TRACER

TOOL_DESCRIPTIONS = [
    {
        "type": "function",
        "function": {
            "name": "fetch_db_info",
            "description": "Fetch a list of users from a database. This tool can only be used by users who are authorized. Make sure to use the auth_user tool before using this tool.",
            "parameters": {
                "type": "object",
                "properties": {}
            },
            "required": []
        }
    },
    {
        "type": "function",
        "function": {
            "name": "auth_user",
            "description": "Authenticate a user with a password. Need username and password.",
            "parameters": {
                "type": "object",
                "properties": {
                    "username": {"type": "string"},
                    "password": {"type": "string"}
                },
                "required": ["username", "password"]
            },
            "required": []
        }
    }
]

@TRACER.tool()
def fetch_db_info():
    # 1-3秒のランダム遅延
    delay = random.uniform(1.0, 3.0)
    time.sleep(delay)
    
    # 10%の確率で失敗
    if random.random() < 0.1:
        raise RuntimeError("Failed to fetch users data")
    
    # ダミーのユーザーデータを生成して返す
    dummy_users = [
        {"id": 1, "name": "Taro Tanaka", "email": "tanaka@example.com", "age": 28},
        {"id": 2, "name": "Hana Sato", "email": "sato@example.com", "age": 34},
        {"id": 3, "name": "Suzuki Ichiro", "email": "suzuki@example.com", "age": 42},
        {"id": 4, "name": "Yuko Ito", "email": "ito@example.com", "age": 25},
        {"id": 5, "name": "Ken Yamada", "email": "yamada@example.com", "age": 31}
    ]
    
    return dummy_users

@TRACER.chain()
def get_password(username):
    passwords = {
        "ais_user": ["aishift2025"],
        "admin": ["admin_master_password"]
    }
    return passwords[username]

@TRACER.chain()
def get_auth_users():
    users = ["ais_user", "admin"]
    return users

@TRACER.tool()
def auth_user(username, password):
    auth_users = get_auth_users()
    auth_password = get_password(username)
    if password in auth_password and username in auth_users:
        return "Authorized"
    else:
        return "Not authorized"