流式请求

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("小狗"))