订单计划数据结合 AI 分析

一、项目目标

在现有系统中，通过 MCP (Model Context Protocol) 封装数据库查询与数据质量分析能力
借助第三方 AI 模型（兼容 OpenAI API）或者本地部署的模型实现 自然语言驱动的数据问题诊断与分析

二、总体架构

三、环境准备

四、核心代码实现

数据库操作与 MCP 工具定义

Vue 前端轮询调用示例

五、部署与运行

5.1 本地开发运行

六、测试与验证

总结

下面是代码部分：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


test_mcp/
  ├── data/                        # CSV 数据目录
  │    └── test.csv
  ├── skills/                      # skills 提示词目录
  │    └── data-analysis-skill.md
  ├── __init__.py
  ├── agent_service.py             # Agent 编排服务
  ├── mcp_http_client.py           # 通过 HTTP 直接调用 MCP Server
  ├── order_plan_cycle.py
  ├── router.py
  └── skill_loader.py              # 从 .md 文件加载 Skill

mcp_server.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335


# agent_service.py
"""
通过线程管理提交的 AI 分析任务
"""

from concurrent.futures import ThreadPoolExecutor
from enum import Enum
import os
import threading
import time
from typing import Optional
import uuid

from loguru import logger
from openai import OpenAI
from architecture.AiMcp.mcp_http_client import MCPHttpClient
from architecture.AiMcp.skill_loader import skill_loader


BASE_URL = os.getenv("OPENAI_BASE_URL", "your-base-url")
API_KEY = os.getenv("OPENAI_API_KEY", "your-key-here")
MODEL_NAME = os.getenv("AI_MODEL_NAME", "your-model-name")
MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "your-mcp-server-url")
SKILL_NAME = os.getenv("SKILL_NAME", "default-skill")
MAX_ITERATIONS = int(os.getenv("MAX_ANALYSIS_ITERATIONS", "10"))  # 最大调用工具轮次
MAX_TASK_NUMBER = int(os.getenv("MAX_ANALYSIS_TASK_NUMBER", "20"))  # 最大任务数量
MAX_WORKERS = int(os.getenv("MAX_ANALYSIS_WORKERS", "5"))  # 线程池的最大线程数量


class TaskStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    STOPPED = "stopped"


class Task:
    def __init__(self, task_id: str, question: str, skill_name: str = None):
        self.task_id = task_id
        self.question = question
        self.skill_name = skill_name or SKILL_NAME
        self.status = TaskStatus.PENDING
        self.result: Optional[str] = None
        self.error: Optional[str] = None
        self.created_at = time.time()
        self.started_at: Optional[float] = None
        self.completed_at: Optional[float] = None
        self.steps: list[dict] = []

        # 用于控制线程停止的 Event 对象
        self._stop_event = threading.Event()

    def stop(self):
        """
        设置停止信号
        """
        self._stop_event.set()

    def is_stopped(self) -> bool:
        """
        检查是否收到停止信号
        """
        return self._stop_event.is_set()

    def to_dict(self) -> dict:
        elapsed = None
        if self.started_at:
            end = self.completed_at or time.time()
            elapsed = round(end - self.started_at, 2)

        return {
            "task_id": self.task_id,
            "question": self.question,
            "skill": self.skill_name,
            "status": self.status.value,
            "result": self.result,
            "error": self.error,
            "elapsed_seconds": elapsed,
            "steps": self.steps,
        }


class TaskManager:
    def __init__(self, max_tasks: int = MAX_TASK_NUMBER):
        self._tasks: dict[str, Task] = {}
        self._lock = threading.RLock()
        self.max_tasks = max_tasks

    def create(self, question: str, skill_name: str = None) -> Task:
        with self._lock:
            current_count = len(self._tasks)

            # 如果任务数达到上限的 80%, 触发一次懒惰清理
            if current_count >= self.max_tasks * 0.8:
                self.cleanup(max_age_seconds=600)  # 清理10分钟前的任务
                current_count = len(self._tasks)

            if current_count >= self.max_tasks:
                raise RuntimeError("任务队列已满, 请稍后再试")

            task_id = str(uuid.uuid4())
            task = Task(task_id, question, skill_name)
            self._tasks[task_id] = task
            return task

    def get(self, task_id: str) -> Optional[Task]:
        with self._lock:
            return self._tasks.get(task_id)

    def stop_task(self, task_id: str) -> bool:
        """
        根据 taskId 停止正在运行的任务
        返回是否成功发送停止信号
        """
        with self._lock:
            task = self._tasks.get(task_id)
            if not task:
                return True
            if task and task.status == TaskStatus.RUNNING:
                task.stop()
                logger.info(f"已向任务 {task_id} 发送停止信号")
                return True
            return False

    def cleanup(self, max_age_seconds: int = 600):
        now = time.time()
        with self._lock:
            expired = [
                tid
                for tid, t in self._tasks.items()
                if now - t.created_at > max_age_seconds
                and t.status
                in [TaskStatus.COMPLETED, TaskStatus.STOPPED, TaskStatus.FAILED]
            ]
            for tid in expired:
                del self._tasks[tid]


