Update code

coze_api/.env.example

@@ -0,0 +1,9 @@
+# Coze API 配置
+# 在 https://www.coze.cn 的个人设置中获取 API Token
+COZE_API_TOKEN=your_api_token_here
+
+# Bot ID - 在 Bot 发布后可以获取
+COZE_BOT_ID=your_bot_id_here
+
+# API 基础地址（国内版用 coze.cn，国际版用 coze.com）
+COZE_API_BASE=https://api.coze.cn

coze_api/.gitignore

@@ -0,0 +1,3 @@
+.env
+__pycache__/
+*.pyc

coze_api/README.md

@@ -0,0 +1,28 @@
+# Coze 智能体 API 调用
+
+## 使用说明
+
+### 1. 安装依赖
+```bash
+pip install -r requirements.txt
+```
+
+### 2. 配置环境变量
+复制 `.env.example` 为 `.env`，填写你的 Coze API 凭证：
+```bash
+cp .env.example .env
+```
+
+需要填写：
+- `COZE_API_TOKEN` - Coze 平台的 API Token（在 Coze 平台 → 个人设置 → API Token 获取）
+- `COZE_BOT_ID` - Bot ID（在 Bot 发布后获取）
+
+### 3. 运行测试
+```bash
+python test_coze_api.py
+```
+
+## 文件说明
+- `coze_client.py` - Coze API 封装客户端
+- `test_coze_api.py` - 测试调用脚本
+- `.env.example` - 环境变量模板

coze_api/coze_client.py

@@ -0,0 +1,176 @@
+"""
+Coze 智能体 API 客户端封装
+支持调用 Coze 平台发布的 Bot 进行对话
+"""
+import os
+import json
+import uuid
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class CozeClient:
+    """Coze API 客户端"""
+
+    def __init__(self, api_token=None, bot_id=None, api_base=None):
+        self.api_token = api_token or os.getenv("COZE_API_TOKEN")
+        self.bot_id = bot_id or os.getenv("COZE_BOT_ID")
+        self.api_base = api_base or os.getenv("COZE_API_BASE", "https://api.coze.cn")
+
+        if not self.api_token:
+            raise ValueError("缺少 COZE_API_TOKEN，请在 .env 文件中配置")
+        if not self.bot_id:
+            raise ValueError("缺少 COZE_BOT_ID，请在 .env 文件中配置")
+
+        self.headers = {
+            "Authorization": f"Bearer {self.api_token}",
+            "Content-Type": "application/json",
+        }
+
+    def chat(self, message, user_id=None, conversation_id=None, stream=False, custom_variables=None):
+        """
+        发送消息给 Bot 并获取回复
+
+        Args:
+            message: 用户消息内容
+            user_id: 用户标识（可选，默认生成随机ID）
+            conversation_id: 会话ID（可选，用于多轮对话）
+            stream: 是否使用流式返回
+            custom_variables: 用户自定义变量，如 {"zishu": "200字以内", "fengge": "轻松活泼"}
+
+        Returns:
+            dict: API 响应结果
+        """
+        url = f"{self.api_base}/v3/chat"
+
+        if user_id is None:
+            user_id = str(uuid.uuid4().hex[:16])
+
+        payload = {
+            "bot_id": self.bot_id,
+            "user_id": user_id,
+            "stream": stream,
+            "auto_save_history": True,
+            "additional_messages": [
+                {
+                    "role": "user",
+                    "content": message,
+                    "content_type": "text",
+                }
+            ],
+        }
+
+        if custom_variables:
+            payload["custom_variables"] = custom_variables
+
+        if conversation_id:
+            payload["conversation_id"] = conversation_id
+
+        try:
+            response = requests.post(url, headers=self.headers, json=payload, timeout=60)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as e:
+            return {"error": str(e), "status_code": getattr(e.response, "status_code", None)}
+
+    def retrieve_chat(self, conversation_id, chat_id):
+        """
+        查询对话状态（非流式模式下需要轮询）
+
+        Args:
+            conversation_id: 会话ID
+            chat_id: 对话ID
+
+        Returns:
+            dict: 对话状态
+        """
+        url = f"{self.api_base}/v3/chat/retrieve"
+        params = {
+            "conversation_id": conversation_id,
+            "chat_id": chat_id,
+        }
+
+        try:
+            response = requests.get(url, headers=self.headers, params=params, timeout=30)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as e:
+            return {"error": str(e)}
+
+    def get_chat_messages(self, conversation_id, chat_id):
+        """
+        获取对话的消息列表
+
+        Args:
+            conversation_id: 会话ID
+            chat_id: 对话ID
+
+        Returns:
+            dict: 消息列表
+        """
+        url = f"{self.api_base}/v3/chat/message/list"
+        params = {
+            "conversation_id": conversation_id,
+            "chat_id": chat_id,
+        }
+
+        try:
+            response = requests.get(url, headers=self.headers, params=params, timeout=30)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as e:
+            return {"error": str(e)}
+
+    def chat_and_poll(self, message, user_id=None, custom_variables=None, poll_interval=2, max_wait=120):
+        """
+        发送消息并轮询等待结果（非流式模式的完整调用流程）
+
+        Args:
+            message: 用户消息
+            user_id: 用户标识
+            custom_variables: 用户自定义变量
+            poll_interval: 轮询间隔（秒）
+            max_wait: 最大等待时间（秒）
+
+        Returns:
+            str: Bot 的回复文本
+        """
+        import time
+
+        result = self.chat(message, user_id=user_id, stream=False, custom_variables=custom_variables)
+
+        if "error" in result:
+            return f"请求失败: {result['error']}"
+
+        data = result.get("data", {})
+        conversation_id = data.get("conversation_id")
+        chat_id = data.get("id")
+
+        if not conversation_id or not chat_id:
+            return f"响应异常: {json.dumps(result, ensure_ascii=False, indent=2)}"
+
+        # 轮询等待完成
+        elapsed = 0
+        while elapsed < max_wait:
+            time.sleep(poll_interval)
+            elapsed += poll_interval
+
+            status_result = self.retrieve_chat(conversation_id, chat_id)
+            status = status_result.get("data", {}).get("status")
+
+            if status == "completed":
+                # 获取消息
+                messages_result = self.get_chat_messages(conversation_id, chat_id)
+                messages = messages_result.get("data", [])
+                # 找到 assistant 的回复
+                for msg in messages:
+                    if msg.get("role") == "assistant" and msg.get("type") == "answer":
+                        return msg.get("content", "")
+                return f"未找到回复消息: {json.dumps(messages, ensure_ascii=False, indent=2)}"
+
+            elif status == "failed":
+                return f"对话失败: {json.dumps(status_result, ensure_ascii=False, indent=2)}"
+
+        return "等待超时，请稍后重试"

