rebuild

Author: jayscoder

pydify/README.md

@@ -5,12 +5,12 @@
 包括对话、文本生成、工作流、文件上传等。
 """
 
-from .agent import AgentClient
-from .chatbot import ChatbotClient
-from .chatflow import ChatflowClient
+from .agent import AgentClient, AgentEvent
+from .chatbot import ChatbotClient, ChatbotEvent
+from .chatflow import ChatflowClient, ChatflowEvent
 from .common import DifyBaseClient, DifyType
-from .text_generation import TextGenerationClient
-from .workflow import WorkflowClient
+from .text_generation import TextGenerationClient, TextGenerationEvent
+from .workflow import WorkflowClient, WorkflowEvent
 
 
 def create_client(type: str, base_url: str, api_key: str) -> DifyBaseClient:
@@ -36,4 +36,7 @@ def create_client(type: str, base_url: str, api_key: str) -> DifyBaseClient:
     "AgentClient",
     "TextGenerationClient",
     "create_client",
+    "DifyType",
+    "ChatbotEvent",
+    "WorkflowEvent",
 ]

pydify/__init__.py

@@ -11,17 +11,38 @@
 from typing import Any, BinaryIO, Dict, Generator, List, Optional, Tuple, Union
 
 from .common import DifyBaseClient, DifyType
+from .chatbot import ChatbotClient
+class AgentEvent:
+    """事件类型枚举
 
+    定义了Dify Agent API中可能的事件类型，用于处理流式响应中的事件。
+    """
 
-class AgentClient(DifyBaseClient):
+    MESSAGE = "message"  # LLM返回文本块事件，包含分块输出的文本内容
+    AGENT_MESSAGE = "agent_message"  # Agent模式下返回文本块事件，包含分块输出的文本内容
+    AGENT_THOUGHT = "agent_thought"  # Agent思考步骤事件，包含工具调用相关信息
+    MESSAGE_FILE = "message_file"  # 文件事件，表示有新文件需要展示
+    MESSAGE_END = "message_end"  # 消息结束事件，表示流式返回结束
+    TTS_MESSAGE = "tts_message"  # TTS音频流事件，包含base64编码的音频块
+    TTS_MESSAGE_END = "tts_message_end"  # TTS音频流结束事件
+    MESSAGE_REPLACE = "message_replace"  # 消息内容替换事件，用于内容审查后的替换
+    ERROR = "error"  # 流式输出过程中出现的异常事件
+    PING = "ping"  # 保持连接存活的ping事件，每10秒一次
+    
+
+class AgentClient(ChatbotClient):
     """Dify Agent应用客户端类。
 
     提供与Dify Agent应用API交互的方法，包括发送消息、获取历史消息、管理会话、
     上传文件、语音转文字、文字转语音等功能。Agent应用支持迭代式规划推理和自主工具调用。
     """
 
     type = DifyType.Agent
-
+    
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.task_id = None
+        
     def send_message(
         self,
         query: str,
@@ -32,7 +53,7 @@ def send_message(
         files: List[Dict[str, Any]] = None,
         auto_generate_name: bool = True,
         **kwargs,  # 添加kwargs参数，用于接收额外的请求参数
-    ) -> Generator[Dict[str, Any], None, None]:
+    ) -> Generator[Dict[str, Any], None, None] | Dict[str, Any]:
         """
         发送对话消息，创建会话消息。在Agent模式下，只支持streaming流式模式。
 
