2024-11-26
深度学习
00

目录

流式请求
包装接口,流式返回
阻塞请求

流式请求

python
import requests import json import time def chat_completion_generator(query, timeout=50): baidu_url = 'https://llm_ip/v1/chat-messages' baidu_headers = { 'Authorization': 'Bearer app-2IdfEuDM0EwoKGVxEjC8', 'Content-Type': 'application/json' } baidu_data = { "inputs": {}, "query": query, "response_mode": "streaming", # 假设API支持streaming模式 "user": "abc-123" } try: # 设置stream=True以启用流式处理 response = requests.post(baidu_url, headers=baidu_headers, json=baidu_data, timeout=timeout, stream=True) if response.status_code == 200: # 使用迭代的方式逐步获取数据 for line in response.iter_lines(): if line: # 去除前缀 "data: " line = line.decode('utf-8').strip() if line.startswith("data:"): line = line[5:].strip() try: # 解析JSON数据 event_data = json.loads(line) # 处理不同的事件类型 event = event_data.get('event') if event == 'message': answer = event_data.get('answer', '') # 流式输出answer内容 for char in answer: print(char,end='', flush=True) time.sleep(0.05) except json.JSONDecodeError as e: print(f"Failed to decode JSON: {e}") else: print(f"Error: {response.status_code}") except Exception as e: print(f"Request failed, error: {e}") # 示例调用 query = "你能告诉我今天天气怎么样吗?" chat_completion_generator(query)

包装接口,流式返回

bash
import requests import json import time from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() def chat_completion_generator(query, timeout=50): baidu_url = 'https://llm-api/v1/chat-messages' baidu_headers = { 'Authorization': 'Bearer app-2IdfEuDtM0C8', 'Content-Type': 'application/json' } baidu_data = { "inputs": {}, "query": query, "response_mode": "streaming", # 假设API支持streaming模式 "user": "abc-123" } try: # 设置stream=True以启用流式处理 response = requests.post(baidu_url, headers=baidu_headers, json=baidu_data, timeout=timeout, stream=True) if response.status_code == 200: # 使用迭代的方式逐步获取数据 for line in response.iter_lines(): if line: # 去除前缀 "data: " line = line.decode('utf-8').strip() if line.startswith("data:"): line = line[5:].strip() try: # 解析JSON数据 event_data = json.loads(line) # 处理不同的事件类型 event = event_data.get('event') if event == 'message': answer = event_data.get('answer', '') # 流式输出answer内容 for char in answer: print(char,end='', flush=True) yield char time.sleep(0.05) except json.JSONDecodeError as e: print(f"Failed to decode JSON: {e}") #yield "流式获取数据完成" else: yield f"Error: {response.status_code}" except Exception as e: yield f"Request failed, error: {e}" @app.post("/DocQA_streeam") async def doc_qa_stream(request: Request): form_data = await request.form() query = form_data.get("text") return StreamingResponse(chat_completion_generator(query), media_type="text/event-stream") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8100)

阻塞请求

python
import requests def get_chat_completion_content(query="小狐狸", timeout=50): baidu_url = 'https://llm-api/v1/chat-messages' baidu_headers = { 'Authorization': 'Bearer app-2IdfEuEjC8', 'Content-Type': 'application/json' } baidu_data = { "inputs": {}, "query": query, "response_mode": "blocking", "user": "abc-123" } try: response = requests.post(baidu_url, headers=baidu_headers, json=baidu_data, timeout=timeout) result = response.json() return result['answer'], response.status_code except: return None, 508 print(get_chat_completion_content("小狗"))
如果对你有用的话,可以打赏哦
打赏
ali pay
wechat pay

本文作者:Dong

本文链接:

版权声明:本博客所有文章除特别声明外,均采用 CC BY-NC。本作品采用《知识共享署名-非商业性使用 4.0 国际许可协议》进行许可。您可以在非商业用途下自由转载和修改,但必须注明出处并提供原作者链接。 许可协议。转载请注明出处!