class DataAnalysisAgent:
    def __init__(self):
        self.client = OpenAI(base_url=BASE_URL, api_key=API_KEY)
        self.mcp = MCPHttpClient(mcp_url=MCP_SERVER_URL)
        self._tools: Optional[list[dict]] = None

    def _ensure_tools(self) -> list[dict]:
        if self._tools is None:
            logger.info("正在从 MCP Server 获取工具列表...")
            self.mcp.initialize()
            self._tools = self.mcp.get_openai_tools()
            logger.info(f"已加载 {len(self._tools)} 个工具")
        return self._tools

    def _analyze(self, question: str, task: Task):
        """
        在后台线程中执行分析流程
        """
        task.status = TaskStatus.RUNNING
        task.started_at = time.time()
        logger.info(f"任务 {task.task_id} 开始: {question}")

        try:
            tools = self._ensure_tools()
            system_prompt = skill_loader.load(task.skill_name)

            messages = [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": question},
            ]

            for i in range(MAX_ITERATIONS):
                if task.is_stopped():
                    task.status = TaskStatus.STOPPED
                    task.result = "任务已被用户手动停止"
                    task.completed_at = time.time()
                    logger.info(f"任务 {task.task_id} 已停止")
                    return

                logger.info(f"任务 {task.task_id} 第 {i + 1} 轮推理")
                # 如果正在执行下面这行代码时（网络请求中）收到停止信号, 会等到本次请求返回后才会响应停止
                response = self.client.chat.completions.create(
                    model=MODEL_NAME,
                    messages=messages,
                    tools=tools,
                    tool_choice="auto",
                )
                # 再次检查, 防止在请求过程中收到了停止信号
                if task.is_stopped():
                    task.status = TaskStatus.STOPPED
                    task.result = "任务已被用户手动停止"
                    task.completed_at = time.time()
                    logger.info(f"任务 {task.task_id} 已停止 (请求返回后检测)")
                    return

                choice = response.choices[0]
                assistant_msg = choice.message

                # 如果没有 tool_calls, 说明模型认为任务完成
                if choice.finish_reason != "tool_calls" or not assistant_msg.tool_calls:
                    task.result = assistant_msg.content or "（模型未返回有效回答）"
                    task.status = TaskStatus.COMPLETED
                    task.completed_at = time.time()
                    logger.info(f"任务 {task.task_id} 完成")
                    return

                # 记录助手消息（含 tool_calls）
                messages.append(
                    {
                        "role": "assistant",
                        "content": assistant_msg.content,
                        "tool_calls": [
                            {
                                "id": tc.id,
                                "type": "function",
                                "function": {
                                    "name": tc.function.name,
                                    "arguments": tc.function.arguments,
                                },
                            }
                            for tc in assistant_msg.tool_calls
                        ],
                    }
                )

                # 执行工具调用
                for tc in assistant_msg.tool_calls:
                    # 在执行每个工具前也检查一下停止信号, 提高响应速度
                    if task.is_stopped():
                        task.status = TaskStatus.STOPPED
                        task.result = "任务已被用户手动停止"
                        task.completed_at = time.time()
                        logger.info(f"任务 {task.task_id} 已停止 (工具调用前)")
                        return

                    step = {
                        "round": i + 1,
                        "tool": tc.function.name,
                        "arguments": tc.function.arguments,
                    }
                    logger.info(
                        f">>>【任务 {task.task_id} 调用工具 {tc.function.name}】"
                    )

                    result_text = self.mcp.handle_tool_call(
                        tc.function.name, tc.function.arguments
                    )
                    step["result_preview"] = result_text
                    task.steps.append(step)

                    # 将工具结果回传给模型
                    messages.append(
                        {
                            "role": "tool",
                            "tool_call_id": tc.id,
                            "content": result_text,
                        }
                    )

            task.result = f"分析轮次超过上限 {MAX_ITERATIONS} 次, 请简化问题或分步提问"
            task.status = TaskStatus.COMPLETED
            task.completed_at = time.time()

        except Exception as e:
            # 如果异常是因为主动停止导致的, 可以忽略或者记录
            if task.is_stopped():
                logger.info(f"任务 {task.task_id} 因停止而中断")
            else:
                logger.error(f"任务 {task.task_id} 异常: {e}", exc_info=True)
                task.error = str(e)
                task.status = TaskStatus.FAILED
                task.completed_at = time.time()


# ============ 全局单例与线程池 ============
task_manager = TaskManager()
_executor = ThreadPoolExecutor(
    max_workers=MAX_WORKERS, thread_name_prefix="ai-mcp-task"
)
_agent_lock = threading.Lock()
_agent_instance: Optional[DataAnalysisAgent] = None


def get_agent() -> DataAnalysisAgent:
    global _agent_instance
    if _agent_instance is None:
        with _agent_lock:
            if _agent_instance is None:
                _agent_instance = DataAnalysisAgent()
    return _agent_instance


def submit_analysis(question: str, skill_name: str = None) -> str:
    """
    提交分析任务, 立即返回 task_id
    内部使用线程池控制并发
    """
    agent = get_agent()
    task = task_manager.create(question, skill_name)
    _executor.submit(agent._analyze, question, task)
    return task.task_id


def get_task_result(task_id: str) -> Optional[dict]:
    task = task_manager.get(task_id)
    if task is None:
        return None
    return task.to_dict()


def stop_analysis_task(task_id: str) -> bool:
    """
    停止任务
    根据 taskId 将对应的任务状态进行修改/标记
    """
    return task_manager.stop_task(task_id)


def get_tools():
    agent = get_agent()
    return agent._ensure_tools()


def get_skills():
    skill_name_list = skill_loader.list_skills()

    skill_data = []
    for skill_name in skill_name_list:
        content = skill_loader.load(skill_name)
        skill_data.append(
            {
                "skillName": skill_name,
                "content": content,
            }
        )
    return skill_data

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216


# mcp_http_client.py
import httpx
import json
from typing import Any

# 工具调用返回的最大字符数, 超出部分会被截断
MAX_TOOL_RESULT_LENGTH = 20000


