admin
9fb191884e
- 更新考勤jupyter notebook的登录账号、密码与请求方式 - 新增Dify API调用脚本test.py用于测试对话请求 - 新增database.py实现批量格式化处理教学计划数据并生成更新SQL
131 lines
3.6 KiB
Python
131 lines
3.6 KiB
Python
import requests
|
||
import json
|
||
|
||
"""
|
||
请求 Dify API 的示例脚本
|
||
接口: POST http://net2.hxzhxy.cn:5093/v1/chat-messages
|
||
|
||
使用方法：
|
||
1. 设置环境变量 DIFY_API_KEY，或者直接在下面填入 api_key
|
||
2. 运行脚本
|
||
"""
|
||
|
||
def send_chat_message(
|
||
query: str,
|
||
api_key: str = None,
|
||
api_url: str = "http://net2.hxzhxy.cn:5093/v1/chat-messages",
|
||
user: str = "abc-123",
|
||
response_mode: str = "streaming", # streaming 或 blocking
|
||
conversation_id: str = "",
|
||
inputs: dict = None,
|
||
files: list = None
|
||
):
|
||
"""发送对话消息到 Dify API"""
|
||
|
||
if api_key is None:
|
||
import os
|
||
api_key = os.environ.get("DIFY_API_KEY", "")
|
||
|
||
if inputs is None:
|
||
inputs = {}
|
||
|
||
if files is None:
|
||
files = []
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
payload = {
|
||
"inputs": inputs,
|
||
"query": query,
|
||
"response_mode": response_mode,
|
||
"conversation_id": conversation_id,
|
||
"user": user,
|
||
"files": files
|
||
}
|
||
|
||
print(f"请求地址: {api_url}")
|
||
print(f"用户: {user}")
|
||
print(f"消息: {query}")
|
||
print(f"响应模式: {response_mode}")
|
||
|
||
if response_mode == "streaming":
|
||
# 流式响应
|
||
response = requests.post(
|
||
url=api_url,
|
||
headers=headers,
|
||
json=payload,
|
||
stream=True,
|
||
timeout=120
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
print(f"请求失败: {response.status_code}")
|
||
print(response.text)
|
||
return
|
||
|
||
print("\n--- 流式响应 ---")
|
||
for line in response.iter_lines():
|
||
if line:
|
||
try:
|
||
data = json.loads(line.decode("utf-8").lstrip("data: "))
|
||
if "answer" in data:
|
||
print(data["answer"], end="", flush=True)
|
||
except (json.JSONDecodeError, KeyError):
|
||
pass
|
||
print("\n--- 响应结束 ---")
|
||
return response
|
||
|
||
else:
|
||
# 非流式响应（blocking）
|
||
response = requests.post(
|
||
url=api_url,
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=120
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
print(f"请求失败: {response.status_code}")
|
||
print(response.text)
|
||
return response
|
||
|
||
result = response.json()
|
||
print(f"\n对话ID: {result.get('conversation_id', '')}")
|
||
print(f"回复: {result.get('answer', '')}")
|
||
return result
|
||
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# ===== 配置参数（请在运行前填写）=====
|
||
API_KEY = "app-T0kpinzer52gZDGCtipsShVm" # Dify API Key
|
||
USER_ID = "xtlft@qq.com"
|
||
test = {
|
||
"id": 293,
|
||
"batch_name": "英语期末模拟",
|
||
"fixed_header": "[\"姓名\", \"学号\", \"听力\", \"写作\", \"答题卡\", \"总分\"]",
|
||
"variable_header": "[\"姓名\", \"学号\", \"听力\", \"写作\", \"答题卡\", \"总分\"]",
|
||
"team_id": 19,
|
||
"create_time": "2026-01-12 10:19:43",
|
||
"update_time": "2026-06-09 11:32:44",
|
||
"operator": "handongmei",
|
||
"tea_id": 743,
|
||
"tea_name": "韩冬梅",
|
||
"state": 1,
|
||
"semester": None,
|
||
"subject": "英语",
|
||
"exam_time": None,
|
||
"grade_id": None,
|
||
"batch_no": None
|
||
},
|
||
# 示例1: 流式对话
|
||
send_chat_message(
|
||
query=json.dumps(test, ensure_ascii=False),
|
||
api_key=API_KEY,
|
||
user=USER_ID,
|
||
response_mode="streaming"
|
||
) |