当前位置：首页 > web >正文

如何测试调用RagFlow的API功能

web 2025/5/2 5:49:02

环境：

Ragflowv0.17.2

问题描述：

如何测试调用RagFlow的API功能

在这里插入图片描述

解决方案：

1.测试新建助手聊天

逻辑是先创建助手-创建会话- 获取chatid 、会话id然后再开始聊天

测试python脚本内容如下：

import requests
import json# 配置
API_BASE_URL = "http://192.168.208.103:9222/api/v1"  # 替换为实际地址
API_KEY = "ragflow-EwM2M3MGMyWY4MjgzYWIyNT"  # 替换为您的 API Key# 创建聊天助手
def create_chat_assistant(name, dataset_ids, llm=None, prompt=None):url = f"{API_BASE_URL}/chats"  # 确保路径正确headers = {"Content-Type": "application/json","Authorization": f"Bearer {API_KEY}"}payload = {"name": name,"dataset_ids": dataset_ids,"llm": llm if llm else {"model_name": "qwq:32b","temperature": 0.1,"top_p": 0.3,"presence_penalty": 0.4,"frequency_penalty": 0.7,"max_tokens": 512},"prompt": prompt if prompt else {"similarity_threshold": 0.2,"keywords_similarity_weight": 0.3,"top_n": 6,"variables": [{"key": "knowledge", "optional": False}],"empty_response": "Sorry! No relevant content was found in the knowledge base!","opener": "Hi! I'm your assistant, what can I do for you?","prompt": "You are an intelligent assistant. Please summarize the content of the knowledge base to answer the question. Please list the data in the knowledge base and answer in detail. When all knowledge base content is irrelevant to the question, your answer must include the sentence \"The answer you are looking for is not found in the knowledge base!\" Answers need to consider chat history.\n","rerank_model": "","top_k": 1024,"show_quote": True}}# 调试信息print(f"Request URL: {url}")print(f"Request Payload: {json.dumps(payload, indent=2)}")response = requests.post(url, headers=headers, json=payload)# 调试信息print(f"Status Code: {response.status_code}")print(f"Response Text: {response.text}")if response.status_code == 200:try:response_data = response.json()if response_data.get("code") == 0 and response_data.get("data"):return response_data["data"]["id"]  # 返回创建的聊天助手 IDelse:print(f"Failed to create chat assistant: {response_data.get('message')}")return Noneexcept KeyError as e:print(f"Failed to parse response: {e}")return Noneelse:print(f"Failed to create chat assistant: {response.text}")return None# 创建会话
def create_session(chat_id, name, user_id=None):url = f"{API_BASE_URL}/chats/{chat_id}/sessions"  # 确保路径拼写正确headers = {"Content-Type": "application/json","Authorization": f"Bearer {API_KEY}"}payload = {"name": name}if user_id:payload["user_id"] = user_id# 调试信息print(f"Request URL: {url}")print(f"Request Payload: {json.dumps(payload, indent=2)}")response = requests.post(url, headers=headers, json=payload)# 调试信息print(f"Status Code: {response.status_code}")print(f"Response Text: {response.text}")if response.status_code == 200:try:response_data = response.json()if response_data.get("code") == 0 and response_data.get("data"):return response_data["data"]["id"]  # 返回创建的会话 IDelse:print(f"Failed to create session: {response_data.get('message')}")return Noneexcept KeyError as e:print(f"Failed to parse response: {e}")return Noneelse:print(f"Failed to create session: {response.text}")return None# 发送消息（使用 /completions2 接口）
def send_message(chat_id, session_id, question, stream=True):url = f"{API_BASE_URL}/chats/{chat_id}/completions2"  # 确保路径拼写正确headers = {"Content-Type": "application/json","Authorization": f"Bearer {API_KEY}"}payload = {"question": question,"stream": stream,"session_id": session_id  # 可选，如果不需要会话 ID 可以去掉}# 调试信息print(f"Request URL: {url}")print(f"Request Payload: {json.dumps(payload, indent=2)}")if stream:# 处理流式响应with requests.post(url, headers=headers, json=payload, stream=True) as response:for line in response.iter_lines():if line:  # 过滤空行try:# 解码并去除 SSE 格式的 "data:" 前缀line_str = line.decode("utf-8")if line_str.startswith("data:"):json_str = line_str[5:].strip()  # 提取 JSON 部分response_data = json.loads(json_str)if response_data.get("code") == 0 and response_data.get("data"):# 检查 response_data['data'] 是否为字典，并且包含 'answer' 字段if isinstance(response_data['data'], dict) and 'answer' in response_data['data']:print(f"Assistant: {response_data['data'].get('answer')}")else:# 如果 response_data['data'] 不是字典或没有 'answer' 字段，忽略continueexcept json.JSONDecodeError as e:print(f"Failed to parse response line: {line_str}")except Exception as e:print(f"Unexpected error: {e}")else:# 处理非流式响应response = requests.post(url, headers=headers, json=payload)print(f"Status Code: {response.status_code}")print(f"Response Text: {response.text}")if response.status_code == 200:try:response_data = response.json()if response_data.get("code") == 0 and response_data.get("data"):return response_data["data"]  # 返回助手回复else:print(f"Failed to send message: {response_data.get('message')}")return Noneexcept KeyError as e:print(f"Failed to parse response: {e}")return Noneelse:print(f"Failed to send message: {response.text}")return None# 主程序
def main():# 创建聊天助手chat_name = "MyChatAssistant15"dataset_ids = ["07b093be24f811f0952c15368ba25e2d"]  # 替换为实际数据集知识库 IDchat_id = create_chat_assistant(chat_name, dataset_ids)if not chat_id:print("Failed to create chat assistant. Exiting.")returnprint(f"Chat assistant created with ID: {chat_id}")# 创建会话session_name = "MySession1"session_id = create_session(chat_id, session_name)if not session_id:print("Failed to create session. Exiting.")returnprint(f"Session created with ID: {session_id}")# 开始聊天while True:user_input = input("You: ")if user_input.lower() in ["exit", "quit"]:print("Exiting chat.")break# 发送消息并获取助手回复send_message(chat_id, session_id, user_input, stream=True)# 运行主程序
if __name__ == "__main__":main()