class MCPHttpClient:
    """
    MCP HTTP 客户端
    通过 JSON-RPC 调用 MCP Server
    """

    def __init__(self, mcp_url: str = "http://localhost:8000/mcp"):
        self.mcp_url = mcp_url
        self._request_id = 0
        self._initialized = False
        self._client = httpx.Client(timeout=60.0)  # 连接复用

    def _next_id(self) -> int:
        self._request_id += 1
        return self._request_id

    def _post(self, payload: dict) -> dict:
        """
        同步发送 JSON-RPC 请求
        """
        resp = self._client.post(
            self.mcp_url,
            json=payload,
            headers={
                "Content-Type": "application/json",
                "Accept": "application/json, text/event-stream",
            },
        )
        resp.raise_for_status()

        # 处理 SSE 响应（MCP streamable-http 可能返回 SSE）
        content_type = resp.headers.get("content-type", "")
        if "text/event-stream" in content_type:
            return self._parse_sse_response(resp.text)
        else:
            return resp.json()

    async def _apost(self, payload: dict) -> dict:
        """
        异步发送 JSON-RPC 请求
        """
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.post(
                self.mcp_url,
                json=payload,
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json, text/event-stream",
                },
            )
            resp.raise_for_status()

            content_type = resp.headers.get("content-type", "")
            if "text/event-stream" in content_type:
                return self._parse_sse_response(resp.text)
            else:
                return resp.json()

    def _parse_sse_response(self, sse_text: str) -> dict:
        """
        解析 SSE 格式的响应, 收集所有 data 块
        """
        data_parts = []
        for line in sse_text.split("\n"):
            if line.startswith("data: "):
                data_str = line[6:].strip()
                if data_str:
                    data_parts.append(data_str)

        if data_parts:
            # 如果有多个 data, 取最后一个
            return json.loads(data_parts[-1])
        return {"error": "无法解析 SSE 响应"}

    # ==================== 协议操作 ====================
    def initialize(self) -> dict:
        """
        初始化 MCP 会话
        """
        payload = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {},
                "clientInfo": {"name": "mcp-http-client", "version": "1.0.0"},
            },
        }
        result = self._post(payload)
        self._initialized = True

        # 发送 initialized 通知（MCP 协议要求）
        notif = {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
        }
        try:
            self._post(notif)
        except Exception:
            pass
        return result  # 通知不需要响应, 可能返回 202 或无内容

    def list_tools(self) -> list[dict]:
        """
        获取 MCP 工具列表
        """
        if not self._initialized:
            self.initialize()

        payload = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": "tools/list",
            "params": {},
        }
        result = self._post(payload)
        return result.get("result", {}).get("tools", [])

    def call_tool(self, name: str, arguments: dict[str, Any] = None) -> str:
        """
        调用 MCP 工具, 返回结果文本
        """
        if not self._initialized:
            self.initialize()

        payload = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": "tools/call",
            "params": {
                "name": name,
                "arguments": arguments or {},
            },
        }
        result = self._post(payload)

        # 解析结果
        content_items = result.get("result", {}).get("content", [])
        text_parts = []
        for item in content_items:
            if item.get("type") == "text":
                text_parts.append(item["text"])

        if text_parts:
            result_text = "\n".join(text_parts)
        elif "result" in result:
            result_text = json.dumps(
                result["result"], separators=(",", ":"), ensure_ascii=False, default=str
            )
        else:
            result_text = json.dumps(
                result, separators=(",", ":"), ensure_ascii=False, default=str
            )

        # 统一截断, 防止 token 溢出
        if len(result_text) > MAX_TOOL_RESULT_LENGTH:
            result_text = result_text[:MAX_TOOL_RESULT_LENGTH] + "\n...（结果已截断）"
        return result_text

    # ==================== 格式转换 ====================
    def get_openai_tools(self) -> list[dict]:
        """
        获取 OpenAI function calling 格式的工具列表

        Returns:
            [{"type": "function", "function": {...}}, ...]
        """
        mcp_tools = self.list_tools()
        openai_tools = []

        for tool in mcp_tools:
            openai_tool = {
                "type": "function",
                "function": {
                    "name": tool["name"],
                    "description": tool.get("description", ""),
                    "parameters": tool.get(
                        "inputSchema",
                        {"type": "object", "properties": {}},
                    ),
                },
            }
            openai_tools.append(openai_tool)

        return openai_tools

    def handle_tool_call(self, tool_name: str, arguments_json: str) -> str:
        """
        处理模型返回的单个 tool_call

        Args:
            tool_name: 工具名称
            arguments_json: 模型返回的 arguments JSON 字符串

        Returns:
            工具执行结果文本
        """
        try:
            arguments = json.loads(arguments_json)
        except json.JSONDecodeError:
            arguments = {}

        try:
            return self.call_tool(tool_name, arguments)
        except Exception as e:
            return f"[工具调用失败] {tool_name}: {str(e)}"

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250


# order_plan_cycle.py
"""
订单计划和周期
"""
from typing import Dict, List, Optional

from sqlalchemy import text
from sqlalchemy.orm import Session

from exts import DatabaseManager


def fix_encoding(data):
    """
    处理编码问题
    """
    if isinstance(data, str):
        # str -> bytes (ISO-8859-1) -> str (GBK)
        try:
            fix_data = data.encode("ISO-8859-1").decode("GBK", errors="ignore")
            return fix_data
        except Exception:
            return data
    elif isinstance(data, dict):
        return {key: fix_encoding(value) for key, value in data.items()}
    elif isinstance(data, list):
        return [fix_encoding(item) for item in data]
    else:
        # 其他类型（int, float, None, ...）保持不变
        return data


