当前位置：首页 > news >正文

《AI大模型趣味实战》智能Agent和MCP协议的应用实例：搭建一个能阅读DOC文件并实时显示润色改写过程的Python Flask应用

news 来源：原创 2025/4/28 14:59:20

智能Agent和MCP协议的应用实例：搭建一个能阅读DOC文件并实时显示润色改写过程的Python Flask应用

引言

随着人工智能技术的飞速发展，智能Agent与模型上下文协议(MCP)的应用场景越来越广泛。本报告将详细介绍如何基于Python Flask框架构建一个智能应用，该应用能够读取DOC文件并在前端实时显示润色改写过程。本文将深入探讨如何整合Flask、win32com、MCP协议以及智能Agent技术，为读者提供一个完整的实现方案。

系统架构设计

在这里插入图片描述

整体架构

我们的系统由三个主要部分组成：

前端界面：提供文件上传功能，并实时显示润色改写过程
Flask后端：处理文件上传，使用win32com读取DOC文件内容，通过MCP协议与智能Agent通信
智能Agent：基于MCP协议实现，负责文本的润色改写工作

组件分工

前端：负责与用户的交互，展示上传进度和润色结果
Flask后端：作为中间层，处理文件上传，协调各组件工作
智能Agent：实现文本处理的核心逻辑，通过MCP协议与后端通信

通信机制

前端与后端：使用WebSocket技术实现实时通信
后端与智能Agent：使用MCP协议进行通信

项目文档结构

以下是项目的标准文档结构：

project_name/
├── README.md
├── requirements.txt
├── app/
│   ├── __init__.py
│   ├── routes.py
│   ├── doc_processor.py
│   ├── mcp_agent.py
│   ├── socketio_handler.py
│   ├── templates/
│   │   ├── index.html
│   │   └── result.html
│   └── static/
│       ├── style.css
│       └── script.js
└── tests/├── test_doc_processor.py└── test_mcp_agent.py

文件功能说明

项目根目录

README.md：项目概述、功能介绍、安装指南、使用说明和联系方式
requirements.txt：列出项目所需的所有Python依赖包及其版本号，便于他人快速安装环境

Flask应用主目录 (`app/`)

init.py：初始化Flask应用，配置相关设置（如调试模式、密钥等），并注册蓝图
routes.py：定义应用的路由规则和处理函数，包括：
- 主页面路由（显示上传表单）
- 文件上传处理路由（接收上传的DOC文件）
- 实时显示处理结果的WebSocket路由
doc_processor.py：使用win32com库读取DOC文件内容，提供文本提取和保存的功能
mcp_agent.py：实现MCP协议的智能Agent，负责与大模型通信，执行文本润色任务
socketio_handler.py：处理WebSocket通信，实现实时显示润色过程
templates/
- index.html：主页面，包含文件上传表单和上传按钮
- result.html：显示润色结果的页面，包含实时更新的文本区域
static/
- style.css：定义页面的样式，包括布局、颜色、字体等
- script.js：前端JavaScript代码，处理文件上传、与后端的通信以及实时显示润色过程

测试文件目录 (`tests/`)

test_doc_processor.py：测试doc_processor.py模块的功能，包括文件读取、文本提取等
test_mcp_agent.py：测试mcp_agent.py模块的功能，包括MCP协议的通信、智能Agent的响应等

前端界面设计

前端界面需要实现以下功能：

文件上传区域
上传进度显示
润色过程实时显示
最终结果展示

<!DOCTYPE html>
<html>
<head><title>DOC文件润色</title><script src="https://cdn.jsdelivr.net/npm/socket.io-client@4.7.2/dist/socket.io.min.js"></script><script>const socket = io();// 文件上传处理document.getElementById('uploadBtn').addEventListener('click', function() {const fileInput = document.getElementById('docFile');const file = fileInput.files[0];if (file) {const formData = new FormData();formData.append('docFile', file);fetch('/upload', {method: 'POST',body: formData}).then(response => {// 处理响应});}});// 监听后端消息socket.on('progress', function(data) {document.getElementById('progress').textContent = `处理进度: ${data.percent}%`;document.getElementById('steps').innerHTML = data.steps.join('<br>');});socket.on('result', function(data) {document.getElementById('result').textContent = data.text;});</script>
</head>
<body><h1>DOC文件润色</h1><input type="file" id="docFile" accept=".doc"><button id="uploadBtn">上传并润色</button><div id="progress">处理进度: 0%</div><div id="steps">处理步骤:</div><div id="result"></div>
</body>
</html>

Flask后端实现

环境配置

首先需要安装必要的Python库：

pip install flask flask-socketio win32com-client

文件上传处理

