【大模型开发】智能文档分析系统

基于我们之前打下的坚实的python基础我们先来写几个有意思的ai对话网页。我们设计的宗旨就是我们要有框架思想对于ai生成的内容要带有批判性的眼光看待这次我们先实现最简单的——智能文档分析系统顾名思义就是基于我们上传的文档可以上传多个来与ai进行对话从而可以实现复杂文件的内容提取等工作为后续大模型开发奠定了坚实的基础。下面即为可运行的完整版代码大家注意按照上篇文章的讲述将DEEPSEEK_API_KEY设置为自己的api_key。import os import tempfile import uuid from typing import List, Dict, Any from io import BytesIO from pathlib import Path from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi.responses import FileResponse, JSONResponse from pypdf import PdfReader from docx import Document from openai import OpenAI # ---------- 配置 ---------- DEEPSEEK_API_KEY 请填写你的api_key app FastAPI(title智能文档分析系统) # ---------- 会话存储简单内存存储生产环境应改用数据库 ---------- sessions: Dict[str, Dict[str, Any]] {} # 每个会话存储 # { # docs: [{name: str, text: str}], # 文档列表 # history: [{role: str, content: str}] # 对话历史 # } # ---------- 文档解析函数 ---------- def parse_txt(data: bytes) - str: return data.decode(utf-8, errorsreplace) def parse_pdf(data: bytes) - str: reader PdfReader(BytesIO(data)) text \n.join((page.extract_text() or ) for page in reader.pages) return text.strip() def parse_docx(data: bytes) - str: doc Document(BytesIO(data)) return \n.join(p.text for p in doc.paragraphs).strip() def extract_text(file: UploadFile) - str: 根据扩展名提取文本失败则抛出异常 data file.file.read() ext Path(file.filename or ).suffix.lower() if ext .txt: return parse_txt(data) elif ext .pdf: return parse_pdf(data) elif ext .docx: return parse_docx(data) else: raise ValueError(f不支持的文件类型: {ext}) # ---------- DeepSeek 调用封装 ---------- def call_deepseek(messages: List[Dict[str, str]]) - str: client OpenAI(api_keyDEEPSEEK_API_KEY, base_urlhttps://api.deepseek.com) response client.chat.completions.create( modeldeepseek-chat, messagesmessages, streamFalse, temperature0.7, max_tokens2000 ) return response.choices[0].message.content # ---------- 1. 上传接口创建会话并存储文档 ---------- app.post(/upload) async def upload_files(files: List[UploadFile] File(...)): 批量上传文件创建新会话返回 session_id。 if not files: raise HTTPException(400, 至少上传一个文件) session_id str(uuid.uuid4())# 生成一个全世界几乎不会重复的随机唯一字符串 docs [] errors [] for f in files: try: text extract_text(f) if text: docs.append({name: f.filename, text: text}) else: errors.append(f{f.filename}: 解析后文本为空) except Exception as e: errors.append(f{f.filename}: {str(e)}) if not docs: raise HTTPException(400, f没有成功提取任何文档内容。错误: {errors}) # 存储会话 sessions[session_id] { docs: docs, history: [] # 初始化空历史 } return JSONResponse({ session_id: session_id, uploaded: len(docs), errors: errors if errors else None }) # ---------- 2. 对话接口多轮对话基于文档上下文 ---------- app.get(/chat) async def chat(session_id: str, query: str): 基于会话中的文档内容进行多轮对话。 if session_id not in sessions: raise HTTPException(404, 会话不存在请先上传文件) session sessions[session_id] # 构建系统提示包含文档内容作为上下文 docs_text \n\n.join( f【文档 {idx1}{doc[name]}】\n{doc[text][:2000]} # 限制长度 for idx, doc in enumerate(session[docs]) ) system_prompt f你是一个专业的文档分析助手。以下是一些文档的内容请基于这些文档回答用户的问题。 文档内容 {docs_text} 请注意用户可能进行多轮对话你需要结合上下文历史来回答。 # 构建消息列表系统消息 历史 当前问题 messages [ {role: system, content: system_prompt} ] session[history] [ {role: user, content: query} ] # 调用 AI try: answer call_deepseek(messages) except Exception as e: raise HTTPException(500, fAI 调用失败: {e}) # 更新历史只保留用户和助手的消息 session[history].append({role: user, content: query}) session[history].append({role: assistant, content: answer}) return {answer: answer} # ---------- 3. 生成报告接口基于所有文档生成综合报告并下载 ---------- app.get(/report) async def generate_report(session_id: str): 为指定会话生成综合报告返回文本文件。 if session_id not in sessions: raise HTTPException(404, 会话不存在请先上传文件) session sessions[session_id] # 构建报告提示 docs_info \n\n.join( f文档{doc[name]}\n内容摘要{doc[text][:1000]}... # 可以只取摘要 for doc in session[docs] ) prompt f请根据以下文档内容生成一份综合性的分析报告。报告应包括 1. 文档主题总结 2. 关键观点提炼 3. 各文档之间的关联或差异 4. 最终结论 文档内容 {docs_info} messages [{role: user, content: prompt}] try: report call_deepseek(messages) except Exception as e: raise HTTPException(500, f生成报告失败: {e}) # 将报告保存为临时文件并返回 with tempfile.NamedTemporaryFile(modew, suffix.txt, deleteFalse) as tmp: tmp.write(report) tmp_path tmp.name return FileResponse( pathtmp_path, filenamefreport_{session_id}.txt, media_typetext/plain ) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)接下来向大家讲述如何运行1.首先激活自己的虚拟环境我的虚拟环境名称是my_llm_envconda activate my_llm_env2.进入代码所在目录cd (这里写你复制的完整路径3.运行代码为了方便我一般都是把鼠标放置在代码处然后右击运行如下图4.进入网址点击下面窗口中的链接即可这个网址一共有五组数字前四组代表自己的主机最后一个数字可以自己定义代表端口。5.因为我们没有写前端代码所以要通过FastAPI提供的docs来检查我们撰写的代码是否正确。我们打开网址之后将路径改为如下图所示即在最后添加/docs。6.接下来给大家讲解一下如何使用。我们打开之后会发现我们实现了三个功能可以点击这三个带颜色的框它们的功能分别是上传文件、ai对话、生成报告。我们想要测试第一个功能是否正常我们点击了之后显示上面的页面接下来点击Try it out就可以上传文件。上传之后我们点击Execute即可上传成功我们也可以点击蓝框的内容实现多文档上传绿框中的内容是取消上传文件、重置现在上传的内容。我们将网页向上滑动会出现下面的页面这个红框中的session_id在下面的与ai聊天中需要用到。接下来我们点击chat模块和之前一样先点击Try it out然后将上面的session_id和你想要询问的问题输入点击Execute即可。效果如下图可能会有点慢大家耐心等几秒钟第三个Generate Report模块与之前的操作相同大家单击模块然后单击Try it out再输入session_id即可生成报告可以下载下来看一下。好啦一个简单的文档分析系统就完成啦之后我应该会再写两个带有前端功能的小系统这样比较美观。主旨还是训练我们的框架设计思维并巩固我们学过的python基础为后面的大模型开发奠定坚实基础。大家不要着急觉得学这些有什么用呀但只有我们把基础打牢了后面学起来才毫不费力期待大家跟我一起学习大模型开发一起找到满意的工作ps最近清明节假期先休息一下等到假期结束我就更新其他两个小系统~~~