class OrderPlanAndCycleGateway(object):
    def get_table_msg(
        self, po_session: Session, table_name_list: List[str]
    ) -> List[Optional[Dict]]:
        """
        获取指定表的字段等信息
        """
        select_stmt = text("""
        SELECT
          表名 = CASE WHEN scol.colorder = 1 THEN so.name ELSE so.name END,
          表说明 = CASE WHEN scol.colorder = 1 THEN isnull(ep2.value, '') ELSE isnull(ep2.value, '') END,
          字段序号 = scol.colorder,
          字段名 = scol.name,
          标识 = CASE WHEN COLUMNPROPERTY( scol.id, scol.name, 'IsIdentity')= 1 THEN 'yes' ELSE 'no' END,
          主键 = CASE
            WHEN EXISTS(
              SELECT 1 FROM sysobjects WHERE xtype = 'PK' AND parent_obj = scol.id
              AND name IN (
                SELECT name FROM sysindexes WHERE indid IN(
                  SELECT indid FROM sysindexkeys WHERE id = scol.id AND colid = scol.colid
                )
              )
            ) THEN 'yes'
            ELSE 'no'
          END,
          类型 = st.name,
          占用字节数 = scol.length,
          长度 = COLUMNPROPERTY(scol.id, scol.name, 'PRECISION'),
          小数位数 = isnull(COLUMNPROPERTY(scol.id, scol.name, 'Scale'), 0),
          允许空 = CASE WHEN scol.isnullable = 1 THEN 'yes' ELSE 'no' END,
          默认值 = isnull(scom.text, ''),
          字段说明 = isnull(ep1.[value], '')
        FROM
          syscolumns AS scol
        LEFT JOIN systypes AS st
          ON scol.xusertype = st.xusertype
        INNER JOIN sysobjects AS so
          ON scol.id = so.id
          AND so.xtype = 'U'
          AND so.name <> 'dtproperties'
        LEFT JOIN syscomments AS scom
          ON scol.cdefault = scom.id
        LEFT JOIN sys.extended_properties AS ep1
          ON scol.id = ep1.major_id
          AND scol.colid = ep1.minor_id
        LEFT JOIN sys.extended_properties ep2
          ON so.id = ep2.major_id
          AND ep2.minor_id = 0
        WHERE
          so.name IN :table_name_list
        ORDER BY
          scol.id,
          scol.colorder
        """)
        query_res = (
            po_session.execute(select_stmt, params={"table_name_list": table_name_list})
            .mappings()
            .all()
        )
        query_data = [dict(row) for row in query_res]

        return query_data

    def query_order_data(
        self,
        po_session: Session,
        po_no_list: List[str],
        limit: Optional[int] = None,
        offset: Optional[int] = None,
    ) -> List[Optional[Dict]]:
        select_stmt = text("""
        SELECT
          *
        FROM (

          SELECT
            ROW_NUMBER() OVER(ORDER BY dpr.sort_no, dpr.operation_sort_no) AS row_number,
            dpr.po_no,
            dpr.style_id,
            dpr.operation_id,
            dpr.section_id,
            dpr.material_type,
            dpr.qty,
            dpr.deleted,
            dpr.create_at,
            dpr.plan_start_time,
            dpr.plan_end_time,
            dpr.real_start_time,
            dpr.real_end_time,
            dpr.ordered_time,
            dpr.picked_time,
            dpr.shipped_time,
            dpr.remark,
            dpr.sort_no,
            dpr.operation_sort_no,
            dpr.material_qt_time,
            dpr.required_delivery_date,
            dpr.production_time,
            dpr.real_production_time,

            dws.section_name,
            do.operation_name,
            CASE material_stats
              WHEN 10 THEN '未齐套'
              WHEN 20 THEN '齐套'
              WHEN 30 THEN '部分齐套'
              WHEN 40 THEN '容差齐套'
              WHEN 50 THEN '手动齐套'
            END AS material_stats_desc,
            CASE ordered_status
              WHEN 10 THEN '新建'
              WHEN 20 THEN '进行中'
              WHEN 30 THEN '完成'
            END AS ordered_status_desc,
            CASE picked_status
              WHEN 10 THEN '新建'
              WHEN 20 THEN '进行中'
              WHEN 30 THEN '完成'
            END AS picked_status_desc,
            CASE shipped_status
              WHEN 10 THEN '新建'
              WHEN 20 THEN '进行中'
              WHEN 30 THEN '完成'
            END AS shipped_status_desc,
            DATEDIFF(SECOND, plan_start_time, plan_end_time) AS plan_time_diff_second,
            DATEDIFF(SECOND, real_start_time, real_end_time) AS real_time_diff_second,
            ISNULL(DATEDIFF(SECOND, real_start_time, real_end_time),0) - ISNULL(DATEDIFF(SECOND, plan_start_time, plan_end_time),0) AS real_plan_diff
          FROM
            [bocini].[dmp_production_route] AS dpr
          LEFT JOIN [bocini].[dmp_work_section] AS dws
            ON dws.id = dpr.section_id
            AND dws.deleted = 0
          LEFT JOIN [bocini].[dmp_operation] AS do
            ON do.id = dpr.operation_id
            AND do.deleted = 0
          WHERE 1 = 1
            AND dpr.deleted = 0
            -- AND dpr.section_id = 6
            AND dpr.po_no IN :po_no_list

        ) AS a
        WHERE 1 = 1
          AND a.row_number > :offset and a.row_number <= :right_index
        """)

        limit = limit if limit is not None and limit >= 0 else 999
        offset = offset if offset is not None and offset >= 0 else 0
        right_index = offset + limit

        _params = {
            "po_no_list": po_no_list,
            "offset": offset,
            "right_index": right_index,
        }
        query_res = po_session.execute(select_stmt, params=_params).mappings().all()
        query_data = [dict(row) for row in query_res]

        query_data_fixed = [fix_encoding(row) for row in query_data]

        return query_data_fixed


class OrderPlanAndCycleService(object):
    def __init__(self):
        self.gateway = OrderPlanAndCycleGateway()

    def get_order_plan_cycle_table_msg(self) -> Dict:
        """
        获取订单计划和周期的相关表格信息
        """
        try:
            table_name_list = [
                "dmp_production_route",
                "dmp_work_section",
                "dmp_operation",
            ]

            with DatabaseManager.session("po") as po_session:
                with po_session.begin():
                    data = self.gateway.get_table_msg(po_session, table_name_list)

                    if not data:
                        return {"error": "未找到对应的表信息", "success": False}

            return {"result": data, "success": True}
        except Exception as e:
            return {"error": str(e), "success": False}

    def query_order_plan_cycle_data(
        self,
        po_no_list: List[str],
        limit: Optional[int] = None,
        offset: Optional[int] = None,
    ) -> Dict:
        """
        查询指定订单的计划、周期数据

        Args:
            po_no_list: PO订单号列表
            limit: 分页-每页数据条数, 默认 999
            offset: 分页-数据偏移, 默认 0
        """
        try:
            with DatabaseManager.session("po") as po_session:
                with po_session.begin():
                    data = self.gateway.query_order_data(
                        po_session, po_no_list, limit, offset
                    )

                    if not data:
                        return {
                            "error": "查询指定订单的计划、周期数据结果为空",
                            "success": False,
                        }

            return {"result": data, "success": True}
        except Exception as e:
            return {"error": str(e), "success": False}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178


# router.py
from flask import Blueprint, jsonify, request
from loguru import logger

from .agent_service import (
    get_skills,
    get_task_result,
    get_tools,
    stop_analysis_task,
    submit_analysis,
)
from .test_csv_data_analysis import CsvDataAnalysis

aimcp_api = Blueprint("ai_mcp", __name__, url_prefix="/aiMcp")


