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