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

Langchain从零开始到应用落地案例[AI智能助手]【5】---实现基于FastAPI+Langchain+ocr实现上下文对话大模型对话接口

news 2025/10/29 12:35:41

前言

环境安装

前提代码编写准备

端口代码编写

示例代码

运行效果演示

错误问题

前言

在之前的篇章已经实现langchain大模型对话，然后ocr文件识别，不过都只能在控制台实现效果操作。

接下来，咱们一起学习如何在FastAPI编写请求接口，为之后部署服务和页面上做好准备。

在这里，我的代码不一定适合所有人运行，只作为参考，不同的环境和版本框架内容不一定能运行，这里先说明了一下，如有需要后续代码会上传到github上提供下载和参考，这里谢谢大家了

环境安装

Fastapi环境

pip install fastapi

uvicorn一个ASGI服务器，用来运行fatapi端口

pip install uvicorn

前提代码编写准备

首先就是fastapi的实例化和服务器设置（在保证端口的不冲突下）

实例化和中间建设置

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)

然后uvicorn服务器设置

if __name__ == '__main__':import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

端口代码编写

首先创建请求路径和传递参数

@app.post("/chat_llm")
async def chat_llm(question: str = Form(...),session_id: Optional[str] = Form(None),file_path: Optional[UploadFile] = None,
):

然后会话session_id定义

# 1. 确保 session_id 存在
if not session_id:session_id = str(uuid.uuid4())# 2. 初始化会话历史（如果不存在）
if session_id not in memory_store:memory_store[session_id] = ChatMessageHistory()

对接ocr进行文件内容识别输出

# 3. 处理文件上传（OCR）ocr_content = ""temp_file_path = Nonetry:if file_path is not None and hasattr(file_path, 'filename') and file_path.filename:# 创建临时文件suffix = f".{file_path.filename.split('.')[-1]}" if '.' in file_path.filename else ""with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as temp_file:content = await file_path.read()temp_file.write(content)temp_file_path = temp_file.name# 执行 OCRfile_text = ocr_file(temp_file_path)if file_text and file_text.strip():ocr_content = f"【文件内容】\n{file_text}\n\n"else:ocr_content = "【文件内容】\n（未识别到有效文本）\n\n"finally:# 清理临时文件if temp_file_path and os.path.exists(temp_file_path):os.unlink(temp_file_path)

然后最后面的大模型拼接输出

默认提示词

SYSTEM_PROMPT = ("你是一个问答助手。请根据用户的提问和可能提供的文件内容进行回答。""如果用户上传了文件，请结合文件内容回答问题；否则仅根据问题回答。"
)

函数内部

 # 4. 构建完整消息列表messages = [("system", SYSTEM_PROMPT)]messages.extend(memory_store[session_id].messages)# 将 OCR 内容拼接到用户问题前（作为当前轮次的上下文）current_user_message = ocr_content + questionmessages.append(("human", current_user_message))# 5. 调用模型prompt = ChatPromptTemplate.from_messages(messages)response = (prompt | Setting_llm).invoke({})# 6. 保存原始问题（不含 OCR）到历史，避免重复污染memory_store[session_id].add_user_message(question)memory_store[session_id].add_ai_message(response.content)

示例代码

@app.post("/chat_llm")
async def chat_llm(question: str = Form(...),session_id: Optional[str] = Form(None),file_path: Optional[UploadFile] = None,
):# 1. 确保 session_id 存在if not session_id:session_id = str(uuid.uuid4())# 2. 初始化会话历史（如果不存在）if session_id not in memory_store:memory_store[session_id] = ChatMessageHistory()# 3. 处理文件上传（OCR）ocr_content = ""temp_file_path = Nonetry:if file_path is not None and hasattr(file_path, 'filename') and file_path.filename:# 创建临时文件suffix = f".{file_path.filename.split('.')[-1]}" if '.' in file_path.filename else ""with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as temp_file:content = await file_path.read()temp_file.write(content)temp_file_path = temp_file.name# 执行 OCRfile_text = ocr_file(temp_file_path)if file_text and file_text.strip():ocr_content = f"【文件内容】\n{file_text}\n\n"else:ocr_content = "【文件内容】\n（未识别到有效文本）\n\n"finally:# 清理临时文件if temp_file_path and os.path.exists(temp_file_path):os.unlink(temp_file_path)# 4. 构建完整消息列表messages = [("system", SYSTEM_PROMPT)]messages.extend(memory_store[session_id].messages)# 将 OCR 内容拼接到用户问题前（作为当前轮次的上下文）current_user_message = ocr_content + questionmessages.append(("human", current_user_message))# 5. 调用模型prompt = ChatPromptTemplate.from_messages(messages)response = (prompt | Setting_llm).invoke({})# 6. 保存原始问题（不含 OCR）到历史，避免重复污染memory_store[session_id].add_user_message(question)memory_store[session_id].add_ai_message(response.content)return {"session_id": session_id,"response": response.content}