@aimcp_api.route("/getTools", methods=["GET"])
def get_ai_mcp_tools():
    """
    获取当前 AI 可用的 MCP 工具列表
    """
    try:
        tools = get_tools()
        return jsonify(
            {
                "code": 200,
                "status": True,
                "message": "获取当前 AI 可用的 MCP 工具列表成功",
                "data": {"tools": tools, "count": len(tools)},
            }
        )
    except Exception as e:
        return jsonify(
            {
                "code": 500,
                "status": False,
                "message": f"获取当前 AI 可用的 MCP 工具列表失败, {str(e)}",
            }
        )


@aimcp_api.route("/getSkills", methods=["GET"])
def get_ai_mcp_skills():
    """
    获取当前 AI 可用的 Skill 列表
    """
    try:
        skills = get_skills()
        return jsonify(
            {
                "code": 200,
                "status": True,
                "message": "获取当前 AI 可用的 Skill 列表成功",
                "data": {"skills": skills, "count": len(skills)},
            }
        )
    except Exception as e:
        return jsonify(
            {
                "code": 500,
                "status": False,
                "message": f"获取当前 AI 可用的 Skill 列表失败, {str(e)}",
            }
        )


@aimcp_api.route("/health", methods=["GET"])
def health():
    """
    健康检查
    """
    return jsonify({"status": "healthy"})


@aimcp_api.route("/submitTask", methods=["POST"])
def submit_ai_mcp_task():
    """
    提交 AI 分析任务
    """
    try:
        req_data = request.get_json()
        question = req_data.get("question", "").strip()
        skill_name = req_data.get("skill")

        if not question:
            return jsonify(
                {"code": 500, "status": False, "message": "question 不能为空"}
            )

        task_id = submit_analysis(question, skill_name=skill_name)
        return jsonify(
            {
                "code": 200,
                "status": True,
                "message": "提交成功",
                "data": {
                    "task_id": task_id,
                    "status": "pending",
                    "message": f"请轮询 {task_id} 获取结果",
                },
            }
        )
    except Exception as e:
        logger.error(f"提交 AI 分析任务失败, {str(e)}", exc_info=True)
        return jsonify(
            {"code": 500, "status": False, "message": f"提交 AI 分析任务失败, {str(e)}"}
        )


@aimcp_api.route("/task/<task_id>", methods=["GET"])
def get_ai_mcp_task_result(task_id: str):
    """
    查询任务状态和结果
    """
    try:
        result = get_task_result(task_id)
        if result is None:
            return jsonify({"code": 500, "status": False, "message": "任务不存在"}), 404

        return jsonify(
            {"code": 200, "status": True, "message": "查询成功", "data": result}
        )
    except Exception as e:
        logger.error(f"查询 {task_id} 任务状态和结果失败, {str(e)}", exc_info=True)
        return jsonify(
            {
                "code": 500,
                "status": False,
                "message": f"查询 {task_id} 任务状态和结果失败, {str(e)}",
            }
        )


@aimcp_api.route("/stopTask", methods=["POST"])
def stop_ai_mcp_task():
    """
    停止任务
    """
    try:
        req_data = request.get_json()
        task_id = req_data.get("taskId")

        if not task_id:
            return jsonify(
                {"code": 500, "status": False, "message": "task_id 不能为空"}
            )

        stop_result = stop_analysis_task(task_id)
        if stop_result:
            return jsonify(
                {
                    "code": 200,
                    "status": True,
                    "message": f"{task_id} 停止成功",
                }
            )
        else:
            return jsonify(
                {"code": 500, "status": False, "message": f"{task_id} 停止失败"}
            )
    except Exception as e:
        logger.error(f"停止任务失败, {str(e)}", exc_info=True)
        return jsonify({"code": 500, "status": False, "message": f"停止失败, {str(e)}"})


@aimcp_api.route("/test", methods=["GET"])
def ai_task_test():
    """
    测试 列出文件
    """
    try:
        obj = CsvDataAnalysis()
        result = obj.list_files()

        return jsonify({"code": 200, "data": result})
    except Exception as e:
        logger.error(str(e), exc_info=True)
        return jsonify({"code": 500, "status": False, "message": str(e)})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55


# skill_loader.py
from pathlib import Path
from typing import List


class SkillLoader(object):
    """
    从 .md 文件加载 Skill, 用作 system prompt
    找不到文件时返回默认兜底提示词
    """

    def __init__(self, skill_dir: str = "./architecture/AiMcp/skills"):
        self.skill_dir = Path(skill_dir)
        self.skill_dir.mkdir(exist_ok=True)
        self._cache: dict[
            str, tuple[float, str]
        ] = {}  # 缓存 name → (文件修改时间, 内容)

    def load(self, skill_name: str = "local-csv-analysis-skill") -> str:
        """
        加载 Skill 文件

        文件修改后重新加载（通过 mtime 检测）
        """
        filename = skill_name if skill_name.endswith(".md") else f"{skill_name}.md"
        filepath = self.skill_dir / filename

        if not filepath.exists():
            return self._default_prompt()

        # 检查缓存是否过期
        mtime = filepath.stat().st_mtime
        if skill_name in self._cache:
            cached_mtime, cached_content = self._cache[skill_name]
            if mtime <= cached_mtime:
                return cached_content

        # 读取内容
        content = filepath.read_text(encoding="utf-8")
        self._cache[skill_name] = (mtime, content)
        return content

    def list_skills(self) -> List[str]:
        """
        列出所有可用的 Skill 文件
        """
        return [f.stem for f in self.skill_dir.glob("*.md")]

    def _default_prompt(self) -> str:
        return """你是一个友好的助手, 通过工具访问数据并进行分析, 帮助用户处理各种事务
        一般先根据需要, 看是否有合适的的工具可以调用，再进行预览和分析数据"""


# 全局单例
skill_loader = SkillLoader()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100


# mcp_server.py
"""
MCP 服务器, 将数据操作暴露为 AI 可调用的工具

启动服务 dotenv -f .env run ./.venv/Scripts/python -m mcp_server
或者使用交互式页面 dotenv -f .env run mcp dev mcp_server.py
"""

from typing import Dict, List

