update
This commit is contained in:
ping
parent
a0ff6ba2c5
commit
f60c761735
@ -72,8 +72,9 @@ async def chat_send(ns={}):
|
||||
"""
|
||||
import json
|
||||
import traceback
|
||||
|
||||
model = ns.get('model')
|
||||
|
||||
# model = ns.get('model')
|
||||
model = 'deepseek-v4-pro'
|
||||
if not model:
|
||||
return {'status': False, 'msg': 'model is required'}
|
||||
|
||||
|
||||
@ -10,8 +10,8 @@ async def chat_session_list(ns={}):
|
||||
if not userid:
|
||||
return {'status': False, 'msg': '未找到用户'}
|
||||
|
||||
page_size = int(ns.get('page_size', 50))
|
||||
current_page = int(ns.get('current_page', 1))
|
||||
page_size = int(ns.get('page_size')) if ns.get('page_size') else 100
|
||||
current_page = int(ns.get('current_page')) if ns.get('current_page') else 1
|
||||
offset = (current_page - 1) * page_size
|
||||
|
||||
db = DBPools()
|
||||
|
||||
@ -81,8 +81,11 @@ def build_user_content(ns):
|
||||
|
||||
async def _resolve_chat_config(ns, sor):
|
||||
"""解析 API 地址与 Bearer Token"""
|
||||
api_url = ns.get('api_url')
|
||||
api_key = ns.get('api_key')
|
||||
api_url = 'https://api.deepseek.com/chat/completions'
|
||||
api_key = 'sk-c22d6573e85a4d3fa8ab932386cf2909'
|
||||
|
||||
# api_url = ns.get('api_url')
|
||||
# api_key = ns.get('api_key')
|
||||
|
||||
if not api_url and ns.get('model_id'):
|
||||
doc_rows = await sor.sqlExe(
|
||||
|
||||
262
b/cntoai/test_chat.py
Normal file
262
b/cntoai/test_chat.py
Normal file
@ -0,0 +1,262 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
cntoai 对话相关接口联调测试（dev.opencomputing.cn）
|
||||
|
||||
用法：
|
||||
pip install requests
|
||||
set CNTOAI_USERID=你的用户id
|
||||
set CNTOAI_API_KEY=你的api_key
|
||||
python test_chat.py
|
||||
|
||||
可选环境变量：
|
||||
CNTOAI_BASE_URL 默认 https://dev.opencomputing.cn
|
||||
CNTOAI_MODEL 默认 qwen3.6-plus
|
||||
CNTOAI_LLM_API_URL 默认 https://ai.atvoe.com/llmage/v1/chat/completions
|
||||
CNTOAI_COOKIE 浏览器 Cookie（未传 userid 时用于鉴权）
|
||||
|
||||
单测：
|
||||
python test_chat.py --only models
|
||||
python test_chat.py --only completions
|
||||
python test_chat.py --only send
|
||||
python test_chat.py --only session
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from typing import Any, Dict, Optional, Tuple
|
||||
|
||||
try:
|
||||
import requests
|
||||
except ImportError:
|
||||
print("请先安装依赖: pip install requests")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
BASE_URL = "https://dev.opencomputing.cn"
|
||||
USERID = "hSqZuekZ1yKmhKmCN9UAK"
|
||||
API_KEY = "sk-c22d6573e85a4d3fa8ab932386cf2909"
|
||||
# API_URL = "https://ai.atvoe.com/llmage/v1/chat/completions"
|
||||
API_URL = "https://api.deepseek.com/chat/completions"
|
||||
# MODEL = "qwen3.6-plus"
|
||||
MODEL = "deepseek-v4-pro"
|
||||
COOKIE = "".strip()
|
||||
TIMEOUT = int(120)
|
||||
|
||||
|
||||
class ChatApiClient:
|
||||
def __init__(self, base_url: str = BASE_URL):
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({"Accept": "application/json"})
|
||||
if COOKIE:
|
||||
self.session.headers["Cookie"] = COOKIE
|
||||
|
||||
def _url(self, path: str) -> str:
|
||||
path = path if path.startswith("/") else f"/{path}"
|
||||
return f"{self.base_url}{path}"
|
||||
|
||||
def _auth(self, extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
params: Dict[str, Any] = {}
|
||||
if USERID:
|
||||
params["userid"] = USERID
|
||||
if API_KEY:
|
||||
params["api_key"] = API_KEY
|
||||
if API_URL:
|
||||
params["api_url"] = API_URL
|
||||
if extra:
|
||||
params.update(extra)
|
||||
return params
|
||||
|
||||
def get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
resp = self.session.get(self._url(path), params=self._auth(params), timeout=TIMEOUT)
|
||||
return self._parse(resp)
|
||||
|
||||
def post(self, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
resp = self.session.post(
|
||||
self._url(path),
|
||||
json=self._auth(data),
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=TIMEOUT,
|
||||
)
|
||||
return self._parse(resp)
|
||||
|
||||
@staticmethod
|
||||
def _parse(resp: requests.Response) -> Dict[str, Any]:
|
||||
print(f" HTTP {resp.status_code} {resp.url[:120]}...")
|
||||
try:
|
||||
return resp.json()
|
||||
except Exception:
|
||||
return {"status": False, "msg": f"非 JSON 响应: {resp.text[:300]}"}
|
||||
|
||||
|
||||
def ok(name: str, data: Dict[str, Any]) -> bool:
|
||||
passed = data.get("status") is True
|
||||
tag = "PASS" if passed else "FAIL"
|
||||
print(f"\n[{tag}] {name}")
|
||||
print(json.dumps(data, ensure_ascii=False, indent=2)[:2000])
|
||||
if not passed:
|
||||
print(f" -> {data.get('msg', 'unknown error')}")
|
||||
return passed
|
||||
|
||||
|
||||
def test_model_list(client: ChatApiClient) -> bool:
|
||||
print("\n=== GET /cntoai/model_management_customer_search.dspy ===")
|
||||
data = client.get("/cntoai/model_management_customer_search.dspy", {
|
||||
"page_size": 20,
|
||||
"current_page": 1,
|
||||
})
|
||||
if ok("模型列表", data) and data.get("data"):
|
||||
models = data["data"].get("model_list") or []
|
||||
print(f" 共 {len(models)} 个模型")
|
||||
if models:
|
||||
m0 = models[0]
|
||||
print(f" 首个: {m0.get('model_name')} / {m0.get('display_name')}")
|
||||
return data.get("status") is True
|
||||
|
||||
|
||||
def test_llm_chat_completions(client: ChatApiClient) -> bool:
|
||||
print("\n=== POST /cntoai/llm_chat_completions.dspy ===")
|
||||
data = client.post("/cntoai/llm_chat_completions.dspy", {
|
||||
"model": MODEL,
|
||||
"message": "用一句话介绍你自己",
|
||||
"stream": True,
|
||||
})
|
||||
if ok("直连模型", data) and data.get("data"):
|
||||
print(f" 回复摘要: {(data['data'].get('reply') or '')[:200]}")
|
||||
return data.get("status") is True
|
||||
|
||||
|
||||
def test_chat_send(
|
||||
client: ChatApiClient,
|
||||
session_id: Optional[str] = None,
|
||||
) -> Tuple[bool, Optional[str]]:
|
||||
print("\n=== POST /cntoai/chat_send.dspy ===")
|
||||
payload: Dict[str, Any] = {
|
||||
"model": MODEL,
|
||||
"message": (
|
||||
"你好，这是 test_chat.py 自动化测试"
|
||||
if not session_id
|
||||
else "继续，用一句话回复我"
|
||||
),
|
||||
"stream": True,
|
||||
}
|
||||
if session_id:
|
||||
payload["session_id"] = session_id
|
||||
data = client.post("/cntoai/chat_send.dspy", payload)
|
||||
if ok("发送消息", data) and data.get("data"):
|
||||
sid = data["data"].get("session_id")
|
||||
print(f" session_id: {sid}")
|
||||
print(f" 回复摘要: {(data['data'].get('reply') or '')[:200]}")
|
||||
return True, sid
|
||||
return False, session_id
|
||||
|
||||
|
||||
def test_chat_session_list(client: ChatApiClient) -> bool:
|
||||
print("\n=== GET /cntoai/chat_session_list.dspy ===")
|
||||
data = client.get("/cntoai/chat_session_list.dspy", {"page_size": 10})
|
||||
if ok("会话列表", data) and data.get("data"):
|
||||
sessions = data["data"].get("sessions") or []
|
||||
print(f" 共 {data['data'].get('total_count', len(sessions))} 条会话")
|
||||
for s in sessions[:3]:
|
||||
print(f" - {s.get('id')} | {s.get('title')}")
|
||||
return data.get("status") is True
|
||||
|
||||
|
||||
def test_chat_session_messages(client: ChatApiClient, session_id: str) -> bool:
|
||||
print("\n=== GET /cntoai/chat_session_messages.dspy ===")
|
||||
data = client.get("/cntoai/chat_session_messages.dspy", {"session_id": session_id})
|
||||
if ok("会话消息", data) and data.get("data"):
|
||||
msgs = data["data"].get("messages") or []
|
||||
print(f" 消息数: {len(msgs)}")
|
||||
for m in msgs:
|
||||
print(f" [{m.get('role')}] {str(m.get('content') or '')[:80]}")
|
||||
return data.get("status") is True
|
||||
|
||||
|
||||
def test_chat_session_delete(client: ChatApiClient, session_id: str) -> bool:
|
||||
print("\n=== GET /cntoai/chat_session_delete.dspy ===")
|
||||
data = client.get("/cntoai/chat_session_delete.dspy", {"session_id": session_id})
|
||||
return ok("删除会话", data)
|
||||
|
||||
|
||||
def check_config(require_userid: bool = True) -> bool:
|
||||
print("配置:")
|
||||
print(f" BASE_URL = {BASE_URL}")
|
||||
print(f" MODEL = {MODEL}")
|
||||
print(f" API_URL = {API_URL or '(走服务端配置)'}")
|
||||
print(f" USERID = {USERID or '(未设置)'}")
|
||||
print(f" API_KEY = {'已设置' if API_KEY else '(未设置)'}")
|
||||
print(f" COOKIE = {'已设置' if COOKIE else '(未设置)'}")
|
||||
|
||||
if require_userid and not USERID and not COOKIE:
|
||||
print("\n错误: 持久化接口需要 CNTOAI_USERID 或 CNTOAI_COOKIE")
|
||||
return False
|
||||
if not API_KEY:
|
||||
print("\n警告: 未设置 CNTOAI_API_KEY，将依赖服务端 Key")
|
||||
return True
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="cntoai chat API 联调测试")
|
||||
parser.add_argument(
|
||||
"--only",
|
||||
choices=["models", "completions", "send", "session", "delete", "all"],
|
||||
default="all",
|
||||
)
|
||||
parser.add_argument("--keep-session", action="store_true", help="不删除测试会话")
|
||||
parser.add_argument("--base-url", default=BASE_URL)
|
||||
args = parser.parse_args()
|
||||
|
||||
client = ChatApiClient(base_url=args.base_url)
|
||||
results = []
|
||||
session_id: Optional[str] = None
|
||||
|
||||
if args.only in ("all", "models"):
|
||||
results.append(("models", test_model_list(client)))
|
||||
|
||||
if args.only in ("all", "completions"):
|
||||
if check_config(require_userid=False):
|
||||
results.append(("completions", test_llm_chat_completions(client)))
|
||||
else:
|
||||
results.append(("completions", False))
|
||||
|
||||
if args.only in ("all", "send", "session", "delete"):
|
||||
if not check_config(require_userid=True):
|
||||
return 1
|
||||
|
||||
if args.only in ("all", "send"):
|
||||
passed, session_id = test_chat_send(client)
|
||||
results.append(("send_1", passed))
|
||||
if passed and session_id:
|
||||
time.sleep(1)
|
||||
passed2, session_id = test_chat_send(client, session_id=session_id)
|
||||
results.append(("send_2_multiturn", passed2))
|
||||
|
||||
if args.only in ("all", "session") and session_id:
|
||||
results.append(("session_list", test_chat_session_list(client)))
|
||||
results.append(("session_messages", test_chat_session_messages(client, session_id)))
|
||||
elif args.only == "session":
|
||||
results.append(("session_list", test_chat_session_list(client)))
|
||||
|
||||
if args.only in ("all", "delete") and session_id and not args.keep_session:
|
||||
results.append(("delete", test_chat_session_delete(client, session_id)))
|
||||
elif session_id and args.keep_session:
|
||||
print(f"\n保留测试会话: {session_id}")
|
||||
|
||||
print("\n" + "=" * 50)
|
||||
print("汇总:")
|
||||
failed = sum(1 for _, p in results if not p)
|
||||
for name, passed in results:
|
||||
print(f" {'OK' if passed else 'FAIL'} {name}")
|
||||
print("=" * 50)
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
x
Reference in New Issue
Block a user