huimv-employment/doc/灵活用工项目后端与智能体接口文档.md

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

基本规则

• 基础路径：所有接口均以 /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:

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

• Response: Server-Sent Events（SSE）流

参数详细解释

---

参考 Python 脚本

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

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")

使用方法：

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：

{
  "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）

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()