from mcp.server.fastmcp import FastMCP

from architecture.AiMcp.test_csv_data_analysis import CsvDataAnalysis
from architecture.AiMcp.order_plan_cycle import OrderPlanAndCycleService

# 创建 MCP 服务器, 启用无状态 HTTP 模式用于生产部署
mcp = FastMCP("ai-mcp", stateless_http=True)
csv_data_service = CsvDataAnalysis()
order_plan_cycle_service = OrderPlanAndCycleService()


# ================================================== 测试用 - 读取分析 CSV 文件 ==================================================
@mcp.tool()
async def list_data_files() -> list:
    """
    列出数据目录下的所有文件
    """
    return csv_data_service.list_files()


@mcp.tool()
async def preview_csv(filename: str) -> Dict:
    """
    加载 CSV 文件并返回预览信息（前5行、列名、行数）

    Args:
        filename: CSV 文件名
    """
    result = csv_data_service.load_csv(filename)
    if result is None:
        return {"error": f"文件 {filename} 不存在"}
    return result


@mcp.tool()
async def get_summary_stats(filename: str, column: str = None) -> Dict:
    """
    获取数据的描述性统计

    Args:
        filename: CSV 文件名
        column: 可选, 指定列名；不指定则返回所有数值列的统计
    """
    result = csv_data_service.load_csv(filename)
    if result is None:
        return {"error": f"文件 {filename} 不存在", "success": False}
    return csv_data_service.get_summary_stats(filename, column)


# @mcp.tool()
# async def execute_query(filename: str, query: str) -> Dict:
#     """
#     在 CSV 文件上执行 SQL 查询
#     Args:
#         filename: CSV 文件名
#         query: SQL 查询语句, 例如:
#             A = sales * quantity
#     """
#     return csv_data_service.execute_query(filename, query)


# ================================================== 订单计划和周期 ==================================================
@mcp.tool()
async def get_order_plan_cycle_table_msg() -> Dict:
    """
    获取订单计划和周期的相关表格信息
    """
    return order_plan_cycle_service.get_order_plan_cycle_table_msg()


@mcp.tool()
async def query_order_plan_cycle_data(
    po_no_list: List[str], limit: int, offset: int
) -> Dict:
    """
    查询指定订单的计划、周期数据

    Args:
        po_no_list: PO订单号列表
        limit: 分页-每页数据条数, 默认 999
        offset: 分页-数据偏移, 默认 0
    """
    return order_plan_cycle_service.query_order_plan_cycle_data(
        po_no_list, limit, offset
    )


if __name__ == "__main__":
    # mcp.run(transport="stdio")  # stdio 模式用于本地测试
    mcp.run(transport="streamable-http")

前端部分：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537


<!-- 订单计划、周期分析页面 -->
<template>
  <div
    class="plan-cycle-analysis"
    :style="{ height: `${analysisContainerHeight}px` }"
  >
    <div
      class="ai-mcp-container"
      v-loading="operationLoading"
      element-loading-text="请稍后..."
    >
      <!-- 提问 -->
      <div class="sticky-question">
        <el-input
          v-model="question"
          type="textarea"
          :autosize="{ minRows: 5, maxRows: 10 }"
          placeholder="请输入分析问题，如: 帮我看看 CSI25153160CA 这个订单的情况"
          :disabled="isRunning"
        />

        <div class="input-footer">
          <div class="input-footer-left">
            <el-button
              icon="el-icon-connection"
              circle
              @click="showSkillContent"
              :loading="isRunning"
            />
            <span style="margin-left: 5px; color: #606266; cursor: text">
              已选择 {{ skill }}
            </span>
          </div>

          <div class="input-footer-right">
            <el-button
              v-if="isRunning"
              type="danger"
              round
              @click="stopIntervalAndReset"
            >
              停止
            </el-button>

            <el-button
              type="primary"
              plain
              round
              icon="el-icon-data-analysis"
              @click="startAnalysis"
              :loading="isRunning"
              :disabled="!question.trim()"
            >
              {{ isRunning ? '运行中...' : '开始分析' }}
            </el-button>
          </div>
        </div>
      </div>

      <!-- 分析过程和结果 -->
      <div v-if="!error && taskId" class="result-wrapper">
        <!-- 任务概览信息 -->
        <el-card shadow="never" class="info-card">
          <div class="task-info">
            <div class="info-item">
              <span class="label">任务ID:</span>
              <span class="value mono">{{ taskId }}</span>
            </div>
            <div class="info-item">
              <span class="label">调用技能:</span>
              <span class="value mono">{{ skill }}</span>
            </div>
            <div class="info-item">
              <span class="label">耗时:</span>
              <span class="value mono" v-if="elapsedSeconds">
                {{ elapsedSeconds.toFixed(1) }}s
              </span>
            </div>
            <div class="info-item">
              <span class="label">状态:</span>
              <el-tag size="small" :type="statusTagType" effect="dark">
                {{ statusText }}
              </el-tag>
            </div>
          </div>
        </el-card>

        <!-- 分析步骤时间线 -->
        <el-card shadow="never" class="timeline-card">
          <div slot="header" class="card-header timeline-header">
            <span>
              <i class="el-icon-s-marketing"></i>
              分析步骤
            </span>
          </div>

          <el-timeline>
            <el-timeline-item
              v-for="step in steps"
              :key="step.round"
              :timestamp="`第 ${step.round} 轮`"
              placement="top"
              type="primary"
            >
              <el-card shadow="hover" class="step-card">
                <div class="step-header">
                  <span class="tool-label">调用工具:</span>
                  <el-tag size="small" effect="plain">{{ step.tool }}</el-tag>
                </div>

                <!-- 参数折叠面板 -->
                <el-collapse
                  v-if="hasContent(step.arguments)"
                  class="step-collapse"
                >
                  <el-collapse-item title="查看请求参数">
                    <pre class="json-block">{{
                      formatJson(step.arguments)
                    }}</pre>
                  </el-collapse-item>
                </el-collapse>

                <!-- 结果预览折叠面板 -->
                <el-collapse
                  v-if="hasContent(step.result_preview)"
                  class="step-collapse"
                >
                  <el-collapse-item title="查看结果预览">
                    <pre class="json-block">{{
                      formatJson(step.result_preview)
                    }}</pre>
                  </el-collapse-item>
                </el-collapse>
              </el-card>
            </el-timeline-item>
          </el-timeline>

          <!-- 运行中等待动画 -->
          <div v-if="isRunning" class="thinking-placeholder">
            <i class="el-icon-loading"></i>
            AI 正在思考...
          </div>
        </el-card>

        <el-card
          shadow="never"
          class="result-card"
          v-if="status === 'completed' && result"
        >
          <div slot="header" class="card-header result-header">
            <span>
              <i class="el-icon-s-claim"></i>
              结果
            </span>
          </div>
          <div class="result-content" v-html="resultHtmlContent"></div>
        </el-card>
      </div>

      <!-- 错误信息 -->
      <div v-if="error" class="result-wrapper">
        <span>{{ error }}</span>
      </div>
    </div>

    <el-dialog
      :title="skillsDialog.title"
      :visible.sync="skillsDialog.visible"
      width="70%"
      top="100px"
      :close-on-click-modal="false"
      append-to-body
    >
      <div v-loading="operationLoading" element-loading-text="请稍后...">
        <el-select
          v-model="skill"
          placeholder="请选择"
          filterable
          style="width: 300px"
        >
          <el-option
            v-for="(item, index) in skillData"
            :key="index"
            :label="item.skillName"
            :value="item.skillName"
          />
        </el-select>

        <div class="skill-content-preview">
          <pre>{{ selectedSkillContent }}</pre>
        </div>
      </div>
    </el-dialog>

    <!-- markdown 编辑器, 可用于将 markdown 转 html -->
    <VabMarkdownEditor
      ref="mde"
      v-show="false"
      v-model="result"
      @show-html="handleShowResultHtml"
    />
  </div>
