接上一个实践，需求经理希望用户检索的时候，在5类可供选择的同时也可以自定义字段。借助大模型的语义理解来做此内容匹配引擎，可以在数据库文本数据中智能识别与特定关键词相关联的内容。

组长要我说给每一条工单都打上true/flase的标记来表示是否为用户输入的字段，所以还是需要用大模型遍历每一条工单的content。

1数据库提取文本

cur = conn.cursor()
cur.execute('SELECT id, content FROM system.work_order WHERE "isDanger" = true;')

游标 (cursor)：用于执行 SQL 命令和获取结果。

SQL 查询：从 system.work_order 表中选择 id 和 content 字段，且只选择 isDanger 为 true 的记录。

def fetch_contents():
    """
    核心数据库查询函数
    功能：从PostgreSQL中查询所有标记为危险的内容
    特点：使用参数化查询防止SQL注入，完整的异常处理
    """
    try:
        # 建立数据库连接 - 使用psycopg2连接PostgreSQL
        conn = psycopg2.connect(
            host=DB_HOST, port=DB_PORT, 
            dbname=DB_NAME, user=DB_USER, 
            password=DB_PASSWORD
        )
        # 创建游标对象执行SQL查询
        cur = conn.cursor()
        
        # 执行参数化查询 - 只获取isDanger为true的记录
        cur.execute('SELECT id, content FROM system.work_order WHERE "isDanger" = true;')
        
        # 获取所有查询结果
        rows = cur.fetchall()
        
        # 关闭游标和连接，释放资源
        cur.close()
        conn.close()
        
        # 将查询结果转换为字典列表，便于后续处理
        return [{"id": r[0], "content": r[1], "isDanger": True} for r in rows]
    
    except Exception as e:
        # 异常处理：记录错误信息并返回空列表，保证系统稳定性
        print("数据库操作异常:", e)
        return []

由于数据库表格里面所存储的content记录格式很不统一，所以需要进行数据清洗

def preprocess_contents(contents):
    processed_texts = []
    for item in contents:
        # 清理文本：去除多余空格、特殊字符等
        clean_text = item['content'].strip()
        
        # 截断或分块（如果文本过长）
        if len(clean_text) > 4000:  # 模型有token限制
            clean_text = clean_text[:4000] + "..."
        
        processed_texts.append({
            'id': item['id'],
            'text': clean_text
        })
    return processed_texts

2提示词工程

因为需要逐行判断，但是如果一个一个传输的话，我会被限制封号。所以在总体量不大的情况下我将提取出来的一整个发送，每个工单内容用id号进行区分。

def call_llm_batch(words: List[str], contents: List[dict]):
    """调用大语言模型进行批量分析"""
    prompt = {
        "role": "user",
        "content": (
            "用户输入的词语: " + ", ".join(words) + "\n\n"
            "请逐条判定下面的 content 是否与用户输入的任意词语有关联。"
            "输出 JSON 数组，每条为 {\"id\": 内容id, \"match\": true/false}。\n\n"
            "待判定内容如下：\n" +
            "\n".join([f"{c['id']}: {c['content']}" for c in contents])
        )
    }
    # 发送请求并处理响应

3输出标准化

大模型具有”创造力”，为保证后续逻辑接收统一格式，所以需要进行标准化

def clean_llm_output(text: str):
    """
    核心输出清理函数
    功能：处理LLM返回的各种格式，提取标准JSON
    特点：支持多种JSON格式，强大的容错能力
    """
    try:
        # 移除常见的JSON标记符号 ```json 和 ```
        text = re.sub(r"^```json\s*|\s*```$", "", text.strip(), flags=re.DOTALL).strip()

        # 处理标准JSON数组格式 [{"id": 1, "match": true}, ...]
        if text.startswith("[") and text.endswith("]"):
            return json.loads(text)

        # 处理对象包装格式 {"results": [{"id": 1, "match": true}]}
        if text.startswith("{") and text.endswith("}"):
            obj = json.loads(text)
            if "results" in obj:
                return obj["results"]
            return obj

        # 使用正则表达式提取可能嵌入在文本中的JSON数组
        match = re.search(r"\[.*\]", text, re.DOTALL)
        if match:
            return json.loads(match.group(0))

    except Exception as e:
        # 解析失败时的异常处理
        print("clean_llm_output 出错:", e)

    # 所有解析尝试都失败时返回空列表
    return []

