LLM標準介面 - Intelligent Media Services

即時工作流程支援您按照規定的標準協議來接入自研大語言模型（LLM）。

自研LLM標準介面（OpenAI規範）

如果您的LLM介面符合OpenAI標準規範，您可以通過OpenAI標準的配置將您的自研LLM服務整合到工作流程中。目前只支援以流式方式請求符合OpenAI規範的服務。

首先您需要LLM大語言模型節點中，選擇自研接入（OpenAI規範），並配置以下參數：

名稱	類型	必填	描述	樣本值
ModelId	String	是	OpenAI標準model欄位，表示模型名稱	abc
API-KEY	String	是	OpenAI標準api_key欄位，表示API鑒權資訊	AUJH-pfnTNMPBm6iWXcJAcWsrscb5KYaLitQhHBLKrI
目標模型HTTPS地址	String	是	OpenAI標準base_url欄位，表示目標服務要求地址說明按照OpenAI規範，阿里雲會在base_url後自動添加/chat/completions，請確保您的路徑與此匹配。	http://www.abc.com

在即時工作流程運行期間，將按照以下格式，組裝OpenAI標準資料，通過POST請求訪問您配置的自研模型HTTPS地址，擷取相應結果，輸入參數如下：

名稱	類型	必填	描述	樣本值
messages	Array	是	歷史對話上下文，最多保留20條上下文記錄，在數組中越靠前的位置，表示越早的提問或者回答。說明系統將自動將使用者本次講話內容與歷史對話記錄結合，輸入至大模型中。	[{'role': 'user', 'content': '今天天氣怎麼樣？'},{'role': 'assistant', 'content': '今天天氣晴朗。'},{'role': 'user', 'content': '明天天氣晴怎麼樣？'}]
model	String	是	模型名稱資訊	abc
stream	Boolean	是	是否流式，目前只支援串流	True
extendData	Object	是	補充資訊	{'instanceId':'68e00b6640e*****3e943332fee7','channelId':'123','sentenceId':'3',userData':'{"aaaa":"bbbb"}'}
instanceId	String	是	執行個體ID資訊	68e00b6640e*****3e943332fee7
channelId	String	是	房間ID資訊	123
sentenceId	Int	是	問答ID資訊說明對於使用者的同一次提問，智能體回答會使用相同的ID資訊。	3
callerNumber	String	否	電話通話時的主叫號碼。	13800000001
calleeNumber	String	否	電話通話時的被叫號碼。	13800000002
userData	String	否	啟動執行個體時您傳入的UserData業務欄位資料	{"aaaa":"bbbb"}

自訂LLM（OpenAI規範）Server

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 = "這是一個類比的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 = "這是一個類比的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, 
                    "tool_calls": [  
                        {
                            "id": "call_abc123",  
                            "type": "function",
                            "function": {
                                "name": "hangup", 
                                "arguments": "\{\}"  
                            }
                        }
                    ]
                },
                "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}")
    app.run(port=8083, debug=True)