RAGFLOW使用flask转发的open ai接口

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.jsonlogger.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.jsonlogger.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.jsonlogger.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

上面的重点是两部分：

1. 第一部分，yield 后面的格式：

yield f"data:{json.dumps(chunk.to_dict(), ensure_ascii=False)}" + "\n\n"

2. 第二部分，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 OpenAIdef 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"
}