Intelligent Media Services:LLM へのアクセス

Access self-developed LLMs

OpenAI 仕様に準拠した LLM をワークフローに統合できます。現在、ストリーミングリクエストのみがサポートされています。

以下の手順に従って、自社開発の LLM にアクセスします。

1. LLM ノードで、[自社開発モデルへのアクセス（OpenAI 仕様に基づく）] を選択し、以下のパラメーターを構成します。

2. リアルタイムワークフローの実行中は、データが OpenAI 標準フォーマットでアセンブルされ、POST リクエストを介して構成済みの自社開発モデルの HTTPS ベース URL に送信されます。入力パラメーターは以下のとおりです。

Custom LLM server (OpenAI specifications)

Python

import json
import time
from loguru import logger
from flask import Flask, request, jsonify, Response

app = Flask(__name__)

API_KEY = "YOURAPIKEY"

@app.route('/v1/chat/completions', methods=['POST'])
def chat_completion():
    # APIキーを確認する
    auth_header = request.headers.get('Authorization')
    if not auth_header or auth_header.split()[1] != API_KEY:
        return jsonify({"error": "Unauthorized"}), 401
    
    data = request.json
    logger.info(f"data is {data}")
    task_id = request.args.get('task_id')
    room_id = request.args.get('room_id')
    for header, value in request.headers.items():
        logger.info(f"{header}: {value}")
    
    # クエリパラメータを出力する
    logger.info("\nQuery Parameters:")
    for key, value in request.args.items():
        logger.info(f"{key}: {value}")
    
    logger.info(f"task_id: {task_id}, room_id: {room_id}")
    stream = data.get('stream', False)
    
    if stream:
        return Response(generate_stream_response(data), content_type='text/event-stream')
    else:
        return jsonify(generate_response(data))

def generate_response(data):
    response = "This is a simulated AI assistant response. In actual applications, a real AI model should be called here." # これはAIアシスタントのシミュレーション応答です。実際のアプリケーションでは、ここで実際のAIモデルを呼び出す必要があります。
    
    return {
        "id": "chatcmpl-123",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": data['model'],
        "choices": [{
            "index": 0,
            "message": {
                "role": "assistant",
                "content": response
            },
            "finish_reason": "stop"
        }],
        "usage": {
            "prompt_tokens": sum(len(m['content']) for m in data['messages']),
            "completion_tokens": len(response),
            "total_tokens": sum(len(m['content']) for m in data['messages']) + len(response)
        }
    }

def generate_stream_response(data):
    response = "This is a simulated AI assistant streaming response. In actual applications, a real AI model should be called here." # これはAIアシスタントのストリーミングシミュレーション応答です。実際のアプリケーションでは、ここで実際のAIモデルを呼び出す必要があります。
    words = list(response)
    for i, word in enumerate(words):
        chunk = {
            "id": "chatcmpl-123",
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": data['model'],
            "choices": [{
                "index": 0,
                "delta": {
                    "content": word 
                },
                "finish_reason": None if i < len(words) - 1 else "stop"
            }]
        }
        logger.info(chunk)
        yield f"data: {json.dumps(chunk)}\n\n"
        time.sleep(0.1)  # 処理時間をシミュレートする
    
    yield "data: [DONE]\n\n"

if __name__ == '__main__':
    logger.info(f"Server is running with API_KEY: {API_KEY}") # サーバーはAPI_KEYで実行されています: {API_KEY}
    app.run(port=8083, debug=True)