@@ -55,7 +76,7 @@ def send_message(
         """
         if response_mode != "streaming":
             raise ValueError("Agent mode only supports streaming response mode")
-
+        
         payload = {
             "query": query,
             "user": user,
@@ -74,7 +95,7 @@ def send_message(
 
         return self.post_stream(endpoint, json_data=payload, **kwargs)  # 传递额外参数
 
-    def stop_response(self, task_id: str, user: str) -> Dict[str, Any]:
+    def stop_task(self, task_id: str, user: str) -> Dict[str, Any]:
         """
         停止正在进行的响应流。此方法仅在流式模式下有效。
 
@@ -95,7 +116,7 @@ def stop_response(self, task_id: str, user: str) -> Dict[str, Any]:
         endpoint = f"chat-messages/{task_id}/stop"
         payload = {"user": user}
         return self.post(endpoint, json_data=payload)
-
+    
     def get_meta(self) -> Dict[str, Any]:
         """
         获取应用Meta信息，用于获取工具icon等。
@@ -107,171 +128,4 @@ def get_meta(self) -> Dict[str, Any]:
             requests.HTTPError: 当API请求失败时
         """
         return self.get("meta")
-
-    def process_streaming_response(
-        self,
-        stream_generator: Generator[Dict[str, Any], None, None],
-        handle_message=None,
-        handle_agent_message=None,
-        handle_agent_thought=None,
-        handle_message_file=None,
-        handle_message_end=None,
-        handle_tts_message=None,
-        handle_tts_message_end=None,
-        handle_message_replace=None,
-        handle_error=None,
-        handle_ping=None,
-        break_on_error=True,
-    ) -> Dict[str, Any]:
-        """
-        处理流式响应，调用相应事件处理器。
-
-        Args:
-            stream_generator: 流式响应生成器
-            handle_message: LLM返回文本块事件处理函数
-            handle_agent_message: Agent模式下返回文本块事件处理函数
-            handle_agent_thought: Agent模式下思考步骤事件处理函数
-            handle_message_file: 文件事件处理函数
-            handle_message_end: 消息结束事件处理函数
-            handle_tts_message: TTS音频流事件处理函数
-            handle_tts_message_end: TTS音频流结束事件处理函数
-            handle_message_replace: 消息内容替换事件处理函数
-            handle_error: 错误事件处理函数
-            handle_ping: ping事件处理函数
-            break_on_error: 当遇到错误时是否中断处理，默认为True
-
-        Returns:
-            Dict[str, Any]: 处理结果，包含消息ID、会话ID等信息
-
-        示例:
-            ```python
-            def on_agent_message(chunk):
-                # 打印Agent返回的文本块
-                print(f"{chunk['answer']}")
-
-            def on_agent_thought(chunk):
-                print(f"Agent思考: {chunk['thought']}")
-                print(f"使用工具: {chunk['tool']}")
-                print(f"工具输入: {chunk['tool_input']}")
-                print(f"观察结果: {chunk['observation']}")
-
-            def on_message_end(chunk):
-                print(f"消息结束: ID={chunk['message_id']}")
-
-            client = AgentClient(api_key)
-            stream = client.send_message(
-                query="帮我分析最近的股市走势",
-                user="user123"
-            )
-            result = client.process_streaming_response(
-                stream,
-                handle_agent_message=on_agent_message,
-                handle_agent_thought=on_agent_thought,
-                handle_message_end=on_message_end
-            )
-            ```
-        """
-        result = {"agent_thoughts": []}
-        answer_chunks = []
-
-        for chunk in stream_generator:
-            event = chunk.get("event")
-
-            if event == "message" and handle_message:
-                handle_message(chunk)
-                # 累积回答内容
-                if "answer" in chunk:
-                    answer_chunks.append(chunk["answer"])
-                # 保存消息和会话ID
-                if "message_id" in chunk and not result.get("message_id"):
-                    result["message_id"] = chunk["message_id"]
-                if "conversation_id" in chunk and not result.get("conversation_id"):
-                    result["conversation_id"] = chunk["conversation_id"]
-                if "task_id" in chunk and not result.get("task_id"):
-                    result["task_id"] = chunk["task_id"]
-
-            elif event == "agent_message" and handle_agent_message:
-                handle_agent_message(chunk)
-                # 累积回答内容
-                if "answer" in chunk:
-                    answer_chunks.append(chunk["answer"])
-                # 保存消息和会话ID
-                if "message_id" in chunk and not result.get("message_id"):
-                    result["message_id"] = chunk["message_id"]
-                if "conversation_id" in chunk and not result.get("conversation_id"):
-                    result["conversation_id"] = chunk["conversation_id"]
-                if "task_id" in chunk and not result.get("task_id"):
-                    result["task_id"] = chunk["task_id"]
-
-            elif event == "agent_thought" and handle_agent_thought:
-                if handle_agent_thought:
-                    handle_agent_thought(chunk)
-                # 保存Agent思考内容
-                thought_data = {
-                    "id": chunk.get("id"),
-                    "position": chunk.get("position"),
-                    "thought": chunk.get("thought"),
-                    "observation": chunk.get("observation"),
-                    "tool": chunk.get("tool"),
-                    "tool_input": chunk.get("tool_input"),
-                    "message_files": chunk.get("message_files", []),
-                    "created_at": chunk.get("created_at"),
-                }
-                result["agent_thoughts"].append(thought_data)
-
-            elif event == "message_file" and handle_message_file:
-                handle_message_file(chunk)
-                # 保存文件信息
-                if not result.get("files"):
-                    result["files"] = []
-                result["files"].append(
-                    {
-                        "id": chunk.get("id"),
-                        "type": chunk.get("type"),
-                        "belongs_to": chunk.get("belongs_to"),
-                        "url": chunk.get("url"),
-                    }
-                )
-
-            elif event == "message_end" and handle_message_end:
-                if handle_message_end:
-                    handle_message_end(chunk)
-                # 保存元数据
-                if "metadata" in chunk:
-                    result["metadata"] = chunk["metadata"]
-                if "message_id" in chunk and not result.get("message_id"):
-                    result["message_id"] = chunk["message_id"]
-                if "conversation_id" in chunk and not result.get("conversation_id"):
-                    result["conversation_id"] = chunk["conversation_id"]
-
-            elif event == "tts_message" and handle_tts_message:
-                handle_tts_message(chunk)
-
-            elif event == "tts_message_end" and handle_tts_message_end:
-                handle_tts_message_end(chunk)
-
-            elif event == "message_replace" and handle_message_replace:
-                handle_message_replace(chunk)
-                # 替换回答内容
-                if "answer" in chunk:
-                    answer_chunks = [chunk["answer"]]
-
-            elif event == "error" and handle_error:
-                handle_error(chunk)
-                if break_on_error:
-                    # 添加错误信息到结果中
-                    result["error"] = {
-                        "status": chunk.get("status"),
-                        "code": chunk.get("code"),
-                        "message": chunk.get("message"),
-                    }
-                    break
-
-            elif event == "ping" and handle_ping:
-                handle_ping(chunk)
-
-        # 合并所有回答块
-        if answer_chunks:
-            result["answer"] = "".join(answer_chunks)
-
-        return result
+