coze_api/requirements.txt

@@ -0,0 +1,2 @@
+requests>=2.31.0
+python-dotenv>=1.0.0

coze_api/test_coze_api.py

@@ -0,0 +1,90 @@
+"""
+Coze 智能体 API 调用测试脚本
+运行前请确保已配置 .env 文件
+"""
+from coze_client import CozeClient
+
+
+def test_basic_chat():
+    """基础对话测试"""
+    print("=" * 50)
+    print("🤖 大沃智能助手 - API 调用测试")
+    print("=" * 50)
+
+    try:
+        client = CozeClient()
+    except ValueError as e:
+        print(f"\n❌ 初始化失败: {e}")
+        print("请先配置 .env 文件（参考 .env.example）")
+        return
+
+    # 测试用例
+    test_messages = [
+        {
+            "message": "帮我写一段招商文案",
+            "variables": {"zishu": "200字以内", "fengge": "专业正式"},
+            "description": "测试招商Agent",
+        },
+        {
+            "message": "早上好",
+            "variables": {"zishu": "100字以内", "fengge": "轻松活泼"},
+            "description": "测试问候Agent",
+        },
+        {
+            "message": "介绍一下公司",
+            "variables": {"zishu": "300字以内", "fengge": "专业正式"},
+            "description": "测试公司介绍Agent",
+        },
+    ]
+
+    for i, test in enumerate(test_messages, 1):
+        print(f"\n--- 测试 {i}: {test['description']} ---")
+        print(f"📝 输入: {test['message']}")
+        print(f"⚙️  参数: 字数={test['variables']['zishu']}, 风格={test['variables']['fengge']}")
+        print("⏳ 等待回复中...")
+
+        reply = client.chat_and_poll(
+            message=test["message"],
+            custom_variables=test["variables"],
+        )
+
+        print(f"💬 回复:\n{reply}")
+        print()
+
+
+def test_interactive():
+    """交互式对话测试"""
+    print("=" * 50)
+    print("🤖 大沃智能助手 - 交互模式")
+    print("输入 'quit' 退出")
+    print("=" * 50)
+
+    try:
+        client = CozeClient()
+    except ValueError as e:
+        print(f"\n❌ 初始化失败: {e}")
+        return
+
+    while True:
+        user_input = input("\n你: ").strip()
+        if user_input.lower() in ("quit", "exit", "q"):
+            print("再见！")
+            break
+        if not user_input:
+            continue
+
+        print("⏳ 思考中...")
+        reply = client.chat_and_poll(
+            message=user_input,
+            custom_variables={"zishu": "200字以内", "fengge": "轻松活泼"},
+        )
+        print(f"\n助手: {reply}")
+
+
+if __name__ == "__main__":
+    import sys
+
+    if len(sys.argv) > 1 and sys.argv[1] == "--interactive":
+        test_interactive()
+    else:
+        test_basic_chat()