From abd4c3ce77d7f7d3523f54dfe78291016d5cf499 Mon Sep 17 00:00:00 2001 From: ZF sun <34314687@qq.com> Date: Wed, 22 Apr 2026 16:26:07 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E6=95=B0=E5=AD=97?= =?UTF-8?q?=E4=BA=BA=E6=9C=8D=E5=8A=A1=E5=8F=8A=E7=9B=B8=E5=85=B3=E9=83=A8?= =?UTF-8?q?=E7=BD=B2=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- digital-human.py | 374 +++++++++++++------------------- digital-human2.py | 258 ++++++++++++++++++++++ downloads/digital-human.py | 176 +++++++++++++++ downloads/digital-human.service | 11 + downloads/digital-human.sh | 33 +++ 5 files changed, 624 insertions(+), 228 deletions(-) create mode 100644 digital-human2.py create mode 100644 downloads/digital-human.py create mode 100644 downloads/digital-human.service create mode 100644 downloads/digital-human.sh diff --git a/digital-human.py b/digital-human.py index 7c69a1b..53e1be9 100644 --- a/digital-human.py +++ b/digital-human.py @@ -1,258 +1,176 @@ -""" -数字人对接中转服务 ------------------ -作为客户端与 Dify LLM 平台之间的中间转发层，处理请求路由、 -认证转发及 Redis 缓存，提升响应效率并保护后端 API Key。 - -环境变量： - DIFY_URL - Dify 服务地址，如 https://atwdify.antiwearvalve.com/v1 - DIFY_API_KEY - Dify API 密钥，如 app-PDkSHiHZsZIluoZF7N2OlNZM - REDIS_HOST - Redis 主机地址，默认 localhost - REDIS_PORT - Redis 端口，默认 6379 -""" - -import os import json -import hashlib -import logging -from contextlib import asynccontextmanager -from typing import Any, Dict, Optional - -import httpx import redis -from redis.exceptions import ConnectionError as RedisConnectionError -from fastapi import FastAPI, HTTPException, Request -from fastapi.responses import JSONResponse +import uuid +import requests +from fastapi.responses import StreamingResponse +from fastapi import FastAPI, Query, Request +from pydantic import BaseModel +from typing import Optional, List -# ────────────────────────────────────────────────────────────── -# 日志配置 -# ────────────────────────────────────────────────────────────── -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s - %(message)s", -) -logger = logging.getLogger("digital-human") - -# ────────────────────────────────────────────────────────────── -# 应用配置（从环境变量读取） -# ────────────────────────────────────────────────────────────── -DIFY_URL: str = os.getenv("DIFY_URL", "") -DIFY_API_KEY: str = os.getenv("DIFY_API_KEY", "") -REDIS_HOST: str = os.getenv("REDIS_HOST", "localhost") -REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379")) - -# 验证必填配置 -if not DIFY_URL or not DIFY_API_KEY: - logger.warning( - "DIFY_URL 或 DIFY_API_KEY 未配置，" - "部分功能可能不可用。请通过环境变量设置。" - ) - -# ────────────────────────────────────────────────────────────── -# Redis 客户端（懒加载） -# ────────────────────────────────────────────────────────────── -_redis_client: Optional[redis.Redis] = None +app = FastAPI() +# 连接到Redis服务器（默认端口是6379） +redis_instance = redis.Redis(host='localhost', port=6379, db=0) +mock_time_created = 20230512084843 +mock_id1 = "131424542124" +mock_id2 = "1231414141541252525" +mock_user = "abc-123" +dify_url = "http:///v1/chat-messages" +dify_api_key = 'Bearer ' -def get_redis() -> redis.Redis: - """获取 Redis 客户端实例（延迟初始化）。""" - global _redis_client - if _redis_client is None: - _redis_client = redis.Redis( - host=REDIS_HOST, - port=REDIS_PORT, - decode_responses=True, - socket_connect_timeout=5, - ) - return _redis_client +class Message(BaseModel): + content: str -def get_cache_key(payload: Dict[str, Any]) -> str: - """根据请求体生成 Redis 缓存键。""" - raw = json.dumps(payload, sort_keys=True) - return f"dh:cache:{hashlib.md5(raw.encode()).hexdigest()}" +class Resquest_Body(BaseModel): + messages: List[Message] = [] + app_id: str + user: str + session_id: str + is_stream: bool -# ────────────────────────────────────────────────────────────── -# FastAPI 应用生命周期 -# ────────────────────────────────────────────────────────────── -@asynccontextmanager -async def lifespan(app: FastAPI): - """应用启动 / 关闭钩子。""" - logger.info("数字人对接服务启动中...") - logger.info(f" Dify URL : {DIFY_URL}") - logger.info(f" Redis : {REDIS_HOST}:{REDIS_PORT}") - - # 尝试连接 Redis（仅记录，不阻塞启动） - try: - get_redis().ping() - logger.info("Redis 连接成功") - except RedisConnectionError as e: - logger.warning(f"Redis 连接失败（服务将继续运行）: {e}") - - yield - - logger.info("数字人对接服务已关闭") - - -app = FastAPI( - title="数字人对接中转服务", - version="1.0.0", - description="对接 Dify LLM 平台的数字人应用中转服务", - lifespan=lifespan, -) - - -# ────────────────────────────────────────────────────────────── -# 健康检查 -# ────────────────────────────────────────────────────────────── -@app.get("/") -async def root(): - """服务健康检查端点。""" - return { - "status": "ok", - "service": "digital-human-relay", - "version": "1.0.0", +# 对接dify，非流式传输 +def request_dify_api(question, request, conversation_id): + # 目标URL + url = dify_url + headers_json = { + 'Content-Type': 'application/json', + 'Authorization': dify_api_key + } + # 请求的数据 + data = { + "inputs": {}, + "query": question, + "response_mode": "blocking", + "user": mock_user } + # 发送POST请求 + response = requests.post(url, data=json.dumps(data), headers=headers_json) + data = json.loads(response) + return data["answer"] -@app.get("/health") -async def health_check(): - """详细健康检查（包含 Redis 状态）。""" - redis_ok = False - try: - get_redis().ping() - redis_ok = True - except Exception: - pass - return { - "status": "healthy" if redis_ok else "degraded", - "redis": "connected" if redis_ok else "disconnected", - "dify_configured": bool(DIFY_URL and DIFY_API_KEY), +# 对接dify，流式传输 +async def generate_streaming_messages(question, request, session_id, conversation_id): + url = dify_url + headers_json = { + 'Content-Type': 'application/json', + 'Authorization': dify_api_key } - -# ────────────────────────────────────────────────────────────── -# Dify 响应格式转 OpenAI 兼容格式 -# ────────────────────────────────────────────────────────────── -def dify_to_openai(dify_resp: Dict[str, Any]) -> Dict[str, Any]: - """ - 将 Dify /chat-messages 的响应转换为 OpenAI /chat/completions 兼容格式。 - Dify 响应示例: { answer: "...", conversation_id: "...", task_id: "..." } - OpenAI 目标格式: { id, model, choices: [{ message: { role: "assistant", content: "..." } }] } - """ - return { - "id": f"chatcmpl-{dify_resp.get('task_id', 'unknown')}", - "object": "chat.completion", - "created": 0, - "model": "dify", - "choices": [ - { - "index": 0, - "message": { - "role": "assistant", - "content": dify_resp.get("answer", ""), - }, - "finish_reason": "stop", - } - ], - "usage": { - "prompt_tokens": 0, - "completion_tokens": 0, - "total_tokens": 0, - }, + # 请求的数据 + request_data = { + "inputs": {}, + "query": question, + "conversation_id": conversation_id, + "response_mode": "streaming", + "user": mock_user } + r = requests.post(url, json=request_data, headers=headers_json, verify=False, stream=True) + if r.status_code != 200: + raise Exception(r.text) + data = "" -# ────────────────────────────────────────────────────────────── -# 核心转发接口 -# ────────────────────────────────────────────────────────────── -@app.post("/v1/chat/completions") -async def chat_completions(request: Request): - """ - 接收 OpenAI 兼容格式请求，转换为 Dify /chat-messages 格式后转发。 - OpenAI 格式: { model, messages: [{role, content}], stream } - Dify 格式: { query, user, response_mode, inputs } - """ - body = await request.json() - - # ── 格式转换：OpenAI -> Dify ────────────────────────────── - # 提取用户最新消息 - user_content = "" - for msg in reversed(body.get("messages", [])): - if msg.get("role") == "user": - user_content = msg.get("content", "") + # 在这里，我们使用了 iter_lines 方法来逐行读取响应体 + for line in r.iter_lines(): + if await request.is_disconnected(): + # 客户端已主动断开，中止流式输出，避免不必要的问题，打印报错信息方便定位 + print("# 客户端已主动断开，中止流式输出") break - if not user_content: - raise HTTPException(status_code=400, detail="未找到用户消息") + # 将每一行解码为字符串 + line = line.decode('utf-8').strip() - # 生成或复用 user 标识（用于 Dify 会话上下文） - dify_user = body.get("user", "anonymous") - response_mode = "blocking" if not body.get("stream", False) else "streaming" + # 接下来我们要处理三种不同的情况： + # 1. 如果当前行是空行，则表明前一个数据块已接收完毕（即前文提到的，通过两个换行符结束数据块传输），我们可以对该数据块进行反序列化，并打印出对应的 content 内容； + # 2. 如果当前行为非空行，且以 data: 开头，则表明这是一个数据块传输的开始，我们去除 data: 前缀后，首先判断是否是结束符 [DONE]，如果不是，将数据内容保存到 data 变量； + # 3. 如果当前行为非空行，但不以 data: 开头，则表明当前行仍然归属上一个正在传输的数据块，我们将当前行的内容追加到 data 变量尾部； - # 构建 Dify 请求体 - dify_payload = { - "query": user_content, - "user": dify_user, - "response_mode": response_mode, - "inputs": body.get("inputs", {}), - } + if len(line) == 0: + if data: + try: + chunk = json.loads(data) + except: + continue - # 若应用要求 inputs 字段但为空，补上空对象 - if "inputs" not in dify_payload or dify_payload["inputs"] is None: - dify_payload["inputs"] = {} + if chunk["event"] != "message": + continue + # 如果redis里未缓存该session_id，则使用reids关联session_id和conversation_id，保证后续实现多轮对话 + if not redis_instance.exists(session_id): + conversiation_id = str(chunk["conversation_id"]) + redis_instance.set(session_id, conversiation_id) - # ── 检查 Redis 缓存（基于 OpenAI 原始请求体）────────────── - cache_key = get_cache_key(body) - try: - cached = get_redis().get(cache_key) - if cached: - logger.info(f"缓存命中: {cache_key}") - return dify_to_openai(json.loads(cached)) - except RedisConnectionError: - logger.warning("Redis 不可用，跳过缓存读取") + # 设置redis中key（session_id）的超时时间，一天换算86400s + redis_instance.expire(session_id, 86400) + content = chunk["answer"] + new_data = { + "id": mock_id2, + "created": mock_time_created, + "choices": [{ + "message": { + "content": content + } + }] + } + yield f"data: {json.dumps(new_data, ensure_ascii=False)}\n\n" - # ── 转发到 Dify ───────────────────────────────────────── - headers = { - "Authorization": f"Bearer {DIFY_API_KEY}", - "Content-Type": "application/json", - } - dify_endpoint = f"{DIFY_URL.rstrip('/')}/chat-messages" + data = "" # 重置 data + elif line.startswith("data: "): + data = line.lstrip("data: ") - try: - async with httpx.AsyncClient(timeout=120.0) as client: - resp = await client.post(dify_endpoint, json=dify_payload, headers=headers) - resp.raise_for_status() - result = resp.json() - except httpx.HTTPStatusError as e: - logger.error(f"Dify 返回错误 {e.response.status_code}: {e.response.text}") - raise HTTPException( - status_code=e.response.status_code, - detail=f"Dify 请求失败: {e.response.text}", + # 当数据块内容为 [DONE] 时，则表明所有数据块已发送完毕，可断开网络连接 + if data == "[DONE]": + break + else: + # 仍然在追加内容时，为其添加一个换行符 + data = data + "\n" + line + + # 发送结束标记 + yield "data: [DONE]\n\n" + + +@app.post("/digital-human") +async def digital_human(body: Resquest_Body, request: Request): + question = body.messages[0].content + session_id = body.session_id + conversation_id = "" + + # 如果redis_instance里有缓存session_id，则说明为多轮对话，获取对应的dify中的多轮对话关键参数conversation_id + if redis_instance.exists(session_id): + # decode()用于将bytes转换为str，因为redis_instance存储的是bytes类型的数据。 + conversation_id = redis_instance.get(session_id).decode("utf-8") + + # 选择流式传输 + if body.is_stream: + return StreamingResponse( + generate_streaming_messages(question, request, session_id, conversation_id), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + } ) - except httpx.RequestError as e: - logger.error(f"无法连接 Dify 服务: {e}") - raise HTTPException(status_code=502, detail=f"无法连接 Dify 服务: {e}") - # ── 写入缓存（存 Dify 原始格式）─────────────────────────── - try: - get_redis().setex(cache_key, 300, json.dumps(result)) - except RedisConnectionError: - pass - - # ── 响应转 OpenAI 兼容格式 ───────────────────────────────── - return dify_to_openai(result) + # 选择非流式传输 + answer = request_dify_api(question, request, conversation_id) + choices = [{ + "index": 0, + "message": { + "content": answer + } + }] + return { + "id": mock_id1, + "created": mock_time_created, + "choices": choices + } -# ────────────────────────────────────────────────────────────── -# 错误处理器 -# ────────────────────────────────────────────────────────────── -@app.exception_handler(Exception) -async def global_exception_handler(request: Request, exc: Exception): - logger.exception(f"未处理的异常: {exc}") - return JSONResponse( - status_code=500, - content={"error": "Internal Server Error", "detail": str(exc)}, - ) +# 启动应用 +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/digital-human2.py b/digital-human2.py new file mode 100644 index 0000000..7c69a1b --- /dev/null +++ b/digital-human2.py @@ -0,0 +1,258 @@ +""" +数字人对接中转服务 +----------------- +作为客户端与 Dify LLM 平台之间的中间转发层，处理请求路由、 +认证转发及 Redis 缓存，提升响应效率并保护后端 API Key。 + +环境变量： + DIFY_URL - Dify 服务地址，如 https://atwdify.antiwearvalve.com/v1 + DIFY_API_KEY - Dify API 密钥，如 app-PDkSHiHZsZIluoZF7N2OlNZM + REDIS_HOST - Redis 主机地址，默认 localhost + REDIS_PORT - Redis 端口，默认 6379 +""" + +import os +import json +import hashlib +import logging +from contextlib import asynccontextmanager +from typing import Any, Dict, Optional + +import httpx +import redis +from redis.exceptions import ConnectionError as RedisConnectionError +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import JSONResponse + +# ────────────────────────────────────────────────────────────── +# 日志配置 +# ────────────────────────────────────────────────────────────── +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s - %(message)s", +) +logger = logging.getLogger("digital-human") + +# ────────────────────────────────────────────────────────────── +# 应用配置（从环境变量读取） +# ────────────────────────────────────────────────────────────── +DIFY_URL: str = os.getenv("DIFY_URL", "") +DIFY_API_KEY: str = os.getenv("DIFY_API_KEY", "") +REDIS_HOST: str = os.getenv("REDIS_HOST", "localhost") +REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379")) + +# 验证必填配置 +if not DIFY_URL or not DIFY_API_KEY: + logger.warning( + "DIFY_URL 或 DIFY_API_KEY 未配置，" + "部分功能可能不可用。请通过环境变量设置。" + ) + +# ────────────────────────────────────────────────────────────── +# Redis 客户端（懒加载） +# ────────────────────────────────────────────────────────────── +_redis_client: Optional[redis.Redis] = None + + +def get_redis() -> redis.Redis: + """获取 Redis 客户端实例（延迟初始化）。""" + global _redis_client + if _redis_client is None: + _redis_client = redis.Redis( + host=REDIS_HOST, + port=REDIS_PORT, + decode_responses=True, + socket_connect_timeout=5, + ) + return _redis_client + + +def get_cache_key(payload: Dict[str, Any]) -> str: + """根据请求体生成 Redis 缓存键。""" + raw = json.dumps(payload, sort_keys=True) + return f"dh:cache:{hashlib.md5(raw.encode()).hexdigest()}" + + +# ────────────────────────────────────────────────────────────── +# FastAPI 应用生命周期 +# ────────────────────────────────────────────────────────────── +@asynccontextmanager +async def lifespan(app: FastAPI): + """应用启动 / 关闭钩子。""" + logger.info("数字人对接服务启动中...") + logger.info(f" Dify URL : {DIFY_URL}") + logger.info(f" Redis : {REDIS_HOST}:{REDIS_PORT}") + + # 尝试连接 Redis（仅记录，不阻塞启动） + try: + get_redis().ping() + logger.info("Redis 连接成功") + except RedisConnectionError as e: + logger.warning(f"Redis 连接失败（服务将继续运行）: {e}") + + yield + + logger.info("数字人对接服务已关闭") + + +app = FastAPI( + title="数字人对接中转服务", + version="1.0.0", + description="对接 Dify LLM 平台的数字人应用中转服务", + lifespan=lifespan, +) + + +# ────────────────────────────────────────────────────────────── +# 健康检查 +# ────────────────────────────────────────────────────────────── +@app.get("/") +async def root(): + """服务健康检查端点。""" + return { + "status": "ok", + "service": "digital-human-relay", + "version": "1.0.0", + } + + +@app.get("/health") +async def health_check(): + """详细健康检查（包含 Redis 状态）。""" + redis_ok = False + try: + get_redis().ping() + redis_ok = True + except Exception: + pass + + return { + "status": "healthy" if redis_ok else "degraded", + "redis": "connected" if redis_ok else "disconnected", + "dify_configured": bool(DIFY_URL and DIFY_API_KEY), + } + + +# ────────────────────────────────────────────────────────────── +# Dify 响应格式转 OpenAI 兼容格式 +# ────────────────────────────────────────────────────────────── +def dify_to_openai(dify_resp: Dict[str, Any]) -> Dict[str, Any]: + """ + 将 Dify /chat-messages 的响应转换为 OpenAI /chat/completions 兼容格式。 + Dify 响应示例: { answer: "...", conversation_id: "...", task_id: "..." } + OpenAI 目标格式: { id, model, choices: [{ message: { role: "assistant", content: "..." } }] } + """ + return { + "id": f"chatcmpl-{dify_resp.get('task_id', 'unknown')}", + "object": "chat.completion", + "created": 0, + "model": "dify", + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": dify_resp.get("answer", ""), + }, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + }, + } + + +# ────────────────────────────────────────────────────────────── +# 核心转发接口 +# ────────────────────────────────────────────────────────────── +@app.post("/v1/chat/completions") +async def chat_completions(request: Request): + """ + 接收 OpenAI 兼容格式请求，转换为 Dify /chat-messages 格式后转发。 + OpenAI 格式: { model, messages: [{role, content}], stream } + Dify 格式: { query, user, response_mode, inputs } + """ + body = await request.json() + + # ── 格式转换：OpenAI -> Dify ────────────────────────────── + # 提取用户最新消息 + user_content = "" + for msg in reversed(body.get("messages", [])): + if msg.get("role") == "user": + user_content = msg.get("content", "") + break + + if not user_content: + raise HTTPException(status_code=400, detail="未找到用户消息") + + # 生成或复用 user 标识（用于 Dify 会话上下文） + dify_user = body.get("user", "anonymous") + response_mode = "blocking" if not body.get("stream", False) else "streaming" + + # 构建 Dify 请求体 + dify_payload = { + "query": user_content, + "user": dify_user, + "response_mode": response_mode, + "inputs": body.get("inputs", {}), + } + + # 若应用要求 inputs 字段但为空，补上空对象 + if "inputs" not in dify_payload or dify_payload["inputs"] is None: + dify_payload["inputs"] = {} + + # ── 检查 Redis 缓存（基于 OpenAI 原始请求体）────────────── + cache_key = get_cache_key(body) + try: + cached = get_redis().get(cache_key) + if cached: + logger.info(f"缓存命中: {cache_key}") + return dify_to_openai(json.loads(cached)) + except RedisConnectionError: + logger.warning("Redis 不可用，跳过缓存读取") + + # ── 转发到 Dify ───────────────────────────────────────── + headers = { + "Authorization": f"Bearer {DIFY_API_KEY}", + "Content-Type": "application/json", + } + dify_endpoint = f"{DIFY_URL.rstrip('/')}/chat-messages" + + try: + async with httpx.AsyncClient(timeout=120.0) as client: + resp = await client.post(dify_endpoint, json=dify_payload, headers=headers) + resp.raise_for_status() + result = resp.json() + except httpx.HTTPStatusError as e: + logger.error(f"Dify 返回错误 {e.response.status_code}: {e.response.text}") + raise HTTPException( + status_code=e.response.status_code, + detail=f"Dify 请求失败: {e.response.text}", + ) + except httpx.RequestError as e: + logger.error(f"无法连接 Dify 服务: {e}") + raise HTTPException(status_code=502, detail=f"无法连接 Dify 服务: {e}") + + # ── 写入缓存（存 Dify 原始格式）─────────────────────────── + try: + get_redis().setex(cache_key, 300, json.dumps(result)) + except RedisConnectionError: + pass + + # ── 响应转 OpenAI 兼容格式 ───────────────────────────────── + return dify_to_openai(result) + + +# ────────────────────────────────────────────────────────────── +# 错误处理器 +# ────────────────────────────────────────────────────────────── +@app.exception_handler(Exception) +async def global_exception_handler(request: Request, exc: Exception): + logger.exception(f"未处理的异常: {exc}") + return JSONResponse( + status_code=500, + content={"error": "Internal Server Error", "detail": str(exc)}, + ) diff --git a/downloads/digital-human.py b/downloads/digital-human.py new file mode 100644 index 0000000..53e1be9 --- /dev/null +++ b/downloads/digital-human.py @@ -0,0 +1,176 @@ +import json +import redis +import uuid +import requests +from fastapi.responses import StreamingResponse +from fastapi import FastAPI, Query, Request +from pydantic import BaseModel +from typing import Optional, List + +app = FastAPI() +# 连接到Redis服务器（默认端口是6379） +redis_instance = redis.Redis(host='localhost', port=6379, db=0) +mock_time_created = 20230512084843 +mock_id1 = "131424542124" +mock_id2 = "1231414141541252525" +mock_user = "abc-123" +dify_url = "http:///v1/chat-messages" +dify_api_key = 'Bearer ' + + +class Message(BaseModel): + content: str + + +class Resquest_Body(BaseModel): + messages: List[Message] = [] + app_id: str + user: str + session_id: str + is_stream: bool + + +# 对接dify，非流式传输 +def request_dify_api(question, request, conversation_id): + # 目标URL + url = dify_url + headers_json = { + 'Content-Type': 'application/json', + 'Authorization': dify_api_key + } + # 请求的数据 + data = { + "inputs": {}, + "query": question, + "response_mode": "blocking", + "user": mock_user + } + + # 发送POST请求 + response = requests.post(url, data=json.dumps(data), headers=headers_json) + data = json.loads(response) + return data["answer"] + + +# 对接dify，流式传输 +async def generate_streaming_messages(question, request, session_id, conversation_id): + url = dify_url + headers_json = { + 'Content-Type': 'application/json', + 'Authorization': dify_api_key + } + + # 请求的数据 + request_data = { + "inputs": {}, + "query": question, + "conversation_id": conversation_id, + "response_mode": "streaming", + "user": mock_user + } + r = requests.post(url, json=request_data, headers=headers_json, verify=False, stream=True) + if r.status_code != 200: + raise Exception(r.text) + + data = "" + + # 在这里，我们使用了 iter_lines 方法来逐行读取响应体 + for line in r.iter_lines(): + if await request.is_disconnected(): + # 客户端已主动断开，中止流式输出，避免不必要的问题，打印报错信息方便定位 + print("# 客户端已主动断开，中止流式输出") + break + + # 将每一行解码为字符串 + line = line.decode('utf-8').strip() + + # 接下来我们要处理三种不同的情况： + # 1. 如果当前行是空行，则表明前一个数据块已接收完毕（即前文提到的，通过两个换行符结束数据块传输），我们可以对该数据块进行反序列化，并打印出对应的 content 内容； + # 2. 如果当前行为非空行，且以 data: 开头，则表明这是一个数据块传输的开始，我们去除 data: 前缀后，首先判断是否是结束符 [DONE]，如果不是，将数据内容保存到 data 变量； + # 3. 如果当前行为非空行，但不以 data: 开头，则表明当前行仍然归属上一个正在传输的数据块，我们将当前行的内容追加到 data 变量尾部； + + if len(line) == 0: + if data: + try: + chunk = json.loads(data) + except: + continue + + if chunk["event"] != "message": + continue + # 如果redis里未缓存该session_id，则使用reids关联session_id和conversation_id，保证后续实现多轮对话 + if not redis_instance.exists(session_id): + conversiation_id = str(chunk["conversation_id"]) + redis_instance.set(session_id, conversiation_id) + + # 设置redis中key（session_id）的超时时间，一天换算86400s + redis_instance.expire(session_id, 86400) + content = chunk["answer"] + new_data = { + "id": mock_id2, + "created": mock_time_created, + "choices": [{ + "message": { + "content": content + } + }] + } + yield f"data: {json.dumps(new_data, ensure_ascii=False)}\n\n" + + data = "" # 重置 data + elif line.startswith("data: "): + data = line.lstrip("data: ") + + # 当数据块内容为 [DONE] 时，则表明所有数据块已发送完毕，可断开网络连接 + if data == "[DONE]": + break + else: + # 仍然在追加内容时，为其添加一个换行符 + data = data + "\n" + line + + # 发送结束标记 + yield "data: [DONE]\n\n" + + +@app.post("/digital-human") +async def digital_human(body: Resquest_Body, request: Request): + question = body.messages[0].content + session_id = body.session_id + conversation_id = "" + + # 如果redis_instance里有缓存session_id，则说明为多轮对话，获取对应的dify中的多轮对话关键参数conversation_id + if redis_instance.exists(session_id): + # decode()用于将bytes转换为str，因为redis_instance存储的是bytes类型的数据。 + conversation_id = redis_instance.get(session_id).decode("utf-8") + + # 选择流式传输 + if body.is_stream: + return StreamingResponse( + generate_streaming_messages(question, request, session_id, conversation_id), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + } + ) + + # 选择非流式传输 + answer = request_dify_api(question, request, conversation_id) + choices = [{ + "index": 0, + "message": { + "content": answer + } + }] + return { + "id": mock_id1, + "created": mock_time_created, + "choices": choices + } + + +# 启动应用 +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/downloads/digital-human.service b/downloads/digital-human.service new file mode 100644 index 0000000..ea341eb --- /dev/null +++ b/downloads/digital-human.service @@ -0,0 +1,11 @@ +[Unit] +Description=Digital Human Service +After=network.target + +[Service] +Type=simple +ExecStart=nohup python3 /digital-human.py & +Restart=on-failure + +[Install] +WantedBy=multi-user.target diff --git a/downloads/digital-human.sh b/downloads/digital-human.sh new file mode 100644 index 0000000..25f08a1 --- /dev/null +++ b/downloads/digital-human.sh @@ -0,0 +1,33 @@ +#!/bin/bash + +# 回到根路径 +cd / +rm -rf /digital-human.rar +rm -rf /digital-human.py +rm -rf /etc/systemd/system/digital-human.service + +# apt安装rar和redis-server +sudo apt update +sudo apt install rar redis-server + +#启动redis-server，并设置开机自启动 redis-server +sudo systemctl enable redis-server +sudo systemctl start redis-server + +# pip安装所需的包 +pip install redis fastapi uvicorn -i https://pypi.tuna.tsinghua.edu.cn/simple + +# 从obs桶获取转发服务代码 +wget https://documentation-samples.obs.cn-north-4.myhuaweicloud.com/solution-as-code-publicbucket/solution-as-code-moudle/building-a-dify-llm-application-development-platform/human/digital-human.rar +rar x digital-human.rar + +# 替换dify对应的url和api key +sed -i "s//$1/g" digital-human.py +sed -i "s//$2/g" digital-human.py + +#启动对接数字人的python服务，并设置开机自启动 +mv digital-human.service /etc/systemd/system/digital-human.service +sudo systemctl daemon-reload +sudo systemctl enable digital-human.service +sudo systemctl start digital-human.service +