Python实现AnythingLLM API流输出调用的方法步骤
在使用 Python 调用 AnythingLLM API 时,流式输出(Streaming)是一种常见的需求,尤其是在处理长文本或实时交互场景时。流式输出可以让你逐步接收响应内容,而不是等待整个响应完成后再处理。
以下是如何实现流式输出的详细步骤和代码示例:
1. 流式输出的基本原理
流式输出允许客户端逐步接收响应数据,而不是一次性接收完整响应。这在处理大模型的长文本输出时非常有用,可以减少延迟并提高用户体验。
stream=True
参数来实现。Content-Type: text/event-stream
或类似机制实现)。2. 代码实现
以下是一个完整的 Python 示例,展示如何使用 requests
库实现流式输出:
import requests
import json
def ask_anythingllm(question, workspace_name, api_key):
url = f"http://ip:port/api/v1/workspace/{workspace_name}/stream-chat" #注意api中有一个不带stream的接口,该接口会一次性返回
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"accept": "text/event-stream"
}
data = {
"message": question,
"mode": "query", # 可选chat/query模式
'max_tokens': 1024, # 控制生成文本的长度
"stream":True
}
with requests.post(url, headers=headers, json=data, stream=True) as response:
if response.status_code == 200:
# 逐行读取流式响应
for chunk in response.iter_lines():
if chunk: # 确保数据块不为空
print(chunk.decode("utf-8"))
# 示例调用
api_key = "WQ59GRH-1JC4M3R-MS2NN3X-VBQCY7H" # 替换成你自己的apikey
workspace = "8ceb3fb1-4e75-40fe-87db-570d5a689113"
question = "三字经讲的是什么,用50个字概括"
ask_anythingllm(question, workspace, api_key)
3.获取workspace_name方法
可以用此方法获取workspace_namel列表,当然也可以直接在AnythingLLM找到
import requests
headers = {
"Authorization": "Bearer WQ59GRH-1JC4M3R-MS2NN3X-VBQCY7H", # 替换为你的API Key
"accept": "application/json"
}
response = requests.get("http://ip:3001/api/v1/workspaces", headers=headers)
if response.status_code == 200:
print("现有工作空间列表:", response.json())
else:
print(f"获取失败,状态码:{response.status_code},错误信息:{response.text}")
4. 关键点说明
-
stream=True
参数: - 在
requests.post()
中设置stream=True
,这样可以逐行读取响应内容,而不是一次性接收完整响应。 - 使用
response.iter_lines()
方法逐行读取流式响应。 -
响应头设置:
- 确保请求头中包含
accept: text/event-stream
,这告诉服务器你期望流式响应。 - 服务器端需要支持流式响应(如返回
Content-Type: text/event-stream
)。 -
逐行处理响应:
- 流式响应通常以行的形式返回,每一行可能是一个完整的事件或部分数据。
- 你可以根据需要对每一行进行解析或处理。
5. 调试和注意事项
6. 扩展应用
流式输出非常适合以下场景:
通过上述方法,你可以轻松实现 Python 调用 AnythingLLM API 的流式输出功能。
作者:豆芽脚脚