4整体流程

1. 启动服务：通过run_all.py启动FastAPI服务(端口8000)

2. 接收请求：用户通过HTTP请求传入查询词语

3. 处理查询参数：将逗号分隔的词语转换为列表

4. 获取数据：从数据库查询所有isDanger=True的工单内容

5. 分批处理：将工单内容分成每批20条，调用大模型API

6. 调用大模型：

   • 构建提示词，包含用户输入的词语和待判定的工单内容

   • 发送请求到火山方舟API

   • 接收模型返回的JSON格式响应

7. 清理响应：处理模型返回的可能包含```json标记的字符串，提取标准JSON

8. 处理异常：对网络请求、JSON解析等可能出现的异常进行捕获和处理

9. 返回结果：

   • /match接口：返回所有工单的判定结果(包括match为true和false的)

   • /match_true接口：只返回match为true且isDanger为true的结果

10. 格式化响应：将结果封装为JSON格式返回给客户端

"""
service.py
核心业务逻辑：连接数据库，取出 work_order 表中的 content；
调用大模型逐条判定是否与用户输入的词语有关联；
增加异常捕获和日志打印，避免 Internal Server Error。
"""

import psycopg2
import requests
import json
from typing import List
import re

# ================== 数据库配置 ==================
DB_HOST = "localhost"
DB_PORT = 5432
DB_NAME = "wujiang_system"
DB_USER = "postgres"
DB_PASSWORD = "12345"

# ================== 大模型配置 ==================
API_KEY = "9b6b8bd0-    4e53-ae63-91f89dbbddb8"
API_URL = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

BATCH_SIZE = 20  # 批次大小


def fetch_contents():
    """只取 isDanger=True 的工单"""
    try:
        conn = psycopg2.connect(
            host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASSWORD
        )
        cur = conn.cursor()
        cur.execute('SELECT id, content FROM system.work_order WHERE "isDanger" = true;')
        rows = cur.fetchall()
        cur.close()
        conn.close()
        return [{"id": r[0], "content": r[1], "isDanger": True} for r in rows]
    except Exception as e:
        print("数据库操作异常:", e)
        return []


def clean_llm_output(text: str):
    """
    清理大模型输出，确保能转成 JSON。
    处理情况：
    1. 去掉 ```json ... ``` 包裹
    2. 自动提取 JSON 数组部分
    """
    try:
        text = re.sub(r"^```json\s*|\s*```$", "", text.strip(), flags=re.DOTALL).strip()

        if text.startswith("[") and text.endswith("]"):
            return json.loads(text)

        if text.startswith("{") and text.endswith("}"):
            obj = json.loads(text)
            if "results" in obj:
                return obj["results"]
            return obj

        match = re.search(r"\[.*\]", text, re.DOTALL)
        if match:
            return json.loads(match.group(0))

    except Exception as e:
        print("clean_llm_output 出错:", e)

    return []


def call_llm_batch(words: List[str], contents: List[dict]):
    """调用 deepseek-v3-250324 模型，一次处理一个批次"""
    try:
        prompt = {
            "role": "user",
            "content": (
                "用户输入的词语: " + ", ".join(words) + "\n\n"
                "请逐条判定下面的 content 是否与用户输入的任意词语有关联。"
                "输出 JSON 数组，每条为 {\"id\": 内容id, \"match\": true/false}。\n\n"
                "待判定内容如下：\n" +
                "\n".join([f"{c['id']}: {c['content']}" for c in contents])
            )
        }

        payload = {
            "model": "deepseek-v3-250324",
            "messages": [prompt],
            "temperature": 0
        }

        resp = requests.post(
            API_URL,
            headers=HEADERS,
            json=payload,
            timeout=60
        )
        resp.raise_for_status()
        data = resp.json()
        text = data["choices"][0]["message"]["content"]

        result = clean_llm_output(text)
        if not isinstance(result, list):
            print("LLM 返回非列表:", result)
            result = []

        if not result:
            result = [{"id": c["id"], "match": False} for c in contents]

        # 补充 isDanger=True 字段
        id_map = {c["id"]: c for c in contents}
        for item in result:
            item["isDanger"] = True
            if "content" not in item and item["id"] in id_map:
                item["content"] = id_map[item["id"]]["content"]

        return result

    except requests.exceptions.RequestException as e:
        print("LLM 请求异常:", e)
        return [{"id": c["id"], "match": False, "isDanger": True} for c in contents]
    except Exception as e:
        print("LLM 调用未知异常:", e)
        return [{"id": c["id"], "match": False, "isDanger": True} for c in contents]


def match_contents(words: List[str]):
    """核心逻辑：分批调用大模型，解析 JSON"""
    try:
        contents = fetch_contents()
        results = []

        for i in range(0, len(contents), BATCH_SIZE):
            batch = contents[i: i + BATCH_SIZE]
            batch_result = call_llm_batch(words, batch)
            results.extend(batch_result)

        return results
    except Exception as e:
        print("match_contents 出现异常:", e)
        return []

"""
clean_llm_json.py
功能：处理 LLM 返回的带 ```json 的字符串，生成标准 JSON。
"""

import json

def clean_llm_output(raw_str):
    """
    清理 LLM 返回的字符串，返回 Python 对象。

    参数:
        raw_str (str): LLM 原始输出，可能包含 ```json ... ```

    返回:
        list/dict: 解析后的 JSON 对象
    """
    if not raw_str:
        return []

    # 去掉首尾空格
    clean_str = raw_str.strip()

    # 去掉 ```json 前缀
    if clean_str.startswith("```json"):
        clean_str = clean_str[len("```json"):].strip()

    # 去掉 ``` 后缀
    if clean_str.endswith("```"):
        clean_str = clean_str[:-3].strip()

    try:
        # 转成 Python 对象
        result = json.loads(clean_str)
        return result
    except json.JSONDecodeError:
        # 解析失败返回空列表
        return []

"""
main.py
FastAPI 服务，提供两个接口：
1. /match       返回全部 content 判定结果（true + false）
2. /match_true  只返回 match 为 true 且 isDanger 为 true 的结果
依赖 service.py 提供的 match_contents 方法。
"""

from fastapi import FastAPI, Query
from service import match_contents

app = FastAPI(title="内容关联系统")

@app.get("/match")
def match_endpoint(
    query_words: str = Query(..., description="用户输入的词语，多个词用逗号分隔")
):
    # 用户输入词语转成列表
    words = [w.strip() for w in query_words.split(",") if w.strip()]
    results = match_contents(words)
    return {"results": results}


@app.get("/match_true")
def match_true_endpoint(
    query_words: str = Query(..., description="用户输入的词语，多个词用逗号分隔")
):
    words = [w.strip() for w in query_words.split(",") if w.strip()]
    all_results = match_contents(words)

    # 只保留 match=True 的条目（isDanger 已经是 True）
    true_results = [r for r in all_results if r.get("match")]

    return {"results": true_results}

"""
run_all.py
统一启动 FastAPI 服务（main.py），提供 /match 和 /match_true 两个接口。
"""

import uvicorn

if __name__ == "__main__":
    # 启动服务，host=0.0.0.0 可外网访问，port=8000 可修改
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

'''
http://127.0.0.1:8000/match?query_words=自杀,举报
http://127.0.0.1:8000/match_true?query_words=自杀,举报
'''

Apifox访问