自研LLM標(biāo)準(zhǔn)接口（OpenAI規(guī)范）

如果您的LLM接口符合OpenAI標(biāo)準(zhǔn)規(guī)范，您可以通過(guò)OpenAI標(biāo)準(zhǔn)的配置將您的自研LLM服務(wù)整合到工作流中。目前只支持以流式方式請(qǐng)求符合OpenAI規(guī)范的服務(wù)。

1. 首先您需要LLM大語(yǔ)言模型節(jié)點(diǎn)中，選擇自研接入（OpenAI規(guī)范），并配置以下參數(shù)：

2. 在實(shí)時(shí)工作流運(yùn)行期間，將按照以下格式，組裝OpenAI標(biāo)準(zhǔn)數(shù)據(jù)，通過(guò)POST請(qǐng)求訪問(wèn)您配置的自研模型HTTPS地址，獲取相應(yīng)結(jié)果，輸入?yún)?shù)如下：

自定義LLM（OpenAI規(guī)范）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}")
    
    # 打印查詢參數(shù)
    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 = "這是一個(gè)模擬的AI助手響應(yīng)。實(shí)際應(yīng)用中，這里應(yīng)該調(diào)用真實(shí)的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 = "這是一個(gè)模擬的AI助手流式響應(yīng)。實(shí)際應(yīng)用中，這里應(yīng)該調(diào)用真實(shí)的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)  # 模擬處理時(shí)間
    
    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)

自研LLM標(biāo)準(zhǔn)接口（阿里規(guī)范）

您需要實(shí)現(xiàn)一個(gè)公網(wǎng)可訪問(wèn)的HTTP服務(wù)，按照規(guī)定的輸入輸出規(guī)范，封裝好您的大語(yǔ)言模型，即可在工作流中使用您自研的LLM服務(wù)。

1. 首先您需要在編排管理控制臺(tái)中的LLM 大語(yǔ)言模型節(jié)點(diǎn)中，配置以下參數(shù)：

2. 在實(shí)時(shí)工作流運(yùn)行期間，將按照以下格式，組裝表單數(shù)據(jù)，通過(guò)POST請(qǐng)求訪問(wèn)您配置的自研模型HTTPS地址，獲取相應(yīng)結(jié)果，輸入?yún)?shù)如下：

3. 響應(yīng)參數(shù)您需要組裝成JSON返回給編排服務(wù)，參數(shù)如下：

示例如下：

{
	"Text": "你好，我是您的人工智能助手。"
}

自定義LLM（阿里規(guī)范） Server

Python

import json
from aiohttp import web


async def handle(request):
    data = await request.post()
    message = data.get('Message', "")
    token = data.get('Token', None)
    extend_data = data.get('ExtendData', "")
    print(f"message:{message}, token:{token}, extend_data:{extend_data}")
    # TODO:1. 校驗(yàn)Token是否合法
    # TODO:2. 處理Message獲取LLM處理結(jié)果

    result = "你好，我是您的人工智能助手。"
    # 組裝LLM節(jié)點(diǎn)期望結(jié)果
    resp = {
        "Text": result
    }
    return web.Response(text=json.dumps(resp))

app = web.Application()
app.add_routes([web.post('/', handle)])

if __name__ == '__main__':
    web.run_app(app)

相關(guān)文檔

通過(guò)控制臺(tái)創(chuàng)建實(shí)時(shí)工作流模板