1. 项目概述
在职场办公场景中,我们经常需要处理各种文档格式的文件(Word、PDF、TXT等),进行内容总结和归档。传统方式需要手动打开文件、复制内容、粘贴到AI工具中生成总结,最后再保存结果,整个过程繁琐且重复。本文将基于Azure OpenAI的GPT-4模型,构建一个能够自动完成"读取-总结-保存"全流程的办公助手型Agent。
这个Agent的核心价值在于:
- 支持多种办公文档格式的直接读取
- 利用GPT-4强大的自然语言处理能力生成专业总结
- 自动保存处理结果并记录操作历史
- 通过模块化设计实现功能的灵活扩展
2. 环境准备与配置
2.1 Azure OpenAI服务配置
首先需要确保已经开通了Azure OpenAI服务并部署了GPT-4模型。配置信息保存在azure_config.json文件中,包含以下关键参数:
json复制{
"azure_endpoint": "https://your-resource-name.openai.azure.com/",
"api_key": "your-api-key-here",
"deployment_name": "gpt-4"
}
提示:API密钥是敏感信息,建议不要直接硬编码在脚本中,使用配置文件更安全。
2.2 Python环境与依赖安装
项目需要以下Python库支持:
bash复制pip install python-docx PyPDF2 openai openpyxl
各库的作用:
python-docx:处理Word文档(.docx)PyPDF2:读取PDF文件内容openai:Azure OpenAI官方SDKopenpyxl:可选,用于Excel文件输出
3. 核心模块实现
3.1 文件读取模块
文件读取是Agent的基础功能,需要支持多种格式并具备良好的容错能力:
python复制def read_file(file_path: str) -> str:
"""读取文件内容,自动识别格式并返回纯文本"""
if not os.path.exists(file_path):
return f"❌ 文件不存在:{file_path}"
file_ext = os.path.splitext(file_path)[1].lower()
try:
if file_ext == ".docx":
doc = Document(file_path)
return "\n".join(para.text for para in doc.paragraphs if para.text.strip())
elif file_ext == ".pdf":
reader = PdfReader(file_path)
return "\n".join(page.extract_text() for page in reader.pages if page.extract_text())
elif file_ext == ".txt":
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
else:
return f"❌ 不支持的格式:{file_ext}"
except Exception as e:
return f"❌ 读取失败:{str(e)}"
注意事项:
- PDF文件的文本提取质量取决于原始文件的编码方式,扫描版PDF需要额外OCR处理
- 中文文档确保使用UTF-8编码读取,避免乱码
- 大文件(>10MB)建议先进行分块处理
3.2 文本总结模块
利用GPT-4生成专业、简洁的文档总结:
python复制def summary_tool(text: str) -> str:
"""使用GPT-4生成文本总结"""
try:
response = client.chat.completions.create(
model=config["deployment_name"],
messages=[
{"role": "system", "content": "你是一位专业的办公助手,请用简洁的语言总结以下内容,突出关键信息和行动项,控制在150字以内。"},
{"role": "user", "content": text}
],
temperature=0.2
)
return response.choices[0].message.content
except Exception as e:
return f"❌ 总结失败:{str(e)}"
优化提示词的技巧:
- 明确总结的长度要求
- 指定总结的风格(专业/简洁)
- 要求突出特定类型信息(如数据、决策、行动项)
- 设置较低的temperature值保证输出稳定性
3.3 结果保存模块
将处理结果持久化存储,方便后续查阅:
python复制def save_result(result: str, prefix: str = "result") -> str:
"""保存结果到文件,自动生成带时间戳的文件名"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{prefix}_{timestamp}.txt"
try:
with open(filename, "w", encoding="utf-8") as f:
f.write(f"处理时间:{timestamp}\n")
f.write("="*40 + "\n")
f.write(result)
return f"✅ 结果已保存到:{os.path.abspath(filename)}"
except Exception as e:
return f"❌ 保存失败:{str(e)}"
4. Agent核心逻辑实现
4.1 指令解析与路由
智能解析用户自然语言指令并路由到对应功能:
python复制def parse_instruction(instruction: str) -> dict:
"""解析用户指令,返回动作类型和参数"""
instruction = instruction.lower()
# 提取文件路径
file_match = re.search(r"([a-z]:[\\/][^\s]+\.(docx|pdf|txt))", instruction, re.I)
file_path = file_match.group(1) if file_match else None
# 判断指令类型
if "read" in instruction or "读取" in instruction:
return {"action": "read", "file_path": file_path}
elif "summar" in instruction or "总结" in instruction:
return {"action": "summarize", "file_path": file_path}
elif "save" in instruction or "保存" in instruction:
return {"action": "save"}
else:
return {"action": "unknown"}
4.2 带记忆的会话管理
维护对话历史实现上下文感知:
python复制class OfficeAgent:
def __init__(self):
self.history = []
self.last_content = None
def process(self, instruction: str) -> str:
"""处理用户指令"""
parsed = parse_instruction(instruction)
if parsed["action"] == "read":
content = read_file(parsed["file_path"])
if not content.startswith("❌"):
self.last_content = content
self.history.append(("read", parsed["file_path"]))
return content
elif parsed["action"] == "summarize":
if not self.last_content:
return "❌ 请先读取文件内容"
summary = summary_tool(self.last_content)
self.history.append(("summarize", summary))
return summary
elif parsed["action"] == "save":
if not self.history or self.history[-1][0] != "summarize":
return "❌ 没有可保存的总结结果"
return save_result(self.history[-1][1])
else:
return "❌ 无法识别的指令,支持:读取文件、总结内容、保存结果"
5. 实战测试与优化
5.1 典型测试案例
测试一个项目会议纪要.docx文件:
code复制2023年Q3项目启动会
时间:2023-07-15 14:00-15:30
参会人员:张三(PM)、李四(开发)、王五(设计)
会议决议:
1. 优先开发用户管理模块,7月底完成
2. 数据库采用MySQL 8.0,由李四负责搭建
3. UI设计稿需在7月20日前提交评审
4. 下次会议定于7月22日
Agent处理结果:
code复制【会议总结】
2023年Q3项目启动会确定了以下行动计划:
1. 用户管理模块为开发重点,7月底交付
2. MySQL 8.0数据库由李四负责实施
3. 王五需在7月20日前完成UI设计稿评审
4. 下次会议时间为7月22日
5.2 性能优化建议
- 大文件处理:
python复制# 分块处理大文本
def chunk_text(text, max_length=2000):
return [text[i:i+max_length] for i in range(0, len(text), max_length)]
# 分段总结后合并
chunks = chunk_text(large_content)
summaries = [summary_tool(chunk) for chunk in chunks]
final_summary = summary_tool("\n".join(summaries))
- 异步处理优化:
python复制import asyncio
async def async_summary(text):
# 使用异步API调用
response = await client.chat.completions.create(...)
return response.choices[0].message.content
- 缓存机制:
python复制from functools import lru_cache
@lru_cache(maxsize=100)
def cached_summary(text: str) -> str:
return summary_tool(text)
6. 扩展与进阶功能
6.1 支持更多文件类型
扩展文件读取能力:
python复制# 支持Excel文件
import pandas as pd
def read_excel(file_path: str) -> str:
df = pd.read_excel(file_path)
return df.to_string()
# 支持PPT文件
from pptx import Presentation
def read_ppt(file_path: str) -> str:
prs = Presentation(file_path)
return "\n".join(slide.text for slide in prs.slides if slide.text)
6.2 多语言支持
增强国际化能力:
python复制def detect_language(text: str) -> str:
"""简单语言检测"""
if re.search(r"[\u4e00-\u9fff]", text):
return "zh"
else:
return "en"
def multilingual_summary(text: str) -> str:
lang = detect_language(text)
system_prompt = {
"zh": "你是一位中文办公助手...",
"en": "You are a professional assistant..."
}.get(lang, "en")
# 调用GPT时使用对应的system prompt
6.3 集成到办公软件
通过COM接口集成到Office:
python复制import win32com.client
def word_addin():
word = win32com.client.Dispatch("Word.Application")
doc = word.ActiveDocument
content = doc.Content.Text
summary = summary_tool(content)
# 将总结插入文档
doc.Content.InsertAfter("\n\n【AI总结】\n" + summary)
7. 安全与最佳实践
- 敏感信息处理:
python复制def sanitize_output(text: str) -> str:
"""过滤敏感信息"""
patterns = [
r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}\b", # 信用卡号
r"\b\d{3}-\d{2}-\d{4}\b", # SSN
r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b" # 邮箱
]
for pattern in patterns:
text = re.sub(pattern, "[REDACTED]", text)
return text
- API调用限制:
python复制from ratelimit import limits, sleep_and_retry
@sleep_and_retry
@limits(calls=30, period=60) # 每分钟最多30次调用
def safe_api_call(text):
return client.chat.completions.create(...)
- 错误处理与重试:
python复制from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def reliable_summary(text):
try:
return summary_tool(text)
except Exception as e:
log_error(f"总结失败:{str(e)}")
raise
8. 部署与生产化建议
- 容器化部署:
dockerfile复制FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "office_agent.py"]
- 日志监控:
python复制import logging
from logging.handlers import RotatingFileHandler
logger = logging.getLogger("OfficeAgent")
logger.setLevel(logging.INFO)
handler = RotatingFileHandler("agent.log", maxBytes=5*1024*1024, backupCount=3)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
- 性能监控:
python复制from prometheus_client import start_http_server, Summary
REQUEST_TIME = Summary('request_processing_seconds', 'Time spent processing request')
@REQUEST_TIME.time()
def process_request(instruction):
# 处理逻辑
pass
# 启动指标服务器
start_http_server(8000)
9. 典型问题排查
- 文件读取失败:
- 检查文件路径是否正确(绝对路径/相对路径)
- 验证文件权限
- 确认文件没有被其他程序锁定
- API调用错误:
- 检查Azure终结点和API密钥
- 验证模型部署名称
- 确认区域可用性
- 中文乱码问题:
- 确保文件以UTF-8编码保存
- 在文件打开时明确指定encoding="utf-8"
- 检查系统区域设置
- 总结质量不佳:
- 优化system prompt
- 调整temperature参数(0.1-0.3更稳定)
- 提供更明确的用户指令
10. 未来扩展方向
- 批量处理能力:
python复制def batch_process(folder_path: str):
"""处理整个文件夹中的文档"""
for filename in os.listdir(folder_path):
if filename.endswith((".docx", ".pdf", ".txt")):
file_path = os.path.join(folder_path, filename)
content = read_file(file_path)
summary = summary_tool(content)
save_result(summary, prefix=filename.split(".")[0])
- 自定义模板输出:
python复制def save_as_template(result, template_name="default"):
templates = {
"meeting": "会议纪要:\n{content}\n参会人:\n行动项:",
"report": "报告摘要:\n{content}\n关键数据:\n结论:"
}
return templates.get(template_name, "{content}").format(content=result)
- 与Teams/Outlook集成:
python复制def process_outlook_email():
outlook = win32com.client.Dispatch("Outlook.Application")
mail = outlook.ActiveExplorer().Selection[0]
summary = summary_tool(mail.Body)
mail.Reply().Body = f"【自动总结】\n{summary}\n\n原始邮件内容:\n{mail.Body}"
这个Azure AI办公助手Agent已经具备了实用的核心功能,通过模块化设计可以方便地扩展更多办公场景支持。在实际使用中,建议先从简单的文档处理开始,逐步扩展到更复杂的业务流程自动化。
