from openai import AzureOpenAI
import json
import os
from docx import Document
from PyPDF2 import PdfReader
CONFIG_FILE = "azure_config.json"
def load_azure_config():
"""加载 Azure OpenAI 配置,沿用 上一篇的容错处理,避免配置错误导致程序崩溃"""
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
config = json.load(f)
required_keys = ["azure_endpoint", "api_key", "deployment_name"]
if all(key in config for key in required_keys):
return config
else:
print("⚠️ Azure 配置文件不完整,将重新引导输入")
except Exception as e:
print(f"⚠️ 配置文件读取失败({str(e)}),将重新引导输入")
config = {
"azure_endpoint": input("请输入 Azure Endpoint:"),
"api_key": input("请输入 Azure OpenAI API Key:"),
"deployment_name": input("请输入 GPT-4 模型部署名称:")
}
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False, indent=4)
return config
config = load_azure_config()
client = AzureOpenAI(
azure_endpoint=config["azure_endpoint"],
api_key=config["api_key"],
api_version="2024-08-01-preview"
)
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)
content = "\n".join([para.text for para in doc.paragraphs if para.text.strip()])
elif file_ext == ".pdf":
reader = PdfReader(file_path)
content = "\n".join([page.extract_text() for page in reader.pages if page.extract_text().strip()])
elif file_ext == ".txt":
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
else:
return f"❌ 不支持该文件格式({file_ext}),仅支持.docx/.pdf/.txt"
if not content.strip():
return "❌ 文件内容为空,无法进行处理"
return content
except Exception as e:
return f"❌ 文件读取失败:{str(e)}(可能是文件损坏或格式异常)"
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": f"请总结以下文件内容:\n{text}"}
],
temperature=0.2
)
return response.choices[0].message.content.strip()
except Exception as e:
return f"❌ 总结失败:{str(e)}(请检查 Azure 配置或网络连接)"
def save_result(result: str, file_name: str = None) -> str:
"""保存处理结果到 TXT 文件,自动生成文件名,返回保存路径,延续 上一篇简洁实用的设计思路"""
import datetime
if not file_name:
now = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
file_name = f"agent_result_{now}.txt"
if not file_name.endswith(".txt"):
file_name += ".txt"
try:
with open(file_name, "w", encoding="utf-8") as f:
f.write(f"Agent 处理结果({datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')})\n")
f.write("=" * 50 + "\n")
f.write(result)
return f"✅ 结果已保存至:{os.path.abspath(file_name)}"
except Exception as e:
return f"❌ 结果保存失败:{str(e)}"
class OfficeAgent:
def __init__(self):
self.history = []
self.last_file_path = None
self.tools = {"读取文件": read_file, "总结文件": summary_tool, "保存结果": save_result}
def parse_intent(self, instruction: str) -> tuple:
"""智能解析用户指令,沿用 上一篇的关键词匹配逻辑,优化文件相关指令识别,无需死板输入指令头"""
instruction_lower = instruction.lower()
file_path = None
if any(ext in instruction for ext in [".docx", ".pdf", ".txt"]):
import re
pattern = r"([^\s]+?\.(docx|pdf|txt))"
match = re.search(pattern, instruction)
if match:
file_path = match.group(1)
self.last_file_path = file_path
else:
file_path = self.last_file_path
if any(keyword in instruction_lower for keyword in ["读取", "打开"]):
return "读取文件", file_path, instruction
elif any(keyword in instruction_lower for keyword in ["总结", "概括", "提炼"]):
return "总结文件", file_path, instruction
elif any(keyword in instruction_lower for keyword in ["保存", "导出"]):
return "保存结果", None, instruction
else:
return "未知", file_path, instruction
def run(self, instruction: str) -> str:
"""Agent 运行入口,串联「解析指令→调用工具→保存结果→记忆上下文」全流程,沿用 上一篇核心逻辑"""
print(f"🔍 Agent 接收指令:{instruction}")
intent, file_path, raw_instruction = self.parse_intent(instruction)
if intent == "读取文件":
if not file_path:
result = "❌ 未识别到文件路径,请输入包含.docx/.pdf/.txt 的文件路径(如:读取文件 D:/test.docx)"
else:
result = self.tools["读取文件"](file_path)
if not result.startswith("❌"):
self.last_file_path = file_path
self.history.append(("读取文件", file_path, result[:50] + "..."))
elif intent == "总结文件":
if not file_path:
result = "❌ 未识别到文件路径,请先读取文件(如:读取文件 D:/test.pdf),再输入「总结这个文件」"
else:
file_content = self.tools["读取文件"](file_path)
if file_content.startswith("❌"):
result = file_content
else:
result = self.tools["总结文件"](file_content)
self.history.append(("总结文件", file_path, result))
elif intent == "保存结果":
if not self.history:
result = "❌ 暂无可保存的结果,请先处理文件(读取/总结)"
else:
last_task, last_file, last_result = self.history[-1]
save_msg = self.tools["保存结果"](last_result)
result = f"📌 最近一次处理:{last_task}({last_file})\n{save_msg}"
else:
result = "❌ 未识别到任务,请输入以下类型的指令:\n1. 读取文件:包含.docx/.pdf/.txt 路径(如:读取 D:/会议纪要.docx)\n2. 总结文件:包含文件路径或先读取文件后输入「总结这个文件」\n3. 保存结果:保存最近一次的处理结果(如:保存结果)"
self.history.append(("用户指令", raw_instruction, result))
if len(self.history) > 5:
self.history.pop(0)
return result
if __name__ == "__main__":
office_agent = OfficeAgent()
print("✅ 职场办公型 Azure Agent 已启动(基于 GPT-4 驱动)")
print("📌 支持功能:读取 Word/PDF/TXT + 自动总结 + 保存结果(延续 上一篇记忆 + 多工具能力)")
print("📌 示例指令(直接复制使用):")
print(" 1. 读取文件 D:/docker/会议纪要.docx")
print(" 2. 总结文件 D:/docker/项目计划.pdf 或 先读取文件后输入「总结这个文件」")
print(" 3. 保存结果")
print("📌 输入「退出」即可结束运行")
while True:
user_input = input("\n请输入指令:")
if user_input in ["退出", "exit", "quit"]:
print("❌ Agent 已停止运行")
break
result = office_agent.run(user_input)
print(f"\nAgent:{result}\n")