# 灵活用工项目后端与智能体接口文档

## **基本规则**

- **基础路径**：所有接口均以 `/api/` 为前缀

- Base URL: http://harrison1\.iask\.in:9188

- **主要聊天接口**：`POST /api/console/chat`

- **认证方式**：

    - 本地（localhost）自动免认证

    - 远程访问需 `Authorization: Bearer <token>` \(内网不要求\)

    - 必需 Header：`X-Agent-Id`（默认 `default`）

---

## **所有接口总结**

|方法|接口|描述|认证要求|主要用途|
|---|---|---|---|---|
|**POST**|`/api/console/chat`|**核心聊天接口**（流式 SSE）|本地免认证 / Token|发送消息、维持多轮对话|
|**GET**|`/api/version`|获取版本信息|无|健康检查|

---

### **详细接口文档**

#### **聊天接口**

##### 说明

- **Endpoint**: `POST /api/console/chat`

- **Headers**:

    - `Content-Type: application/json`

    - `X-Agent-Id: default`（必需）

    - `Authorization: Bearer <token>`（远程必需，内网无需）

- **Request Body**:

    ```JSON
    {
      "input": [
        {
          "role": "user",
          "content": [
            { "type": "text", "text": "消息内容" }
          ]
        }
      ],
      "session_id": "会话ID（用于多轮）",
      "user_id": "用户ID",
      "channel": "console"
    }
    ```

- **Response**: Server\-Sent Events（SSE）流

##### **参数详细解释**

---

##### 参考 Python 脚本

使用 `requests` \+ `sseclient` 处理流式响应：

```Python
import requests
import json

class QwenPawChat:
    def __init__(self, base_url: str = "http://localhost:9188", agent_id: str = "default"):
        self.base_url = base_url.rstrip('/')
        self.agent_id = agent_id
        self.token = None  # 如果启用认证，在这里设置

    def set_token(self, token: str):
        self.token = token

    def chat(self, message: str, session_id: str = "default-session", user_id: str = "test-user"):
        headers = {
            "Content-Type": "application/json",
            "X-Agent-Id": self.agent_id
        }
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"

        payload = {
            "input": [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": message}
                    ]
                }
            ],
            "session_id": session_id,
            "user_id": user_id,
            "channel": "console"
        }

        print(f"👤 用户: {message}\n🤖 Assistant: ", end='', flush=True)

        try:
            with requests.post(
                f"{self.base_url}/api/console/chat",
                headers=headers,
                json=payload,
                stream=True,
                timeout=180
            ) as response:
                response.raise_for_status()

                full_response = ""
                for line in response.iter_lines():
                    if line and line.startswith(b'data: '):
                        try:
                            data = json.loads(line[6:])
                            if data.get('error'):
                                print("\n❌ 错误:", data['error'].get('message'))
                                break

                            # 提取 assistant 的文本内容并实时打印
                            if data.get('output'):
                                for item in data['output']:
                                    if item.get('role') == 'assistant':
                                        for content in item.get('content', []):
                                            if content.get('type') == 'text':
                                                text = content.get('text', '')
                                                print(text, end='', flush=True)
                                                full_response += text
                        except:
                            continue
                print("\n")  # 结束一行
                return full_response

        except Exception as e:
            print(f"\n❌ 请求失败: {e}")
            return None


# ==================== 使用示例 ====================
if __name__ == "__main__":
    chat = QwenPawChat(base_url="http://localhost:9188")

    # 多轮对话测试
    chat.chat("你好，请介绍一下你自己", session_id="demo-chat-001")
    chat.chat("我们刚才在聊什么？", session_id="demo-chat-001")   # 测试记忆
    chat.chat("用中文写一段关于AI未来的短文", session_id="demo-chat-002")
```

**使用方法**：

```Bash
pip install requests
python test_qwenpaw.py
```

---

### RAG 接口文档

**基本信息**

- **名称**：RAG 知识库

- **版本**：2\.0\.0

- **前缀**：`/api/rag`

- **功能**：文件上传、文档管理、语义检索、对话自动注入 RAG 上下文

#### **所有接口列表**

