一、新增依赖
根据需求安装对应包：
# 音频转写 + 文档解析 + 基础依赖
pip install pypdf python-docx pydub speechrecognition说明：
python-docx：解析 Word 会议文档pypdf：解析 PDF 纪要speechrecognition+pydub：本地音频转文字（纯离线简易版）；长音频 / 高准确率建议对接阿里云 / 讯飞听见转写 API
二、完整扩展代码
在原有 work_agent.py 基础上，新增会议处理专属工具，并更新提示词、调用逻辑，实现：读取会议文件 / 音频 → 提取会议要点 → 梳理待办事项 → 生成正式纪要 → 推送至群聊
1. 完整整合代码
import os
import requests
import speech_recognition as sr
from pydub import AudioSegment
from dotenv import load_dotenv
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from docx import Document
from langchain_community.document_loaders import PyPDFLoader
# 加载环境变量
load_dotenv()
DASHSCOPE_KEY = os.getenv("DASHSCOPE_API_KEY")
WEBHOOK_URL = os.getenv("WECHAT_WEBHOOK")
# 初始化大模型
llm = ChatOpenAI(
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key=DASHSCOPE_KEY,
model="qwen-turbo",
temperature=0.2
)
# ====================== 原有工具（保留：查数据、读普通文档、发消息） ======================
@tool
def read_local_doc(file_path: str) -> str:
"""读取本地txt/md文本文档"""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
except Exception as e:
return f"读取失败：{str(e)}"
@tool
def query_work_data(date: str = "今日") -> dict:
"""查询工作业务数据"""
mock_data = {
"统计时间": date,
"完成任务数": 8,
"迭代需求数": 3,
"线上问题处理": 2,
"会议参与次数": 4,
"待跟进事项": 2
}
return mock_data
@tool
def send_group_msg(content: str) -> str:
"""企业微信群发消息"""
headers = {"Content-Type": "application/json"}
payload = {"msgtype": "text", "text": {"content": content}}
try:
res = requests.post(WEBHOOK_URL, json=payload, headers=headers)
return "消息发送成功" if res.json().get("errcode") == 0 else f"发送失败：{res.text}"
except Exception as e:
return f"网络异常：{str(e)}"
# ====================== 新增：会议记录专属工具 ======================
@tool
def read_meeting_word(file_path: str) -> str:
"""读取Word格式会议记录文档"""
try:
doc = Document(file_path)
return "\n".join([para.text for para in doc.paragraphs])
except Exception as e:
return f"读取Word失败：{str(e)}"
@tool
def read_meeting_pdf(file_path: str) -> str:
"""读取PDF格式会议记录文档"""
try:
loader = PyPDFLoader(file_path)
pages = loader.load()
return "\n".join([p.page_content for p in pages])
except Exception as e:
return f"读取PDF失败：{str(e)}"
@tool
def audio_to_text(audio_path: str) -> str:
"""
会议音频转文字（支持wav格式）
:param audio_path: 音频文件路径
"""
try:
# 统一转为wav格式
sound = AudioSegment.from_file(audio_path)
wav_path = "temp_meeting.wav"
sound.export(wav_path, format="wav")
r = sr.Recognizer()
with sr.AudioFile(wav_path) as source:
audio = r.record(source)
# 中文识别
text = r.recognize_google(audio, language="zh-CN")
os.remove(wav_path)
return f"音频转写结果：\n{text}"
except Exception as e:
return f"音频转写失败：{str(e)}"
# 整合全部工具
tools = [
read_local_doc,
read_meeting_word,
read_meeting_pdf,
audio_to_text,
query_work_data,
send_group_msg
]
# ====================== 提示词（重点：新增会议处理规则） ======================
system_prompt = """
你是全能办公助手，严格按照指令选择工具并执行：
1. 读取会议文件：Word调用read_meeting_word，PDF调用read_meeting_pdf，纯文本调用read_local_doc，音频调用audio_to_text；
2. 拿到会议原始内容后，必须完成三件事：①提炼会议核心议题 ②梳理决议内容 ③拆分待办事项（标注责任人/截止时间）；
3. 按照职场规范排版会议纪要，结构清晰；
4. 如需推送，调用send_group_msg发送完整纪要；
5. 生成周报调用query_work_data获取数据。
"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# 组装Agent + 对话记忆
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
session_store = {}
def get_history(session_id: str) -> InMemoryChatMessageHistory:
if session_id not in session_store:
session_store[session_id] = InMemoryChatMessageHistory()
return session_store[session_id]
agent_with_memory = RunnableWithMessageHistory(
agent_executor,
get_history,
input_messages_key="input",
history_messages_key="chat_history"
)
# ====================== 测试入口 ======================
if __name__ == "__main__":
session_id = "work_session_001"
# 按需取消注释测试不同场景
# 场景1：读取Word会议记录，生成纪要
# agent_with_memory.invoke(
# {"input": "读取 meeting.docx，生成标准会议纪要"},
# config={"configurable": {"session_id": session_id}}
# )
# 场景2：读取PDF会议记录，提取要点和待办
# agent_with_memory.invoke(
# {"input": "读取 meeting.pdf，总结要点并列出所有待办任务"},
# config={"configurable": {"session_id": session_id}}
# )
# 场景3：会议音频转文字 + 生成纪要
# agent_with_memory.invoke(
# {"input": "将 meeting.wav 转文字，并整理成正式会议纪要"},
# config={"configurable": {"session_id": session_id}}
# )
# 场景4：全流程：读取会议文件 → 生成纪要 → 发送到工作群（最常用）
agent_with_memory.invoke(
{"input": "读取 meeting.docx，生成会议纪要并发送到工作群"},
config={"configurable": {"session_id": session_id}}
)二、核心使用场景 & 指令
直接修改 input 内容，实现不同自动化需求：
- 仅解析文档生成纪要
读取 meeting.docx，生成标准格式会议纪要 - 提取要点 + 待办清单
读取 meeting.pdf，提炼会议要点，逐条列出待办事项 - 音频转写 + 整理纪要
把 meeting.wav 转成文字，整理为会议记录 - 全自动化（解析→纪要→发群）
读取 meeting.docx，生成纪要并发送到工作群 - 结合会议内容 + 业务数据写周报
读取本周会议记录.docx，结合业务数据生成周报并推送
三、关键模块优化说明
1. 音频转写升级（推荐生产使用）
本地 speechrecognition 准确率一般、仅支持短音频，企业场景替换为商用 API（讯飞听见 / 阿里云语音）：
- 优点：支持长音频、实时转写、方言识别、准确率高
- 改造方式：重写
audio_to_text函数，调用对应厂商 HTTP API 即可。
2. 自定义纪要格式
修改 system_prompt 固定排版模板，示例：
会议纪要模板：
一、会议主题
二、参会人员
三、主要议题与讨论内容
四、会议决议
五、待办任务（任务内容 | 负责人 | 完成时限）把以上模板写入系统提示词，Agent 会严格按格式输出。
3. 自动分配待办（进阶）
如果会议记录里有人名，大模型会自动识别并绑定任务；也可配合企业通讯录，实现待办自动提醒。
4. 定时 / 触发执行
- 定时执行：沿用之前
schedule库，每天固定时间扫描会议文件夹，自动处理新文件； - 文件夹监听：使用
watchdog库，监听指定目录，新增会议文件后自动触发处理：
pip install watchdog四、常见问题排查
- Word/PDF 读取乱码 确认文件非加密，路径使用绝对路径，文档内容正常可打开。
- 音频转写报错
优先使用
wav格式；Windows 需额外安装 ffmpeg 用于格式转换。 - 纪要内容杂乱：强化系统提示词，明确要求分模块输出、过滤闲聊内容。
- 消息发送失败：检查企业微信 / 飞书 Webhook 地址、服务器是否能访问外网。
五、拓展方向
- 对接会议软件（飞书会议 / 腾讯会议）：拉取会议录像 / 纪要，实现会议结束自动触发处理；
- 结合 RAG：将历史会议记录入库，支持按关键词检索过往会议内容；
- 待办跟踪：将提取的待办同步到 Jira / 飞书项目，形成闭环管理。
