内容
# 制造质量分析 NLP 代理
**制造意图路由、MCP 工具、证据和 FastAPI**
> 制造自然语言问题分类为意图，并执行所需分析工具，返回答案和证据的制造数据分析代理项目。
<p>
<img src="https://img.shields.io/badge/Python-3.11-3561D8?style=flat-square&logo=python&logoColor=white" alt="Python">
<img src="https://img.shields.io/badge/FastAPI-Agent%20API-21AFC4?style=flat-square&logo=fastapi&logoColor=white" alt="FastAPI">
<img src="https://img.shields.io/badge/LangGraph-Workflow-151F32?style=flat-square" alt="LangGraph">
<img src="https://img.shields.io/badge/FastMCP-4%20Tools-3561D8?style=flat-square" alt="FastMCP">
<img src="https://img.shields.io/badge/Docker-GitHub%20Actions-5F6675?style=flat-square&logo=docker&logoColor=white" alt="Docker and GitHub Actions">
</p>
---
## 为什么有这个项目
制造现场的问题通常以自然语言输入，但每个问题所需的数据和计算方法都不同。
- 不良率问题需要统计产量和不良数量。
- 传感器异常问题需要检查温度、振动和压力记录。
- 生产线状态问题需要总结生产效率和质量信息。
- 原因候选问题需要结合不良率和传感器异常信息，但不能确定为实际原因。
如果所有问题都用一个函数或固定答案处理，问题解释、数据计算和答案生成的责任就会混淆。此外，只返回答案很难验证结论是否正确。
本项目实现了以下流程：
```text
问题
→ 意图
→ 路由
→ 制造工具
→ 总结和证据
→ API 响应
```
`/agent/query` 在 LangGraph 内部直接调用 Python 工具函数。FastMCP 服务器以单独的工具接口暴露相同的 4 个分析功能，代理 API 不通过 MCP 客户端调用服务器。
---
## 项目概述
| 项目 | 内容 |
|---|---|
| **时间** | 2026.04–05 |
| **类型** | 个人项目 |
| **目标** | 将制造自然语言问题连接到分析功能，并返回答案和依据数据 |
| **范围** | 示例制造数据、基于规则的意图、路由器、LangGraph、4 个分析工具、FastMCP 服务器、FastAPI、JSONL 跟踪、PyTorch 模型端点、Docker、GitHub Actions、pytest |
| **NLP 范围** | 将自然语言问题的目的分类为 4 个意图并确定处理路径 |
| **技术** | Python、FastAPI、Pydantic、LangGraph、MCP FastMCP、pandas、SQLite、PyTorch、Docker、GitHub Actions |
| **实现结果** | 4 个意图、4 个代理工具、4 个 MCP 工具、3 个 API 端点、9 个核心测试 |
---
## 问题 → 实现 → 结果
| 问题 | 实现 | 结果 |
|---|---|---|
| 每个问题需要的计算方法不同 | 设置意图和路由器，分离功能工具 | 4 个问题类型连接到 4 个制造分析功能 |
| 仅凭答案难以验证结果 | 工具返回总结和证据行 | API 响应同时返回答案和计算依据 |
| 多个目的表达可能导致路由不同 | 原因、生产效率、不良率、传感器顺序优先级 | 代码明确复合问题的处理规则 |
| 代理 API 与 MCP 的关系可能混淆 | 代理直接调用 Python 工具，MCP 暴露相同函数 | 内部执行和外部工具接口责任分离 |
| 基于规则的工具和模型端点可能看起来像一个流程 | `/agent/query` 和 `/model/sensor-anomaly` 独立配置 | 数据聚合和 PyTorch 推理范围区分 |
| 执行环境不同，验证困难 | 生成示例数据、pytest、Docker、GitHub Actions | 提供本地、容器、CI 执行路径 |
---
## 系统概述
<img src="./docs/assets/mcp-system-overview.png" alt="制造质量分析 NLP 代理系统构成图" width="100%">
### 责任分离
| 层 | 责任 |
|---|---|
| **意图** | 将问题目的分类为 `defect_rate`、`machine_anomaly`、`line_performance`、`quality_issue_candidates` 之一 |
| **路由器** | 将意图连接到 Python 工具函数名称 |
| **制造工具** | 执行 CSV 加载、时间范围过滤、分组、聚合 |
| **答案构建器** | 使用工具总结作为最终答案 |
| **证据** | 结构化计算中使用的行和聚合结果 |
| **JSONL 跟踪** | 记录问题、意图、工具、证据计数、状态 |
| **FastMCP 服务器** | 以 MCP 接口形式暴露相同的 4 个工具功能 |
| **模型 API** | 返回 3 个传感器值的重构误差和异常状态 |
### 范围澄清
- 代理工具的基本数据路径是 CSV。
- SQLite 模式和加载器包含在内，但当前未连接到代理工具执行路径。
- 当前答案不由外部 LLM 生成，而是使用经过验证的工具总结。
- PyTorch 模型端点独立于代理的 `machine_anomaly` 工具。
---
## 实用评估标准
代理项目应综合评估 **路由、工具契约、证据、执行跟踪、接口责任分离**，而非单一的准确率指标。
| 实用视角 | 评估标准 | 项目中验证的依据 |
|---|---|---|
| **路由一致性** | 代表性问题是否正确连接到意图和工具 | 意图代理流程测试 4 个 |
| **工具结果稳定性** | 每个工具是否返回总结和证据 | 制造分析工具测试 4 个 |
| **响应验证可能性** | 答案是否可以通过证据重新验证 | `问题`、`意图`、`工具名称`、`答案`、`证据` 结构 |
| **工具契约明确性** | MCP 工具的名称、描述、输入值是否区分 | FastMCP 4 个工具、`days` 参数 |
| **执行跟踪** | 问题如何通过工具处理 | `logs/agent_trace.jsonl` |
| **责任分离** | 代理、MCP、模型的实际调用关系是否准确 | 代理直接调用、单独的 MCP 服务器、独立的模型端点 |
| **可重复性** | 示例数据和测试是否在相同条件下执行 | Python 3.11 CI、Docker、GitHub Actions |
| **范围说明** | 基于规则的分类和原因候选是否不过度夸大 | 无外部 LLM、候选表达、模型限制明确 |
> 当前公开测试包括代理流程 4 个、工具 4 个、PyTorch 服务 1 个。未表示 FastAPI 端点集成测试和 MCP 协议集成测试。
---
## 意图和工具映射
| 意图 | 代理工具 | MCP 工具 | 问题示例 |
|---|---|---|---|
| `defect_rate` | `get_defect_rate_by_line` | `defect_rate_by_line` | 最近 7 天不良率最高的生产线是？ |
| `machine_anomaly` | `detect_machine_anomalies` | `machine_anomalies` | 设备温度或振动是否异常高？ |
| `line_performance` | `summarize_line_performance` | `line_performance` | 请总结生产线效率和质量状态。 |
| `quality_issue_candidates` | `infer_quality_issue_candidates_tool` | `quality_issue_candidates` | 请提供质量问题原因候选及数据依据。 |
### 路由优先级
```text
01. 原因、候选、为什么 + 不良、质量、异常、defect
→ quality_issue_candidates
02. 生产效率、产量、生产线、总结、performance
→ line_performance
03. 不良、defect、不良率
→ defect_rate
04. 温度、振动、压力、传感器、异常、anomaly
→ machine_anomaly
05. 不匹配
→ line_performance
```
---
## 代理工作流程
<img src="./docs/assets/mcp-agent-workflow.png" alt="制造质量分析 NLP 代理工作流程图" width="100%">
| 阶段 | 实际处理 |
|---|---|
| **问题** | FastAPI 接收自然语言问题 |
| **分类** | `classify_intent()` 基于规则决定意图 |
| **路由** | `tool_map` 将意图连接到 Python 工具函数 |
| **分析** | 使用选定的工具执行 `days=7` 并聚合 CSV 数据 |
| **返回** | 总结作为答案，聚合行作为证据，并记录跟踪 |
> `PromptTemplate` 结构化答案策略，但当前不调用外部 LLM。如果没有证据，返回依据不足的提示。
---
## 技术细节
<details open>
<summary><b>01 | 代理状态和 LangGraph</b></summary>
<br>
```text
问题
→ 意图
→ 工具名称
→ 工具结果
→ 答案
→ 证据
```
```text
route_question
→ call_tool
→ build_answer
→ END
```
当前意图基于规则。LLM 不用于训练或生成最终数值和答案。
</details>
<details>
<summary><b>02 | 制造数据和工具</b></summary>
<br>
| 数据 | 目的 |
|---|---|
| `production_logs.csv` | 生产线产量和运行时间 |
| `quality_inspection.csv` | 检验数量、不良数量、不良类型 |
| `machine_sensor_logs.csv` | 温度、振动、压力传感器记录 |
```text
CSV 加载
→ 日期范围过滤
→ 生产线分组
→ 求和和平均聚合
→ 不良率计算
→ 基于阈值的传感器检查
→ 证据行构造
```
质量问题原因候选不确定实际原因。结合不良率和传感器异常，提出优先检查对象。
</details>
<details>
<summary><b>03 | FastMCP 服务器</b></summary>
<br>
```python
mcp = FastMCP("manufacturing-mcp-agent")
@mcp.tool()
def defect_rate_by_line(days: int = 7) -> dict:
return get_defect_rate_by_line(days=days)
@mcp.tool()
def machine_anomalies(days: int = 7) -> dict:
return detect_machine_anomalies(days=days)
@mcp.tool()
def line_performance(days: int = 7) -> dict:
return summarize_line_performance(days=days)
@mcp.tool()
def quality_issue_candidates(days: int = 7) -> dict:
return infer_quality_issue_candidates_tool(days=days)
```
MCP 层不重复实现分析逻辑，而是使用相同的 Python 工具函数。
</details>
<details>
<summary><b>04 | 独立的 PyTorch 模型 API</b></summary>
<br>
```text
温度、振动、压力
→ 线性(3 → 2)
→ ReLU
→ 线性(2 → 3)
→ 重构误差
→ 阈值 1000.0
→ is_anomaly
```
当前 `SensorAutoEncoder` 在没有训练权重的情况下创建。此端点展示最小的模型服务结构和响应模式，不代表已验证的异常检测性能。
</details>
<details>
<summary><b>05 | 验证和 CI</b></summary>
<br>
```text
tests/test_agent_flow.py 4 cases
tests/test_tools.py 4 cases
tests/test_torch_model.py 1 case
```
GitHub Actions 在 Python 3.11 环境中执行依赖安装、示例数据创建和 pytest。
</details>
---
## API
| 方法 | 端点 | 角色 |
|---|---|---|
| `GET` | `/` | 服务状态和端点指南 |
| `POST` | `/agent/query` | 将自然语言问题连接到意图和工具 |
| `POST` | `/model/sensor-anomaly` | 计算 3 个传感器值的重构误差 |
### POST `/agent/query`
```json
{
"问题": "最近 7 天不良率最高的生产线是？"
}
```
```json
{
"问题": "最近 7 天不良率最高的生产线是。",
"意图": "defect_rate",
"工具名称": "get_defect_rate_by_line",
"答案": "最近 7 天不良率最高的生产线是 LINE_C。",
"证据": [
{
"生产线 ID": "LINE_C",
"产量": 1234,
"不良数量": 54,
"平均不良率": 0.0439
}
]
}
```
### POST `/model/sensor-anomaly`
```json
{
"温度": 95.7,
"振动": 5.6,
"压力": 2.4
}
```
```json
{
"温度": 95.7,
"振动": 5.6,
"压力": 2.4,
"异常分数": 1023.5,
"阈值": 1000.0,
"是否为异常": true,
"模型": "SensorAutoEncoder",
"备注": "基于重构误差的异常结果"
}
```
> API 示例值用于说明响应结构。
---
## 运行和验证
```powershell
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -r .\requirements.txt
python .\scripts_generate_sample_data.py
python -m uvicorn app.main:app --reload
```
MCP 服务器:
```powershell
python -m app.mcp_server.server
```
测试:
```powershell
python -m pytest .\tests -q
```
Docker:
```powershell
docker compose up --build
```
---
## 项目结构
```text
manufacturing-mcp-agent/
├── .github/workflows/ci.yml
├── app/
│ ├── agent/
│ ├── db/
│ ├── mcp_server/
│ ├── models/
│ ├── services/
│ ├── config.py
│ └── main.py
├── data/
├── docs/assets/
├── langflow/
├── tests/
├── Dockerfile
├── docker-compose.yml
├── pytest.ini
├── requirements.txt
├── scripts_generate_sample_data.py
└── README.md
```
---
## 当前范围和限制
### 当前范围
- 公开的制造示例数据
- 基于规则的 4 个意图分类
- LangGraph 3 节点工作流程
- 基于 pandas 的制造数据聚合
- 答案和证据响应
- FastMCP 服务器 4 个工具
- JSONL 代理跟踪
- FastAPI 代理端点
- 独立的 PyTorch 模型端点
- Docker 和 GitHub Actions
### 限制
- 不调用外部 LLM
- 无意图评估数据集和混淆矩阵
- `/agent/query` 的分析期间固定为 `days=7`
- 代理 API 不通过 MCP 客户端调用 MCP 服务器
- SQLite 当前未连接到代理工具执行路径
- PyTorch 自动编码器不使用训练权重
- 无 FastAPI 和 MCP 协议集成测试
- 无身份验证、授权、超时、审计、监控
- 原因候选项是基于规则的检查对象，而不是实际原因诊断
### 下一步
1. 添加意图答案数据集和混淆矩阵
2. 比较基于规则的意图和 LLM 结构化意图
3. 将 `days` 和工具参数扩展为请求模式
4. 添加 FastAPI、MCP 协议和 Docker 构建集成测试
5. 将 SQLite 连接为工具数据源或执行历史记录
6. 实现模型训练、缩放和验证阈值
7. 添加工具超时、审计日志和权限控制
---
## 本项目展示
- 将自然语言制造问题分类为基于规则的意图的经验
- 使用 LangGraph 状态和节点分离处理阶段的经验
- 分离意图、路由器、工具和答案构建的责任的经验
- 基于 pandas 的制造数据过滤和聚合的经验
- 设计区分答案和证据的 API 响应模式的经验
- 在代理和 MCP 接口中重用相同 Python 工具的经验
- 区分代理 API 和模型端点实际范围的经验
- 使用 JSONL 跟踪、Docker 和 GitHub Actions 配置执行和验证路径的经验
---
## 联系方式
- 开发人员：김수진
- GitHub：https://github.com/lightleaping
- 邮箱：workingskyroad@gmail.com
