仪评指标联分析模块

bbit_ai/app/llm/ticketLLM.py

@@ -0,0 +1,152 @@
+import json
+import re
+
+from langchain.schema import HumanMessage
+
+from config.llm import *
+
+
+def get_ticket_response(image_url: str):
+    # 构建 prompt
+    prompt_text = f"""
+你是一位专业的图像分析AI。你的任务是仔细分析提供的图片内容，并按JSON格式输出结果。
+
+## JSON输出结构及字段说明:
+# 蚕茧检测信息数据模型（ImageDescription）
+
+该模型用于描述蚕茧检测信息，每条记录包含以下字段：
+
+## 1. 含水率 (`moisture_content`)
+- **类型**：浮点数
+- **描述**：茧的含水量，单位为百分比（%）
+
+## 2. 下足茧重量 (`cocoon_weight`)
+- **类型**：浮点数
+- **描述**：下足茧的重量，单位为克（可带小数）
+
+## 3. 非好蛹粒数 (`defective_pupa_count`)
+- **类型**：整数
+- **描述**：不合格蛹的数量，即非好蛹的个数
+
+## 4. 鲜壳量 (`fresh_shell_weight`)
+- **类型**：浮点数
+- **描述**：鲜壳重量，单位为克（可带小数）
+
+## 5. 小样粒数 (`sample_count`)
+- **类型**：整数
+- **描述**：检测使用的小样数量，即用于检测的茧粒数
+
+## 6. 净重合计 (`net_weight_total`)
+- **类型**：浮点数
+- **描述**：所有样品的净重总和，单位为克
+
+## 7. 仪评人姓名 (`evaluator`)
+- **类型**：字符串
+- **描述**：进行仪器检测的人员姓名，可能为空
+
+## 8. 复核人员姓名 (`reviewer`)
+- **类型**：字符串
+- **描述**：复核人员姓名，可能为空
+
+---
+
+### 注意事项
+- 所有字段都是必填的（required），在 JSON 实例中必须提供值  
+- 浮点数字段可以包含小数，整数字段只能是整数  
+- `evaluator` 和 `reviewer` 可以为空字符串，但字段必须存在  
+
+最后，只输出严格的 JSON 格式，不要包含其他文字、markdown等内容。
+"""
+    messages = [
+        HumanMessage(
+            content=[
+                {"type": "text", "text": prompt_text},
+                {"type": "image_url", "image_url": {"url": image_url}},
+            ]
+        )
+    ]
+    # 直接调用模型
+    llmRes = llmVision.invoke(messages).content
+    # 去掉 ```json 和 ``` 包裹
+    cleaned = re.sub(r"^```json\s*|\s*```$", "", llmRes.strip())
+    # 解析 JSON
+    jsonRes = json.loads(cleaned)
+    jsonRes["barcode"] = decode_barcode(image_url)
+    return jsonRes
+
+
+import os
+import base64
+import tempfile
+import requests
+from pyzxing import BarCodeReader
+from PIL import Image
+from io import BytesIO
+from fastapi import UploadFile
+
+
+def decode_barcode(input_data) -> str:
+    """
+    自动识别输入类型并解析条码：
+    - URL 字符串
+    - Base64 字符串
+    - UploadFile / bytes / 文件对象
+    返回第一个条码的内容，解析失败返回空字符串
+    """
+    # ---------------- 输入处理 ----------------
+    img = None
+
+    # URL
+    if isinstance(input_data, str) and (
+        input_data.startswith("http://") or input_data.startswith("https://")
+    ):
+        response = requests.get(input_data)
+        response.raise_for_status()
+        img = Image.open(BytesIO(response.content))
+
+    # Base64 字符串
+    elif isinstance(input_data, str):
+        # 过滤 data URI 前缀
+        if "," in input_data:
+            input_data = input_data.split(",")[1]
+        try:
+            img_data = base64.b64decode(input_data)
+            img = Image.open(BytesIO(img_data))
+        except Exception:
+            raise ValueError("无法解析 Base64 字符串")
+
+    # UploadFile / bytes / 文件对象
+    elif isinstance(input_data, UploadFile):
+        content = input_data.file.read()
+        input_data.file.seek(0)
+        img = Image.open(BytesIO(content))
+    elif isinstance(input_data, bytes):
+        img = Image.open(BytesIO(input_data))
+    elif hasattr(input_data, "read"):  # 文件对象
+        content = input_data.read()
+        if hasattr(input_data, "seek"):
+            input_data.seek(0)
+        img = Image.open(BytesIO(content))
+    else:
+        raise ValueError("不支持的输入类型")
+
+    # ---------------- 临时文件处理 ----------------
+    tmp_fd, tmp_path = tempfile.mkstemp(suffix=".png")
+    try:
+        with open(tmp_path, "wb") as f:
+            img.save(f, format="PNG")
+
+        # ---------------- 调用 pyzxing ----------------
+        reader = BarCodeReader()
+        result = reader.decode(tmp_path)
+
+        if result:
+            parsed = result[0].get("parsed", "")
+            if isinstance(parsed, bytes):
+                parsed = parsed.decode("utf-8")
+            return parsed
+        return ""
+    finally:
+        os.close(tmp_fd)
+        if os.path.exists(tmp_path):
+            os.remove(tmp_path)