Assistant (#48)

 
* 助手业务 test_assistant

Author: glide-the

tests/integration_tests/test_assistant.py

@@ -0,0 +1,75 @@
+import os
+import logging
+import logging.config
+import zhipuai
+from zhipuai import ZhipuAI
+
+
+def test_assistant(logging_conf) -> None:
+    logging.config.dictConfig(logging_conf)  # type: ignore
+    client = ZhipuAI()  # 填写您自己的APIKey
+    try:
+
+        generate = client.assistant.conversation(
+            assistant_id="659e54b1b8006379b4b2abd6",
+            model="glm-4-assistant",
+            messages=[
+                {
+                    "role": "user",
+                    "content": [{
+                        "type": "text",
+                        "text": "帮我搜索下智谱的cogvideox发布时间"
+                    }]
+                }
+            ],
+            stream=True,
+            attachments=None,
+            metadata=None,
+            request_id="request_1790291013237211136",
+            user_id="12345678"
+        )
+        for assistant in generate:
+            print(assistant)
+
+    except zhipuai.core._errors.APIRequestFailedError as err:
+        print(err)
+    except zhipuai.core._errors.APIInternalError as err:
+        print(err)
+    except zhipuai.core._errors.APIStatusError as err:
+        print(err)
+
+def test_assistant_query_support(logging_conf) -> None:
+    logging.config.dictConfig(logging_conf)  # type: ignore
+    client = ZhipuAI()  # 填写您自己的APIKey
+    try:
+
+        response = client.assistant.query_support(
+                assistant_id_list=[],
+                request_id="request_1790291013237211136",
+                user_id="12345678"
+            )
+        print(response)
+
+    except zhipuai.core._errors.APIRequestFailedError as err:
+        print(err)
+    except zhipuai.core._errors.APIInternalError as err:
+        print(err)
+    except zhipuai.core._errors.APIStatusError as err:
+        print(err)
+
+def test_assistant_query_conversation_usage(logging_conf) -> None:
+    logging.config.dictConfig(logging_conf)  # type: ignore
+    client = ZhipuAI()  # 填写您自己的APIKey
+    try:
+        response = client.assistant.query_conversation_usage(
+            assistant_id="659e54b1b8006379b4b2abd6",
+            request_id="request_1790291013237211136",
+            user_id="12345678"
+        )
+        print(response)
+    except zhipuai.core._errors.APIRequestFailedError as err:
+        print(err)
+    except zhipuai.core._errors.APIInternalError as err:
+        print(err)
+    except zhipuai.core._errors.APIStatusError as err:
+        print(err)

zhipuai/_client.py

@@ -61,6 +61,7 @@ def __init__(
         self.knowledge = api_resource.Knowledge(self)
         self.tools = api_resource.Tools(self)
         self.videos = api_resource.Videos(self)
+        self.assistant = api_resource.Assistant(self)
 
     @property
     @override

zhipuai/api_resource/__init__.py

@@ -30,6 +30,10 @@
 from .videos import (
     Videos,
 )
+from .assistant import (
+    Assistant,
+)
+
 __all__ = [
     'Videos',
     'AsyncCompletions',
@@ -43,5 +47,6 @@
     'Batches',
     'Knowledge',
     'Tools',
+    'Assistant',
 
 ]

zhipuai/api_resource/assistant/__init__.py

@@ -0,0 +1,7 @@
+
+
+from zhipuai.api_resource.assistant.assistant import Assistant
+
+__all__= [
+    "Assistant"
+]

zhipuai/api_resource/assistant/assistant.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, List, Mapping, cast, Optional, Dict
+from typing_extensions import Literal
+
+from ...types.assistant import AssistantCompletion
+from ...types.assistant.assistant_conversation_resp import ConversationUsageList, ConversationUsageListResp
+from ...types.assistant.assistant_support_resp import AssistantSupportResp
+from ...core import BaseAPI, maybe_transform, StreamResponse
+from ...core import NOT_GIVEN, Body, Headers, NotGiven
+
+import httpx
+
+from ...core import (
+    make_request_options,
+)
+from ...core import deepcopy_minimal, extract_files
+
+if TYPE_CHECKING:
+    from ..._client import ZhipuAI
+
+from ...types.assistant import assistant_create_params
+from ...types.assistant import assistant_conversation_params
+
+__all__ = ["Assistant"]
+
+
+class Assistant(BaseAPI):
+
+    def __init__(self, client: "ZhipuAI") -> None:
+        super().__init__(client)
+
+    def conversation(
+            self,
+            assistant_id: str,
+            model: str,
+            messages: List[assistant_create_params.ConversationMessage],
+            *,
+            stream: bool = True,
+            conversation_id: Optional[str] = None,
+            attachments: Optional[List[assistant_create_params.AssistantAttachments]] = None,
+            metadata: dict | None = None,
+            request_id: str = None,
+            user_id: str = None,
+            extra_headers: Headers | None = None,
+            extra_body: Body | None = None,
+            timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
+    ) -> StreamResponse[AssistantCompletion]:
+        body = deepcopy_minimal(
+            {
+                "assistant_id": assistant_id,
+                "model": model,
+                "messages": messages,
+                "stream": stream,
+                "conversation_id": conversation_id,
+                "attachments": attachments,
+                "metadata": metadata,
+                "request_id": request_id,
+                "user_id": user_id,
+            }
+        )
+        return self._post(
+            "/assistant",
+            body=maybe_transform(body, assistant_create_params.AssistantParameters),
+            options=make_request_options(
+                extra_headers=extra_headers, extra_body=extra_body, timeout=timeout
+            ),
+            cast_type=AssistantCompletion,
+            stream=stream or True,
+            stream_cls=StreamResponse[AssistantCompletion],
+        )
+
+    def query_support(
+            self,
+            *,
+            assistant_id_list: List[str] = None,
+            request_id: str = None,
+            user_id: str = None,
+            extra_headers: Headers | None = None,
+            extra_body: Body | None = None,
+            timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
+    ) -> AssistantSupportResp:
+        body = deepcopy_minimal(
+            {
+                "assistant_id_list": assistant_id_list,
+                "request_id": request_id,
+                "user_id": user_id,
+            }
+        )
+        return self._post(
+            "/assistant/list",
+            body=body,
+            options=make_request_options(
+                extra_headers=extra_headers, extra_body=extra_body, timeout=timeout
+            ),
+            cast_type=AssistantSupportResp,
+        )
+
+    def query_conversation_usage(
+            self,
+            assistant_id: str,
+            page: int = 1,
+            page_size: int = 10,
+            *,
+            request_id: str = None,
+            user_id: str = None,
+            extra_headers: Headers | None = None,
+            extra_body: Body | None = None,
+            timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
+    ) -> ConversationUsageListResp:
+        body = deepcopy_minimal(
+            {
+                "assistant_id": assistant_id,
+                "page": page,
+                "page_size": page_size,
+                "request_id": request_id,
+                "user_id": user_id,
+            }
+        )
+        return self._post(
+            "/assistant/conversation/list",
+
+            body=maybe_transform(body, assistant_conversation_params.ConversationParameters),
+            options=make_request_options(
+                extra_headers=extra_headers, extra_body=extra_body, timeout=timeout
+            ),
+            cast_type=ConversationUsageListResp,
+        )