</template>

<script>
import VabMarkdownEditor from '@/plugins/vabMarkdownEditor'
import {
  getAiMcpSkills,
  getAiMcpTools,
  submitAiMcpTask,
  getAiMcpTaskResult,
  stopAiMcpTask,
} from '@/api/pyapi/commonBusiness'

export default {
  name: 'PlanCycleAnalysis',

  components: {
    VabMarkdownEditor,
  },

  data() {
    return {
      analysisContainerHeight: 0,

      operationLoading: false,
      skillsDialog: {
        title: null,
        visible: false,
      },
      skillData: [],
      question: '',
      skill: 'order-plan-cycle-data-analysis-skill',

      isRunning: false,
      taskId: '',
      status: '',
      result: '',
      error: null,
      elapsedSeconds: 0,
      steps: [],
      resultHtmlContent: '',

      timer: null,

      // Mock 数据相关 (仅用于演示)
      mockRound: 0,
      mockTaskData: {
        task_id: '8bc9bc4d-c8a9-4023-bf72-08c9c8a2a7e1',
        question: '帮我看看 CSI25153160CA 这个订单的情况',
        skill: 'order-plan-cycle-data-analysis-skill',
        steps: [
          {
            round: 1,
            tool: 'get_order_plan_cycle_table_msg',
            arguments: '{}',
            result_preview:
              '{\n  "result": [\n    {\n      "表名": "dmp_production_route",\n      "表说明": "",\n      "字段序号": 1,\n      "字段名": "id",\n      "标识": "yes",\n      "主键": "yes",\n      "类型": "int",\n      "占用字节数": 4,\n      "长度": 10',
          },
          {
            round: 2,
            tool: 'query_order_plan_cycle_data',
            arguments:
              '{"po_no_list": ["CSI25153160CA"], "limit": 999, "offset": 0}',
            result_preview:
              '{\n  "result": [\n    {\n      "row_number": "1",\n      "id": 101950,\n      "poid": 201744,\n      "po_no": "CSI25153160CA",\n      "style_id": "CSI-SS26MT2"',
          },
          {
            round: 3,
            tool: 'query_order_plan_cycle_data',
            arguments:
              '{"po_no_list": ["CSI25153160CA"], "limit": 999, "offset": 0}',
            result_preview:
              '{\n  "result": [\n    {\n      "row_number": "1",\n      "id": 101950,\n      "poid": 201744,\n      "po_no": "CSI25153160CA",\n      "style_id": "CSI-SS26MT2"',
          },
        ],
        final_result:
          '已分析订单 CSI25153160CA 的计划与周期数据: \n\n**整体结论**: 该订单共发现2个异常环节【缝制、质检】。实际总耗时272.5天（计划5天），超时率高达5350%。主要问题集中在生产环节的时间失控和质检流程倒置。\n\n**各环节详情**: \n1. 缝制环节: \n   - 计划耗时: 2.73天（2025-09-03至2025-09-06）\n   - 实际耗时: 272.5天（2025-08-22至2026-05-22）\n   - 偏差率: +20000%（系统显示"未齐套"状态）\n   - 异常原因: 物料长期未齐套导致生产停滞\n\n2. 质检环节: \n   - 计划耗时: 2天（2025-09-06至2025-09-08）\n   - 实际耗时: 272.5天（2025-08-30至2026-05-22）\n   - 异常原因: \n     - 流程异常: 质检提前16天在缝制完成前启动\n     - 检测周期失控（正常应0.5天完成）\n\n**改进建议**: \n1. 建立物料齐套预警机制，当material_qt_time（物料齐套时间）晚于计划开工时间时自动触发采购加急流程\n2. 优化质检流程，设置开工前检查点，防止未完成生产的产品进入质检环节\n3. 重新评估生产排期合理性，当前计划时间（5天）与实际耗时（272.5天）严重不符，建议采用历史数据均值法重新制定标准工时',
      },
    }
  },

  computed: {
    // 状态标签样式
    statusTagType() {
      if (this.status === 'completed') return 'success'
      if (this.status === 'running') return 'warning'
      if (this.status === 'pending') return 'primary'
      if (this.error) return 'danger'
      return 'info'
    },
    // 状态文本
    statusText() {
      if (this.status === 'completed') return '分析完成'
      if (this.status === 'running') return '运行中'
      if (this.status === 'pending') return '等待中'
      if (this.error) return '发生错误'
      return this.status
    },

    // 获取所选 skill 的内容
    selectedSkillContent() {
      const idx = this.skillData.findIndex(
        (item) => item.skillName === this.skill
      )
      return idx !== -1 ? this.skillData[idx].content : ''
    },
  },

  created() {},

  mounted() {
    this.initContainerHeight()
    this.addResizeListener()
  },

  // 组件销毁时
  beforeDestroy() {
    // 清除定时器
    if (this.timer) {
      clearInterval(this.timer)
    }

    this.removeResizeListener()
  },

  methods: {
    async getSkills() {
      try {
        this.operationLoading = true
        const resp = await getAiMcpSkills()

        if (resp.code === 200) {
          this.skillData = resp.data.skills
        } else {
          this.$message.error(resp.message)
        }
      } catch (err) {
        this.$message.error(`查看当前 AI 可用的 Skill 列表失败, ${err.message}`)
      } finally {
        this.operationLoading = false
      }
    },

    async stopIntervalAndReset() {
      try {
        this.operationLoading = true
        const resp = await stopAiMcpTask({ taskId: this.taskId })

        if (resp.code === 200) {
          if (this.timer) {
            clearInterval(this.timer)
          }
          this.isRunning = this.$options.data().isRunning
          this.status = this.$options.data().status
          this.taskId = this.$options.data().taskId
          this.result = this.$options.data().result
          this.error = this.$options.data().error
          this.elapsedSeconds = this.$options.data().elapsedSeconds
          this.steps = this.$options.data().steps
          this.resultHtmlContent = this.$options.data().resultHtmlContent
        } else {
          this.$message.error(resp.message)
        }
      } catch (err) {
        this.$message.error(`停止失败, ${err.message}`)
      } finally {
        this.operationLoading = false
      }
    },

    // 开始分析
    async startAnalysis() {
      try {
        if (!this.question.trim() || this.isRunning) return

        // 重置数据
        this.isRunning = true
        this.status = 'running'
        this.taskId = this.$options.data().taskId
        this.result = this.$options.data().result
        this.error = this.$options.data().error
        this.elapsedSeconds = this.$options.data().elapsedSeconds
        this.steps = this.$options.data().steps
        this.resultHtmlContent = this.$options.data().resultHtmlContent

        // ---------- 接口调用 ----------
        // const resp = await submitAiMcpTask({
        //   question: this.question,
        //   skill: this.skill,
        // })
        // this.taskId = resp.data.task_id
        // this.pollTask()  // 开始轮询

        // ---------- Mock 模拟 ----------
        this.mockRound = 0
        this.taskId = this.mockTaskData.task_id
        this.skill = this.mockTaskData.skill
        this.mockPollTask()

        this.$message.success('AI 正在分析中，请稍候...')
      } catch (err) {
        this.$message.error('启动分析失败')
        this.isRunning = false
      }
    },

    // 真实轮询逻辑
    async pollTask() {
      this.timer = setInterval(async () => {
        try {
          const resp = await getAiMcpTaskResult({ taskId: this.taskId })
          const data = resp.data

          this.status = data.status
          this.elapsedSeconds = data.elapsed_seconds || 0
          this.error = data.error
          this.steps = data.steps || []

          if (this.status === 'completed') {
            this.result = data.result
            this.$refs.mde.add(this.result)

            this.isRunning = false
            clearInterval(this.timer)
            this.timer = null
            this.$message.success(resp.message)
          } else if (this.error) {
            this.isRunning = false
            clearInterval(this.timer)
            this.timer = null
            this.$message.error(`分析失败: ${this.error}`)
          }
        } catch (err) {
          console.error('轮询失败', err)
          this.isRunning = false
          clearInterval(this.timer)
          this.timer = null
          this.$message.error(`获取分析状态失败, ${err.message}`)
        }
      }, 5000) // 每5秒轮询一次
    },

    // 判断是否有内容
    hasContent(str) {
      return str && str.trim() !== '' && str.trim() !== '{}'
    },

    // 格式化JSON
    formatJson(str) {
      if (!str) return ''
      try {
        const obj = JSON.parse(str)
        return JSON.stringify(obj, null, 2)
      } catch (e) {
        return str
      }
    },

    handleShowResultHtml(html) {
      this.resultHtmlContent = html
    },

    async showSkillContent() {
      this.skillsDialog.title = '请选择技能'
      this.skillsDialog.visible = true
      await this.getSkills()
    },

    // -------------------- 页面高度 --------------------
    // 初始化容器高度
    initContainerHeight() {
      this.$nextTick(() => {
        this.calculateContainerHeight()
      })
    },

    // 根据页面高度设置容器高度
    calculateContainerHeight() {
      // console.log(window.innerHeight)
      this.reportContainerHeight = window.innerHeight - 86
      // console.log(this.reportContainerHeight)
    },

    // 添加监听以自适应高度
    addResizeListener() {
      this.handleResize = this.debounce(this.calculateContainerHeight, 150)
      window.addEventListener('resize', this.handleResize)
    },

    // 移除监听
    removeResizeListener() {
      if (this.handleResize) {
        window.removeEventListener('resize', this.handleResize)
      }
    },

    // 防抖函数
    debounce(fn, delay) {
      let timer = null
      return function (...args) {
        clearTimeout(timer)
        timer = setTimeout(() => fn.apply(this, args), delay)
      }
    },

    // -------------------- Mock 模拟方法 --------------------
    mockPollTask() {
      const mockData = this.mockTaskData
      this.timer = setInterval(() => {
        this.mockRound++
        this.elapsedSeconds += 5

        if (this.mockRound <= mockData.steps.length) {
          // 每次轮询塞入一个step
          this.steps.push(mockData.steps[this.mockRound - 1])
        }

        if (this.mockRound === mockData.steps.length + 1) {
          // 轮询结束
          this.status = 'completed'
          this.result = mockData.final_result
          this.$refs.mde.add(this.result)
          this.elapsedSeconds = 74.42
          this.isRunning = false
          clearInterval(this.timer)
          this.$message.success('分析完成')
        }
      }, 5000)
    },
  },
}
</script>

<style scoped>
省略。。。
</style>