flask转发openai标准接口
背景
搭建RAGFLOW 的过程中，遇到一个比较严重的问题，公司部署的大模型代理需要获取token，且token存在有效期5分钟，在RAGFLOW中不能直接用，所以希望通过flask项目转发请求。
方案
比较好的是，RAGFLOW 可以配置OpenAI-API-Compatible
初始方案是计划准备两个接口，第一个接口/v1/chat/completions
@api_llm_proxy_bp.route("/v1/chat/completions", methods=['POST'])
async def chat_completions():
payload = request.json
logger.debug("chat_completions:{}", payload)
res = await CustommizedAsyncOpenAI().chat.completions.create(**payload)
json_result = res.to_json()
logger.debug("chat_completions response:{}", json_result)
return json_result
CustommizedAsyncOpenAI 是openai._client.AsyncOpenAI的子类，主要作用是封装指定公司大模型的url以及实时获取token放到header里面。
class CustommizedAsyncOpenAI(AsyncOpenAI):
def __init__(self):
base_url = get_config().get("llm_config").get("model_host")
super().__init__(base_url=base_url, api_key="fake key")
async def _prepare_options(self, options: FinalRequestOptions) -> FinalRequestOptions:
options = await super()._prepare_options(options)
options.headers = {
'Authorization': f'Bearer {TokenHolder().get_token()}'
}
return options
另一个接口是向量化接口：/v1/embeddings
@api_llm_proxy_bp.route("/v1/embeddings", methods=['POST'])
async def embeddings():
payload = request.json
logger.debug("embeddings:{}", payload)
res = await CustommizedEmbeddingsOpenAI().embeddings.create(**payload)
json_result = res.to_json()
logger.debug("embeddings response:{}", json_result)
return json_result
CustommizedEmbeddingsOpenAI也是自己写的openai._client.AsyncOpenAI的子类，区别是只对应了公司向量模型的url。
在RAGFLOW 中分别配置chat和embeddings，成功添加模型。
STREAM转发的问题与解决
在RAGFLOW中添加本地文件并解析没有问题，但是后续在使用知识库聊天的时候发现不支持stream请求，导致无法聊天。资料查了半天，对比RAGFLOW处理stream请求的代码，加上了flask 转发open ai的steam标准响应结构：
@api_llm_proxy_bp.route("/v1/chat/completions", methods=['POST'])
async def chat_completions():
payload = request.json
logger.debug("chat_completions:{}", payload)
if payload.get("stream") is True:
res = CustommizedOpenAI().chat.completions.create(**payload)
def stream_response():
for chunk in res:
logger.info("chat_completions response:{}", chunk.to_json())
yield f"data:{json.dumps(chunk.to_dict(), ensure_ascii=False)}" + "\n\n"
return Response(stream_response(), mimetype="text/event-stream")
else:
res = await CustommizedAsyncOpenAI().chat.completions.create(**payload)
json_result = res.to_json()
logger.debug("chat_completions response:{}", json_result)
return json_result
上面的重点是两部分：
- 第一部分，yield 后面的格式：
yield f"data:{json.dumps(chunk.to_dict(), ensure_ascii=False)}" + "\n\n"
- 第二部分，mimetype
Response(stream_response(), mimetype="text/event-stream")
使用的CustommizedOpenAI，是自己实现的openai._client.OpenAI的子类，没有用异步是因为前面异步一直没有成功，现在看原因是data的格式问题，因为有些资料提到flask 异步stream 做的不好，所以先当时暂时使用非异步的方案。
stream 转发的验证
前面因为stream 转发一直没有处理很好，在RAGFLOW中一直没有成功，问题很难排查，原因在于有两层转发，一层是自己的flask项目中的stream转发，另一层是RAGFLOW中也对openai api stream响应做了二次处理，然后再以stream方式响应。
后面反应过来，可以抛开RAGFLOW，验证自己的stream转发是否成功，直接用openai接口去调用自己的接口：
from openai import OpenAI
def test_steam():
client = OpenAI(api_key="fake key", base_url="http://localhost:5002/agent/api/llm_proxy/v1")
messages = [
{"role": "user", "content": "你是谁"},
]
res = client.chat.completions.create(messages=messages, model="gpt-4o-240806-deploy-gs", stream=True)
for chunk in res:
print(chunk.to_json())
如果能连续打印如下的结构，说明stream 转发没有问题：
{
"id": "chatcmpl-B324oTT5jWz3xeACiotosOKzyUk8J",
"choices": [
{
"delta": {
"content": "忙"
},
"finish_reason": null,
"index": 0,
"logprobs": null
}
],
"created": 1740062666,
"model": "gpt-4o-2024-11-20",
"object": "chat.completion.chunk",
"system_fingerprint": "fp_b705f0c291"
}