|方法|接口|描述|主要参数|备注|
|---|---|---|---|---|
|**GET**|`/api/rag/health`|健康检查|\-||
|**POST**|`/api/rag/files`|上传文件到知识库（异步处理）|`file` \(UploadFile\)|返回 track\_id|
|**PUT**|`/api/rag/files/{file_id}`|更新文件|`file_id`, `file`||
|**GET**|`/api/rag/files`|分页列出所有文档|`page`, `page_size`, `sort_field`, `sort_direction`||
|**GET**|`/api/rag/files/{file_id}`|查询处理状态（task\_id 或 track\_id）|`file_id`||
|**DELETE**|`/api/rag/files/{file_id}`|删除单个文档|`file_id`||
|**POST**|`/api/rag/retrieve`|语义检索知识|`query`, `top_k`, `mode`, `include_references`||
|**POST**|`/api/rag/insert-text`|插入单条文本|`text`, `file_source`||
|**POST**|`/api/rag/reprocess-failed`|重新处理失败文档|\-||

#### **核心接口详细说明**

##### **检索接口**

- **Endpoint**：POST /api/rag/retrieve

- **Headers**：Content\-Type: application/json

- **Request Body**：

```JSON
{
  "query": "查询内容",
  "top_k": 5,
  "mode": "mix",
  "include_references": true
}
```

- **Response**：JSON 对象

**参数详细解释**：

---

##### **上传文件接口**

- **Endpoint**：POST /api/rag/files

- **Headers**：Content\-Type: multipart/form\-data

- **Request**：file（UploadFile）

- **Response**：返回 track\_id，需后续轮询状态

---

##### **查询处理状态**

- **Endpoint**：GET /api/rag/files/\{file\_id\}

- **参数**：file\_id（track\_id 或 doc\_id）

- **Response**：包含 status（processing / processed / failed）

---

#### 测试程序（Python）

```Python
import requests
import json
import time
from pathlib import Path

class LightRAGPluginTester:
    def __init__(self, base_url: str = "http://localhost:9188", timeout: int = 180):
        self.base_url = base_url.rstrip('/')
        self.prefix = "/api/rag"
        self.session = requests.Session()
        self.timeout = timeout
        self.track_ids = []

    def _url(self, path: str):
        return f"{self.base_url}{self.prefix}{path}"

    def health(self): 
        r = self.session.get(self._url("/health"))
        print("🔍 [GET] /health →", r.json())

    def wait_for_completion(self, track_id: str):
        print(f"⏳ 等待处理 (track={track_id})...")
        start = time.time()
        while time.time() - start < self.timeout:
            try:
                r = self.session.get(self._url(f"/files/{track_id}"))
                data = r.json()
                # 安全获取 status
                status = data.get("status")
                if not status and data.get("documents"):
                    status = data["documents"][0].get("status")
                print(f"   [{int(time.time()-start):3}s] Status: {status}")

                if status == "processed":
                    print("✅ 处理完成！")
                    return True
                elif status == "failed":
                    print("❌ 处理失败")
                    return False
            except Exception as e:
                print("   查询异常:", str(e)[:100])
            time.sleep(4)
        print("⏰ 超时")
        return False

    def insert_text(self, text: str):
        print(f"\n📝 [POST] /insert-text")
        payload = {"text": text, "file_source": f"test_{int(time.time())}"}
        r = self.session.post(self._url("/insert-text"), json=payload)
        data = r.json()
        print("   →", data)
        track = data.get("data", {}).get("track_id")
        if track:
            self.track_ids.append(track)
            self.wait_for_completion(track)

    def delete_single(self, file_id: str):
        print(f"\n🗑️ [DELETE] /files/{file_id}")
        r = self.session.delete(self._url(f"/files/{file_id}"))
        print("   →", r.json())

    def retrieve(self, query: str):
        print(f"\n🔍 [POST] /retrieve: {query}")
        payload = {"query": query, "top_k": 5, "mode": "mix", "include_references": True}
        r = self.session.post(self._url("/retrieve"), json=payload)
        print(json.dumps(r.json(), ensure_ascii=False, indent=2)[:700] + "...")

    def run_full_test(self):
        print("🚀 开始 **完整接口** 测试...\n")
        self.health()

        self.insert_text("2026年8月识：人工智能在精准农业中可实现作物病害早期预警、智能灌溉优化和产量预测。")
        self.insert_text("知识：消毒应优先使用碱性消毒剂，高温下病毒存活时间短。")

        self.retrieve("人工智能 精准农业")
        self.retrieve("口蹄疫 消毒剂")

        print("\n🧹 开始清理...")
        for tid in list(self.track_ids)[-4:]:
            self.delete_single(tid)

        print("\n🎉 **全接口测试完成！**")

if __name__ == "__main__":
    tester = LightRAGPluginTester(base_url="http://localhost:9188", timeout=180)
    tester.run_full_test()
```