from flask import Flask, request, send_from_directory
from flask_socketio import SocketIO
from win32com.client import Dispatch
import time
import json
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, cors_allowed_origins="*")
# 存储上传的文件
UPLOAD_FOLDER = 'uploads'
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
@app.route('/upload', methods=['POST'])
def upload_file():if 'docFile' not in request.files:return jsonify({'error': 'No file part'})file = request.files['docFile']if file.filename == '':return jsonify({'error': 'No selected file'})if file:filename = secure_filename(file.filename)file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)file.save(file_path)# 异步处理文件process_file.delay(file_path, filename)return jsonify({'success': True, 'filename': filename})return jsonify({'error': 'Unknown error'})
@socketio.on('connect')
def handle_connect():print('Client connected')
@socketio.on('disconnect')
def handle_disconnect():print('Client disconnected')
if __name__ == '__main__':socketio.run(app)

使用win32com读取DOC文件

import win32com.client as win32
import time
from flask import current_app
from flask_socketio import emit
def read_doc_file(file_path):word = win32.Dispatch('Word.Application')doc = word.Documents.Open(file_path)# 读取文本内容text = doc.Content.Text# 关闭文档和Word应用程序doc.Close()word.Quit()return text
def process_file(file_path, filename):# 模拟处理进度steps = []# 第一步：读取文件steps.append("正在读取DOC文件...")emit('progress', {'percent': 20, 'steps': steps}, room=request.sid)time.sleep(1)try:content = read_doc_file(file_path)# 第二步：通过MCP协议发送给智能Agent处理steps.append("正在通过MCP协议发送文本给智能Agent...")emit('progress', {'percent': 40, 'steps': steps}, room=request.sid)time.sleep(1)# 模拟MCP协议调用processed_text = mcp_agent.process_text(content)# 第三步：接收处理结果steps.append("正在接收处理结果...")emit('progress', {'percent': 80, 'steps': steps}, room=request.sid)time.sleep(1)# 发送最终结果emit('result', {'text': processed_text}, room=request.sid)# 清理临时文件os.remove(file_path)return jsonify({'success': True})except Exception as e:return jsonify({'error': str(e)})

MCP协议服务实现

MCP协议概述

模型上下文协议(MCP)是一个开放协议，标准化了应用程序如何为大型语言模型(LLMs)提供上下文。可以将MCP想象成AI应用的"USB-C接口"，它为AI应用提供了连接万物的接口[80]。
MCP协议的主要目的是解决当前AI模型因数据孤岛限制而无法充分发挥潜力的难题，使AI应用能够安全地访问和操作本地及远程数据[80]。

MCP协议实现

import json
import requests
class MCPClient:def __init__(self, server_url):self.server_url = server_urldef call(self, tool_name, arguments):payload = {"name": tool_name,"arguments": arguments}response = requests.post(f"{self.server_url}/mcp/call",headers={"Content-Type": "application/json","Accept": "application/json"},data=json.dumps(payload))if response.status_code == 200:return json.loads(response.text)else:raise Exception(f"API call failed with status code {response.status_code}")

定义服务接口和Schema

from pydantic import BaseModel, Field
class TextProcessingInput(BaseModel):text: str = Field(..., description="需要处理的文本内容")style: str = Field(..., description="需要的文本风格，例如正式、活泼等")
class TextProcessingOutput(BaseModel):text: str = Field(..., description="处理后的文本内容")changes: list = Field(..., description="文本修改记录")

智能Agent实现

AI模型选择与集成

对于文本润色任务，我们可以选择适合自然语言处理的模型，如GPT-3.5或LLama-2等。这些模型具有强大的文本理解和生成能力，能够完成文本润色、改写等任务。

文本润色改写算法

import time
class TextProcessor:def __init__(self, mcp_client):self.mcp_client = mcp_clientdef process_text(self, text):# 调用MCP服务处理文本result = self.mcp_client.call("text_processing.process",{"text": text,"style": "formal"  # 示例：正式风格})return result["text"]

与MCP协议的集成

class TextProcessingAgent:def __init__(self):passdef process(self, request):# 处理文本润色请求text = request["arguments"]["text"]style = request["arguments"].get("style", "formal")# 进行文本处理processed_text = self._process_text(text, style)return {"name": "text_processing.process","arguments": {"text": processed_text}}def _process_text(self, text, style):# 实现具体的文本处理逻辑# 这里只是一个示例，实际可以根据需要实现更复杂的逻辑if style == "formal":# 正式风格处理逻辑processed_text = f"正式风格的润色文本：{text}"else:# 其他风格处理逻辑processed_text = f"其他风格的润色文本：{text}"return processed_text