diff --git a/b/cntoai/chat_send.dspy b/b/cntoai/chat_send.dspy index 0611dfb..ba93073 100644 --- a/b/cntoai/chat_send.dspy +++ b/b/cntoai/chat_send.dspy @@ -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'} diff --git a/b/cntoai/chat_session_list.dspy b/b/cntoai/chat_session_list.dspy index c814ef6..e4bddfc 100644 --- a/b/cntoai/chat_session_list.dspy +++ b/b/cntoai/chat_session_list.dspy @@ -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() diff --git a/b/cntoai/llm_chat_completions.dspy b/b/cntoai/llm_chat_completions.dspy index d910986..6303999 100644 --- a/b/cntoai/llm_chat_completions.dspy +++ b/b/cntoai/llm_chat_completions.dspy @@ -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( diff --git a/b/cntoai/test_chat.py b/b/cntoai/test_chat.py new file mode 100644 index 0000000..3432eac --- /dev/null +++ b/b/cntoai/test_chat.py @@ -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())