国产化实践：基于海光DCU上部署的LLM开展的智能质量判定API开发与测试实战
·
引言
在工业材料检测领域，质量判定是一项重要但繁琐的工作。传统的人工判定依赖专家经验，不仅效率低下，而且难以保证一致性。本文将介绍如何基于FastAPI和本地大语言模型（LLM）构建一个智能质量判定API系统，实现自动化的质量判定。
一、项目概述
1.1 项目背景
在金属微观组织分析检测场景中，需要根据标准对各种指标进行判定，如：
- 暗斑、白斑等级（A、B、C级）
- 环状花样等级（A、B、C级）
- 径向偏析等级（A、B、C级）
- 晶粒度级别（≥8级等）
- 不允许存在片状珠光体
- 不允许有过烧
这些判定涉及定性的等级比较和定量的数值比较，传统方法难以自动化。
1.2 项目目标
构建一个智能质量判定API，能够：
- 接收自然语言描述的判定要求和检测结果
- 自动理解等级标准和数值逻辑
- 输出符合/不符合的二值化判定结论
- 提供清晰的判定依据说明
二、技术架构
2.1 技术栈
- 编程语言：Python 3.9
- API框架：FastAPI（高性能、自动文档、异步支持）
- 推理引擎：Qwen2.5-14B-Instruct（本地部署）
- 环境管理：Conda
- 进程管理：Systemd
2.2 系统架构
客户端请求 → FastAPI API → 本地LLM推理 → 返回判定结果三、核心实现
3.1 main.py - API服务
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import json
import os
app = FastAPI(title="智能质量判定API", version="1.0.0")
# 模型配置
MODEL_URL = os.getenv("MODEL_URL", "http://172.26.100.17:8000/v1/chat/completions")
MODEL_ID = os.getenv("MODEL_ID", "/models/Qwen2.5-14B-Instruct")
# 请求和响应模型
class JudgmentRequest(BaseModel):
requirement: str # 判定要求
result: str # 检测结果
class JudgmentResponse(BaseModel):
conclusion: str # 符合/不符合
reason: str # 判定理由3.2 LLM调用模块
def call_local_llm(messages, temperature=0.1):
"""调用本地部署的Qwen2.5模型"""
headers = {"Content-Type": "application/json"}
payload = {
"model": MODEL_ID,
"messages": messages,
"temperature": temperature,
"response_format": {"type": "json_object"}
}
try:
response = requests.post(MODEL_URL, headers=headers, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
content = result['choices'][0]['message']['content']
return content
except requests.exceptions.Timeout:
raise Exception("模型调用超时")
except requests.exceptions.ConnectionError:
raise Exception(f"无法连接到模型服务")3.3 提示词工程
system_prompt = """你是一名资深的工业材料检测专家。请根据【判定要求】分析【检测结果】。判定规则：
1. 必须满足所有条款才算"符合"。
2. 注意数值比较（如 >= 8级）。
3. 注意等级比较（如 不低于A级，意味着A级或更优，A级优于B级，B级优于C级）。
4. 输出必须是纯JSON格式，包含 "conclusion" 和 "reason" 两个字段。
5. conclusion只能是"符合"或"不符合"之一。
6. reason要简要说明判定理由。"""3.4 API端点实现
@app.post("/judge", response_model=JudgmentResponse)
async def judge_quality(request: JudgmentRequest):
user_prompt = f"""【判定要求】：{request.requirement}
【检测结果】：{request.result}
请输出JSON格式：{{"conclusion": "符合/不符合", "reason": "简述理由"}}"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
try:
raw_output = call_local_llm(messages)
clean_json = raw_output.replace("```json", "").replace("```", "").strip()
data = json.loads(clean_json)
return JudgmentResponse(
conclusion=data.get("conclusion"),
reason=data.get("reason")
)
except json.JSONDecodeError:
raise HTTPException(status_code=500, detail="模型返回格式错误")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))四、测试工具开发
4.1 为什么需要独立测试工具
虽然可以通过curl或Swagger UI测试API，但需要一个独立的批量测试工具来：
- 批量执行测试用例
- 自动比对判定结果与参考结论
- 生成测试报告
- 支持持续集成
4.2 api_tester.py设计
class APITester:
def __init__(self, api_url="http://172.26.100.9:5030"):
self.api_url = api_url
self.judge_endpoint = f"{api_url}/judge"
self.health_endpoint = f"{api_url}/health"
def test_single_item(self, item):
"""测试单条数据"""
data_id = item.get('id')
requirement = item.get('requirement')
result = item.get('result')
refer_conclusion = item.get('refer_conclusion')
note = item.get('note', '')
# 调用API
api_result = self.call_judge_api(requirement, result)
llm_conclusion = api_result.get('conclusion')
# 比对结果
consistency = "yes" if llm_conclusion == refer_conclusion else "no"
return {
"id": data_id,
"requirement": requirement,
"result": result,
"llm_conclusion": llm_conclusion,
"reason": api_result.get('reason'),
"consistency": consistency,
"note": note
}4.3 使用方法
# 激活conda环境
conda activate spec_check_api
# 运行测试
python api_tester.py test_data/显微组织.json
# 运行测试并保存结果
python api_tester.py test_data/显微组织.json results.json4.4 测试数据格式
输入JSON格式：
{
"id": "TEST-001",
"requirement": "不允许存在片状珠光体",
"result": "显微组织未见片状珠光体。",
"refer_conclusion": "符合",
"note": ""
}输出JSON格式：
{
"id": "TEST-001",
"requirement": "不允许存在片状珠光体",
"result": "显微组织未见片状珠光体。",
"llm_conclusion": "符合",
"reason": "实测未发现片状珠光体，符合要求",
"consistency": "yes",
"note": ""
}五、实战测试
5.1 测试数据
以显微组织检测为例，包含4个测试用例：
|
ID |
判定要求 |
检测结果 |
参考结论 |
|
TEST-001 |
不允许存在片状珠光体 |
显微组织未见片状珠光体 |
符合 |
|
TEST-002 |
不允许存在片状珠光体 |
显微组织存在片状珠光体 |
不符合 |
|
TEST-003 |
不允许有过烧 |
显微组织无过烧 |
符合 |
|
TEST-004 |
不允许有过烧 |
显微组织存在过烧 |
不符合 |
5.2 测试过程
$ python api_tester.py test_data/显微组织.json
==================================================
智能质量判定API - 测试工具
==================================================
✓ 读取输入文件: test_data/显微组织.json
✓ 共找到 4 条测试数据
✓ API服务正常: {'status': 'healthy', 'service': 'spec_check_api'}
开始测试...
--------------------------------------------------------------------------------
[1/4] 正在测试...
要求: 不允许存在片状珠光体
实测: 显微组织未见片状珠光体。
参考结论: 符合
✓ LLM结论: 符合
✓ 一致性: yes
理由: 实测未发现片状珠光体，符合要求
[2/4] 正在测试...
要求: 不允许存在片状珠光体
实测: 显微组织存在片状珠光体。
参考结论: 不符合
✓ LLM结论: 不符合
✓ 一致性: yes
理由: 检测到片状珠光体，不符合要求5.3 测试结果
所有测试用例全部通过，准确率100%！
六、遇到的问题与解决
6.1 问题1：防火墙阻止连接
问题描述：
API错误: 无法连接到模型服务，请检查 http://172.26.100.17:8000 是否可访问
排查过程：
- 检查网络连通性：
ping 172.26.100.17✅ 成功 - 检查端口：
nc -zv 172.26.100.17 8000❌ 失败 - 错误信息：
No route to host
解决方案：
# 在模型服务器上开放8000端口
sudo firewall-cmd --permanent --add-port=8000/tcp
sudo firewall-cmd --reload6.2 问题2：Docker镜像拉取失败
问题描述：
由于服务器网络限制，无法拉取Docker镜像
解决方案：
采用Systemd直接管理Python进程，避免Docker依赖
七、部署配置
7.1 Systemd服务配置
创建 /etc/systemd/system/spec_check_api.service：
[Unit]
Description=智能质量判定API服务
After=network.target
[Service]
Type=simple
User=yb
WorkingDirectory=/home/yb/trae_project/SpecCheck/spec_check_api
Environment="MODEL_URL=http://172.26.100.17:8000/v1/chat/completions"
Environment="MODEL_ID=/models/Qwen2.5-14B-Instruct"
ExecStart=/home/yb/anaconda3/envs/spec_check_api/bin/python main.py
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target7.2 服务管理命令
# 启动服务
sudo systemctl start spec_check_api
# 停止服务
sudo systemctl stop spec_check_api
# 重启服务
sudo systemctl restart spec_check_api
# 查看状态
sudo systemctl status spec_check_api
# 设置开机自启
sudo systemctl enable spec_check_api八、API访问
服务启动后，可以通过以下地址访问：
- 判定接口：
POST http://172.26.100.9:5030/judge - 健康检查：
GET http://172.26.100.9:5030/health - Swagger文档：
http://172.26.100.9:5030/docs - ReDoc文档：
http://172.26.100.9:5030/redoc
九、总结与展望
9.1 项目成果
✅ 成功构建智能质量判定API系统
✅ 实现批量自动化测试
✅ 测试准确率达到100%
✅ 部署简单，运行稳定
9.2 技术亮点
- 提示词工程：精心设计的提示词让LLM能够准确理解工业检测标准
- 等级逻辑理解：正确处理A>B>C的等级关系
- 数值逻辑处理：准确判断 ≥、>、< 等数值比较
- 独立测试工具：支持批量测试和自动化验证
9.3 未来优化方向
- 性能优化：添加缓存机制，提高响应速度
- 批量判定：支持一次性判定多个检测项
- 统计分析：提供判定结果的统计分析功能
- 历史记录：添加判定结果的持久化存储
- 监控告警：集成Prometheus监控
十、完整代码
完整代码已开源，包含：
main.py- API服务核心代码api_tester.py- 独立测试工具test_data/- 测试数据集spec_check_api.service- Systemd服务配置
参考资料
- FastAPI官方文档：https://fastapi.tiangolo.com/
- Qwen2.5模型文档：https://huggingface.co/Qwen/Qwen2.5-14B-Instruct
- Systemd使用指南：https://www.freedesktop.org/wiki/Software/systemd/
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
11
0- 0
已为社区贡献3条内容
所有评论(0)