在这里插入图片描述

2.测试用上个助手，继续聊天，持续聊天

逻辑是获取之前那个chatid 、会话id然后再开始继续聊天

测试python脚本内容如下：

import requests
import json# 配置
API_BASE_URL = "http://192.168.208.103:9222/api/v1"  # 替换为实际地址
API_KEY = "ragflow-EwM2M3MgzYWIyNT"  # 替换为您的 API Keydef send_message(chat_id, session_id, question, stream=True):"""发送消息并获取助手回复:param chat_id: 聊天助手 ID:param session_id: 会话 ID:param question: 用户输入:param stream: 是否流式返回:return: 助手回复"""url = f"{API_BASE_URL}/chats/{chat_id}/completions2"headers = {"Content-Type": "application/json","Authorization": f"Bearer {API_KEY}"}payload = {"question": question,"stream": stream,"session_id": session_id}try:response = requests.post(url, headers=headers, json=payload)response.raise_for_status()  # 检查 HTTP 状态码是否为 200except requests.exceptions.RequestException as e:print(f"HTTP request failed: {e}")return Nonetry:response_data = response.json()if response_data.get("code") == 0 and response_data.get("data"):return response_data["data"]  # 返回助手回复else:print(f"API returned an error: {response_data.get('message')}")return Noneexcept requests.exceptions.JSONDecodeError:print(f"Failed to parse JSON response. Response text: {response.text}")return Nonedef continue_conversation(chat_id, session_id):"""继续对话:param chat_id: 聊天助手 ID:param session_id: 会话 ID"""while True:user_input = input("You: ")if user_input.lower() in ["exit", "quit"]:print("Exiting chat.")break# 发送消息并获取助手回复response = send_message(chat_id, session_id, user_input, stream=True)if response:print(f"Assistant: {response.get('answer')}")else:print("Failed to send message.")# 示例调用
if __name__ == "__main__":chat_id = "1b6eef5225ae11f08b2ab3eae8a25e0d"session_id = "1b7ddcc425ae11f08b2ab3eae8a25e0d"print(f"Continuing conversation with chat_id={chat_id}, session_id={session_id}")continue_conversation(chat_id, session_id)