""" 人工回覆伺服器 - 偽裝成 OpenAI API 使用方式： 1. 執行此程式：python human_reply_server.py 2. 在 Open WebUI 設定中加入模型：http://localhost:8000/v1 3. 用戶發送訊息後，訪問 http://localhost:8000/admin 查看並回覆 """ from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, StreamingResponse from pydantic import BaseModel import asyncio import json import time from datetime import datetime from typing import Optional import uuid app = FastAPI() # 儲存待處理的對話 pending_conversations = {} # 儲存管理員的回覆 admin_replies = {} class Message(BaseModel): role: str content: str class ChatRequest(BaseModel): model: str messages: list[Message] stream: Optional[bool] = False @app.get("/") async def root(): return {"status": "Human Reply Server Running", "admin_panel": "/admin"} @app.get("/v1/models") async def list_models(): """模擬 OpenAI 的模型列表 API""" return { "object": "list", "data": [ { "id": "human-admin", "object": "model", "created": int(time.time()), "owned_by": "human-admin" } ] } @app.post("/v1/chat/completions") async def chat_completions(request: ChatRequest): """模擬 OpenAI 的聊天完成 API - 等待真人回覆""" # 生成對話 ID conversation_id = str(uuid.uuid4())[:8] # 取得最後一則用戶訊息 user_message = "" for msg in reversed(request.messages): if msg.role == "user": user_message = msg.content break # 儲存對話資訊 pending_conversations[conversation_id] = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "user_message": user_message, "full_history": [{"role": msg.role, "content": msg.content} for msg in request.messages] } print(f"\n[新訊息 {conversation_id}] {user_message}") print(f"→ 等待管理員回覆，請訪問: http://localhost:8000/admin") # 等待管理員回覆 (最多等待 10 分鐘) timeout = 600 elapsed = 0 while conversation_id not in admin_replies and elapsed < timeout: await asyncio.sleep(1) elapsed += 1 # 取得管理員回覆 if conversation_id in admin_replies: reply_content = admin_replies[conversation_id] del admin_replies[conversation_id] del pending_conversations[conversation_id] else: reply_content = "抱歉，管理員暫時無法回覆，請稍後再試。" # 根據是否需要串流回傳 if request.stream: return StreamingResponse( stream_response(reply_content), media_type="text/event-stream" ) else: return { "id": f"chatcmpl-{conversation_id}", "object": "chat.completion", "created": int(time.time()), "model": request.model, "choices": [{ "index": 0, "message": { "role": "assistant", "content": reply_content }, "finish_reason": "stop" }], "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 } } async def stream_response(content: str): """串流方式回傳回覆""" # 開始標記 yield f"data: {json.dumps({'choices': [{'delta': {'role': 'assistant'}, 'index': 0}]})}\n\n" # 逐字回傳 for char in content: chunk = { "choices": [{ "delta": {"content": char}, "index": 0 }] } yield f"data: {json.dumps(chunk)}\n\n" await asyncio.sleep(0.01) # 結束標記 yield f"data: {json.dumps({'choices': [{'delta': {}, 'index': 0, 'finish_reason': 'stop'}]})}\n\n" yield "data: [DONE]\n\n" @app.get("/admin", response_class=HTMLResponse) async def admin_panel(): """管理員後台 - 查看並回覆訊息""" # 生成待處理對話的 HTML conversations_html = "" if pending_conversations: for conv_id, conv_data in pending_conversations.items(): conversations_html += f"""
對話 ID: {conv_id} {conv_data['timestamp']}
用戶: {conv_data['user_message']}
""" else: conversations_html = '

目前沒有待處理的訊息

' html = f""" 管理員後台 - 人工回覆系統

🎯 管理員後台

人工回覆系統 - 待處理訊息數量: {len(pending_conversations)}

{conversations_html}
""" return html @app.post("/admin/reply") async def admin_reply(request: Request): """接收管理員的回覆""" data = await request.json() conversation_id = data.get("conversation_id") reply = data.get("reply") if conversation_id and reply: admin_replies[conversation_id] = reply print(f"[管理員回覆 {conversation_id}] {reply}") return {"status": "success"} return {"status": "error", "message": "Missing conversation_id or reply"} if __name__ == "__main__": import uvicorn print("=" * 60) print("🚀 人工回覆伺服器啟動中...") print("=" * 60) print("📌 管理員後台: http://localhost:8000/admin") print("📌 API 端點: http://localhost:8000/v1") print("=" * 60) print("\n在 Open WebUI 中設定:") print("1. 進入 Settings → Connections") print("2. 添加 OpenAI API:") print(" - API Base URL: http://localhost:8000/v1") print(" - API Key: 隨便填 (例如: sk-human)") print("3. 模型名稱會顯示為: human-admin") print("=" * 60) uvicorn.run(app, host="0.0.0.0", port=8000)