1. 项目背景与核心概念在传统的法律援助服务中用户常常面临咨询渠道不畅、专业律师资源有限、等待时间长以及信息检索效率低下等问题。随着人工智能技术的飞速发展特别是大语言模型LLM的涌现为构建智能化、即时化的在线法律服务平台提供了全新的可能。然而LLM本身存在“幻觉”问题可能生成不准确或虚构的法律条文这对于严谨的法律咨询场景是致命的。因此一个理想的解决方案是将LLM的通用对话能力与精准的法律知识库相结合。本项目“基于AILLM的法律援助在线咨询平台”正是为此而生。它通过整合PostgreSQL 向量数据库、LLM大模型和WebSocket即时通讯三大核心技术构建了一个能够提供专业、实时、有据可查的智能法律咨询系统。其核心工作流程是当用户提出法律问题时系统首先利用向量检索技术从结构化的法律知识库中精准找到相关法条和案例然后将这些“证据”与用户问题一同提交给LLM由LLM生成基于事实、引用准确的回答并通过WebSocket实时推送给用户。接下来我们逐一拆解这几个核心概念LLM大语言模型如 GPT、ChatGLM、通义千问等它们拥有强大的自然语言理解和生成能力是本平台的“大脑”负责理解用户意图并组织语言生成最终回复。检索增强生成RAG这是本项目的核心架构思想。RAG通过“先检索后生成”的方式有效解决了LLM的幻觉和知识更新滞后问题。具体到本项目就是先从法律知识库中检索出最相关的法律条文、司法解释或判例再将它们作为上下文提供给LLM约束其生成内容确保回答的专业性和准确性。PostgreSQL 与 pgvectorPostgreSQL 是一个功能强大的开源关系型数据库。pgvector是其一个扩展插件使其具备了存储和高效检索向量数据的能力。我们将法律文本通过嵌入模型转换为高维向量即语义编码存入 PostgreSQL从而构建起本项目的“法律知识大脑”。当用户提问时将问题同样转换为向量并在数据库中进行相似度搜索快速找到语义上最相关的法律知识。WebSocket一种在单个TCP连接上进行全双工通信的协议。相比于传统的HTTP轮询WebSocket能实现服务器主动向客户端推送消息非常适合在线聊天、实时通知等场景。在本平台中它用于建立用户与AI助手之间稳定、低延迟的对话通道实现流畅的即时咨询体验。简单来说这个平台就像一个拥有海量法律典籍且过目不忘的“AI律师助理”。它不仅能和你实时对话WebSocket还能随时翻查最权威的法律条文PostgreSQL 向量检索并用自己的话LLM清晰、准确地为你解答。2. 技术栈选型与环境准备为了成功复现本项目你需要准备好以下开发环境与技术组件。版本号以当前稳定版为例你可以根据实际情况调整。2.1 核心环境与工具操作系统Ubuntu 20.04 LTS / Windows 10 / macOS (建议在Linux环境下部署生产环境)Python: 3.9 或 3.10 (这是多数AI框架兼容性较好的版本)Java(可选)如果后端使用Spring Boot需 JDK 11 或 17Node.js(可选)如果前端使用Vue/React需 Node.js 16Docker Docker Compose(强烈推荐)用于快速部署数据库等中间件保证环境一致性。2.2 后端技术栈我们将以一个Python FastAPI后端为例进行讲解它轻量、异步非常适合AI应用和WebSocket。Web框架FastAPI(用于构建RESTful API和WebSocket端点)数据库与向量检索PostgreSQL15主数据库存储用户信息、对话记录等。pgvector扩展为PostgreSQL添加向量能力。ORM/数据库工具SQLAlchemyPython SQL工具包和ORM。asyncpg或psycopg2PostgreSQL适配器。alembic数据库迁移工具。大模型接入openai库 (如果使用OpenAI API) 或zhipuai(智谱AI) 等。也可使用本地模型如通过transformers库加载ChatGLM3、Qwen等。文本嵌入模型sentence-transformers用于将法律文本和用户问题转换为向量。推荐模型all-MiniLM-L6-v2(轻量英文) 或paraphrase-multilingual-MiniLM-L12-v2(多语言)。其他工具库langchain(可选)用于简化RAG流程的编排但为了理解原理本文会从底层实现开始。uvicornASGI服务器用于运行FastAPI应用。python-dotenv管理环境变量。2.3 前端技术栈 (示例)框架Vue 3 TypeScriptUI库Element Plus 或 Ant Design VueWebSocket客户端原生WebSocketAPI 或vue-use-websocketHTTP客户端axios2.4 项目初始化与环境配置首先创建项目目录并设置Python虚拟环境。# 创建项目目录 mkdir ai-law-consultation-platform cd ai-law-consultation-platform # 创建后端目录 mkdir backend cd backend # 创建Python虚拟环境 (Linux/macOS) python3 -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 创建 requirements.txt 并安装核心依赖 cat requirements.txt EOF fastapi0.104.1 uvicorn[standard]0.24.0 sqlalchemy2.0.23 asyncpg0.29.0 alembic1.12.1 python-dotenv1.0.0 sentence-transformers2.2.2 openai1.3.0 # 如果使用OpenAI API # transformers4.36.0 # 如果使用本地模型 # torch # 根据transformers需求安装 EOF pip install -r requirements.txt2.5 使用 Docker 启动 PostgreSQL pgvector这是最关键的一步。我们使用Docker Compose来定义和运行服务。在项目根目录 (ai-law-consultation-platform/) 创建docker-compose.yml文件version: 3.8 services: postgres: image: ankane/pgvector:latest # 这个镜像已包含pgvector扩展 container_name: law_pg_vector environment: POSTGRES_USER: law_admin POSTGRES_PASSWORD: your_secure_password_here POSTGRES_DB: law_consultation ports: - 5432:5432 volumes: - postgres_data:/var/lib/postgresql/data - ./init.sql:/docker-entrypoint-initdb.d/init.sql # 初始化脚本 restart: unless-stopped volumes: postgres_data:同时创建数据库初始化脚本init.sql用于创建我们需要的扩展和表结构初版-- 启用 pgvector 扩展 CREATE EXTENSION IF NOT EXISTS vector; -- 创建一个用于测试的法律知识表后续会丰富 CREATE TABLE IF NOT EXISTS law_articles ( id BIGSERIAL PRIMARY KEY, title TEXT NOT NULL, -- 法律条文标题 content TEXT NOT NULL, -- 法律条文内容 source TEXT, -- 来源如《民法典》第xxx条 embedding vector(384) -- 向量字段维度需与嵌入模型匹配例如 all-MiniLM-L6-v2 是384维 ); -- 为向量字段创建索引以加速检索使用IVFFlat或HNSW这里用IVFFlat示例 CREATE INDEX ON law_articles USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);重要说明vector(384)中的维度384必须与你选用的sentence-transformers模型输出维度一致。all-MiniLM-L6-v2模型输出维度就是384。启动数据库服务# 在项目根目录执行 docker-compose up -d执行后一个包含了pgvector扩展的PostgreSQL数据库就在本地的5432端口运行起来了。3. 核心原理与模块拆解在开始编码前我们需要深入理解系统是如何协同工作的。整个平台可以划分为四个核心模块其交互流程如下图所示概念图用户提问 | v [前端] --(WebSocket)-- [后端API/WebSocket服务] | | | v | [问题向量化] | | | v | [向量检索 (PostgreSQL pgvector)] | | | v | [检索结果 (相关法条)] | | | v | [构造Prompt 调用LLM] | | | v | [生成最终答案] | | | v [前端] --(WebSocket)-- [流式/非流式返回答案]3.1 知识库构建与向量化模块这是RAG的“知识底座”。法律条文、案例等非结构化文本需要被转换成计算机能理解的“语义向量”并存储起来。文本预处理清洗法律文本去除无关字符进行分段例如按法条拆分。向量化嵌入使用sentence-transformers模型将每一段文本转换为一个固定维度的浮点数向量。这个向量在数学空间中的“位置”代表了文本的语义。向量存储将(文本内容, 向量)对存入PostgreSQL的law_articles表。pgvector扩展允许我们直接存储vector类型的数据。为什么是向量检索传统的数据库关键词检索如LIKE或全文索引依赖于词汇匹配无法理解“借款合同”和“借贷协议”的语义相似性。向量检索通过比较向量之间的“距离”如余弦相似度能找出语义上最相近的内容即使它们没有相同的字词。3.2 检索增强生成RAG流程模块这是本项目的“智能引擎”。当用户提问时查询向量化将用户问题Q通过相同的嵌入模型转换为向量V_q。语义检索在PostgreSQL中执行近似最近邻搜索找出与V_q余弦相似度最高的前k条法律条文。SQL语句类似于SELECT id, title, content, source, 1 - (embedding ‘[V_q]‘) AS similarity FROM law_articles ORDER BY embedding ‘[V_q]‘ LIMIT 5;是pgvector提供的余弦距离运算符。提示词工程将检索到的法律条文{doc1, doc2, ...}和用户问题Q组合成一个结构化的提示词Prompt提交给LLM。你是一个专业的法律AI助手请严格根据以下提供的法律条文来回答问题。 如果提供的条文不足以回答问题请明确告知“根据现有资料无法回答”。 【相关法律条文】 1. 《民法典》第xxx条... 2. 《合同法》第yyy条... 【用户问题】 Q 【请回答】LLM生成LLM基于这个充满“证据”的上下文生成最终答案。这极大地减少了幻觉提高了答案的准确性和可信度。3.3 实时通讯模块为了提供类聊天的体验我们采用WebSocket。连接建立前端通过ws://your-domain/ws/{session_id}与后端建立持久化连接。对话管理后端需要维护会话状态可能包括关联用户ID、保存对话历史上下文等。消息路由前端发送一个包含问题的JSON消息。后端接收后触发RAG流程并将生成的结果通过同一条WebSocket连接推回前端。流式输出高级为了体验更佳可以让LLM以流式stream方式生成文本后端每生成一段就立刻推送给前端实现打字机效果。3.4 数据持久化模块使用SQLAlchemy ORM来管理用户信息注册、登录本文不展开认证细节。对话会话每个咨询会话的元数据。消息记录用户与AI的每一轮问答关联到会话。这对于后续分析、模型优化和上下文管理至关重要。法律知识库即前面提到的law_articles表。4. 完整实战从零搭建核心后端服务现在我们将一步步实现上述核心模块。我们将创建一个简单的后端包含知识库入库、RAG检索和WebSocket对话接口。4.1 项目结构backend/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── database.py # 数据库连接与ORM定义 │ ├── models.py # SQLAlchemy数据模型 │ ├── schemas.py # Pydantic数据验证模型 │ ├── services/ │ │ ├── __init__.py │ │ ├── embedding_service.py # 文本向量化服务 │ │ ├── retrieval_service.py # 向量检索服务 │ │ └── llm_service.py # LLM调用服务 │ ├── api/ │ │ ├── __init__.py │ │ ├── endpoints.py # RESTful API (如知识库管理) │ │ └── websocket.py # WebSocket处理逻辑 │ └── core/ │ └── config.py # 配置管理 ├── alembic/ # 数据库迁移目录后续生成 ├── requirements.txt ├── .env.example └── docker-compose.yml # 位于项目根目录4.2 配置与数据库模型首先设置配置。创建app/core/config.pyfrom pydantic_settings import BaseSettings class Settings(BaseSettings): # 数据库配置 DATABASE_URL: str postgresqlasyncpg://law_admin:your_secure_password_herelocalhost:5432/law_consultation # 嵌入模型配置 EMBEDDING_MODEL: str sentence-transformers/all-MiniLM-L6-v2 EMBEDDING_DEVICE: str cpu # 或 cuda # LLM配置 (以OpenAI为例) OPENAI_API_KEY: str OPENAI_BASE_URL: str https://api.openai.com/v1 # 或代理地址 OPENAI_MODEL: str gpt-3.5-turbo # 检索配置 RETRIEVAL_TOP_K: int 3 # 每次检索返回的最相关条文数量 class Config: env_file .env settings Settings()创建.env文件参考.env.example并填入你的实际配置尤其是数据库密码和API密钥。定义数据模型app/models.pyfrom sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql import func from pgvector.sqlalchemy import Vector Base declarative_base() class LawArticle(Base): __tablename__ law_articles id Column(Integer, primary_keyTrue, indexTrue) title Column(String(500), nullableFalse) content Column(Text, nullableFalse) source Column(String(300)) embedding Column(Vector(384)) # 维度与模型匹配 created_at Column(DateTime(timezoneTrue), server_defaultfunc.now()) class Conversation(Base): __tablename__ conversations id Column(Integer, primary_keyTrue, indexTrue) user_id Column(String(100), indexTrue) # 简化处理实际应关联用户表 title Column(String(255)) # 会话标题可由首条问题生成 created_at Column(DateTime(timezoneTrue), server_defaultfunc.now()) class Message(Base): __tablename__ messages id Column(Integer, primary_keyTrue, indexTrue) conversation_id Column(Integer, ForeignKey(conversations.id, ondeleteCASCADE), nullableFalse) role Column(String(20), nullableFalse) # user or assistant content Column(Text, nullableFalse) created_at Column(DateTime(timezoneTrue), server_defaultfunc.now())创建数据库连接会话app/database.pyfrom sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from app.core.config import settings # 创建异步引擎 engine create_async_engine( settings.DATABASE_URL, echoTrue, # 开发时显示SQL日志生产环境应关闭 futureTrue ) # 创建异步会话工厂 AsyncSessionLocal sessionmaker( engine, class_AsyncSession, expire_on_commitFalse ) # 依赖注入用的获取会话函数 async def get_db(): async with AsyncSessionLocal() as session: try: yield session finally: await session.close()4.3 实现核心服务层1. 嵌入服务app/services/embedding_service.pyfrom sentence_transformers import SentenceTransformer import numpy as np from app.core.config import settings import logging logger logging.getLogger(__name__) class EmbeddingService: _model None classmethod async def get_model(cls): 懒加载嵌入模型 if cls._model is None: logger.info(fLoading embedding model: {settings.EMBEDDING_MODEL}) cls._model SentenceTransformer(settings.EMBEDDING_MODEL, devicesettings.EMBEDDING_DEVICE) return cls._model classmethod async def embed_texts(cls, texts: list[str]) - list[list[float]]: 将文本列表转换为向量列表 model await cls.get_model() # 模型.encode是同步的在异步环境中使用run_in_executor避免阻塞事件循环 import asyncio from functools import partial loop asyncio.get_event_loop() embeddings await loop.run_in_executor(None, partial(model.encode, texts, convert_to_numpyTrue, normalize_embeddingsTrue)) return embeddings.tolist() classmethod async def embed_single(cls, text: str) - list[float]: 将单条文本转换为向量 embeddings await cls.embed_texts([text]) return embeddings[0]2. 检索服务app/services/retrieval_service.pyfrom sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from pgvector.sqlalchemy import cosine_distance from app.models import LawArticle from app.services.embedding_service import EmbeddingService from app.core.config import settings import logging logger logging.getLogger(__name__) class RetrievalService: staticmethod async def retrieve_relevant_articles(query: str, db: AsyncSession, top_k: int None) - list[LawArticle]: 根据用户问题检索最相关的法律条文。 返回LawArticle 对象列表 if top_k is None: top_k settings.RETRIEVAL_TOP_K # 1. 将查询文本向量化 query_embedding await EmbeddingService.embed_single(query) # 2. 构建向量检索SQL查询 (使用余弦距离距离越小越相似) stmt ( select(LawArticle) .order_by(LawArticle.embedding.cosine_distance(query_embedding)) .limit(top_k) ) # 3. 执行查询 result await db.execute(stmt) articles result.scalars().all() logger.info(fRetrieved {len(articles)} articles for query: {query[:50]}...) return articles3. LLM服务app/services/llm_service.py这里以OpenAI API为例。如果你使用本地模型需要调整调用方式。from openai import AsyncOpenAI from app.core.config import settings import logging logger logging.getLogger(__name__) class LLMService: def __init__(self): self.client AsyncOpenAI( api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_BASE_URL, ) async def generate_answer(self, query: str, contexts: list[str]) - str: 基于用户问题和检索到的上下文调用LLM生成答案。 Args: query: 用户问题 contexts: 检索到的法律条文内容列表 Returns: LLM生成的答案文本 if not settings.OPENAI_API_KEY: return LLM服务未配置。请检查API密钥设置。 # 构造Prompt context_str \n\n.join([f{i1}. {ctx} for i, ctx in enumerate(contexts)]) system_prompt 你是一个专业的中国法律AI助手必须严格根据用户提供的法律条文来回答问题。 你的回答应当准确、清晰、有条理并引用相关条文。 如果提供的条文不足以回答用户的问题请诚实地告知“根据提供的资料我无法给出确切答案”并建议用户咨询专业律师。 禁止编造法律条文或案例。 user_prompt f【用户问题】 {query} 【相关法律条文】 {context_str} 请根据以上条文回答用户的问题。 try: response await self.client.chat.completions.create( modelsettings.OPENAI_MODEL, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt} ], temperature0.1, # 低温度使输出更确定、更少创造性 max_tokens1000, ) answer response.choices[0].message.content return answer.strip() except Exception as e: logger.error(f调用LLM API失败: {e}) return f生成回答时出现错误{str(e)}4.4 实现WebSocket端点与RAG流程整合创建app/api/websocket.py这是整个实时咨询功能的核心。from fastapi import WebSocket, WebSocketDisconnect, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.services.retrieval_service import RetrievalService from app.services.llm_service import LLMService from app.models import Conversation, Message import json import logging import uuid logger logging.getLogger(__name__) class ConnectionManager: 管理WebSocket连接简化版单机可用 def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) async def send_personal_message(self, message: str, websocket: WebSocket): await websocket.send_text(message) manager ConnectionManager() async def websocket_endpoint(websocket: WebSocket, db: AsyncSession Depends(get_db)): await manager.connect(websocket) llm_service LLMService() # 为当前会话创建一个简单的会话记录实际应关联用户 conversation Conversation(user_idfanon_{uuid.uuid4().hex[:8]}, title新咨询) db.add(conversation) await db.commit() await db.refresh(conversation) logger.info(fNew WebSocket connection established for conversation {conversation.id}) try: while True: # 1. 接收用户消息 data await websocket.receive_text() user_message json.loads(data) query_text user_message.get(text, ).strip() if not query_text: await manager.send_personal_message(json.dumps({error: 消息内容为空}), websocket) continue # 保存用户消息 user_msg_obj Message(conversation_idconversation.id, roleuser, contentquery_text) db.add(user_msg_obj) # 2. 检索增强获取相关法律条文 await manager.send_personal_message(json.dumps({status: retrieving, message: 正在检索相关法律条文...}), websocket) relevant_articles await RetrievalService.retrieve_relevant_articles(query_text, db) contexts [f{art.source}: {art.content} for art in relevant_articles] if not contexts: await manager.send_personal_message(json.dumps({status: warning, message: 未找到直接相关的法律条文将尝试基于通用知识回答。}), websocket) contexts [未检索到特定法律条文。] # 3. 调用LLM生成答案 await manager.send_personal_message(json.dumps({status: generating, message: 正在生成回答...}), websocket) answer await llm_service.generate_answer(query_text, contexts) # 4. 发送答案并保存 await manager.send_personal_message(json.dumps({status: completed, message: answer}), websocket) assistant_msg_obj Message(conversation_idconversation.id, roleassistant, contentanswer) db.add(assistant_msg_obj) await db.commit() logger.info(fConversation {conversation.id}: Q: {query_text[:30]}... A: {answer[:30]}...) except WebSocketDisconnect: manager.disconnect(websocket) logger.info(fWebSocket disconnected for conversation {conversation.id}) except Exception as e: logger.exception(fWebSocket error: {e}) await manager.send_personal_message(json.dumps({error: 服务器内部错误}), websocket) manager.disconnect(websocket)4.5 创建FastAPI主应用并注册路由创建app/main.pyfrom fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api import websocket from app.api import endpoints # 假设有RESTful API import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(titleAI法律援助咨询平台API, version1.0.0) # 配置CORS前端跨域 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应指定具体前端地址 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 注册WebSocket路由 app.websocket(/ws)(websocket.websocket_endpoint) # 注册RESTful API路由 # app.include_router(endpoints.router, prefix/api/v1) app.get(/) async def root(): return {message: AI法律援助咨询平台后端服务已启动, docs: /docs} app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: import uvicorn uvicorn.run(app.main:app, host0.0.0.0, port8000, reloadTrue)4.6 运行与测试启动后端服务cd backend uvicorn app.main:app --reload --host 0.0.0.0 --port 8000服务将在http://localhost:8000启动。访问http://localhost:8000/docs可以看到自动生成的API文档。测试WebSocket连接 可以使用websocat命令行工具或在线WebSocket测试客户端如https://www.piesocket.com/websocket-tester。连接地址ws://localhost:8000/ws发送消息{text: 借款合同纠纷的诉讼时效是多久}观察日志你将在后端控制台看到完整的检索和生成日志。5. 常见问题与排查思路在开发和部署过程中你可能会遇到以下典型问题。问题现象可能原因排查步骤与解决方案启动失败数据库连接错误1. Docker容器未运行。2. 数据库URL配置错误密码、端口。3. PostgreSQL未启用pgvector扩展。1.docker-compose ps检查容器状态docker-compose logs postgres查看日志。2. 检查DATABASE_URL环境变量或.env文件。3. 进入数据库执行CREATE EXTENSION vector;。向量检索结果不相关1. 嵌入模型与向量维度不匹配。2. 法律知识库数据质量差或未向量化。3. 检索的top_k值太小或太大。4. 未对向量列创建索引导致全表扫描精度差。1. 确认EMBEDDING_MODEL输出维度与表定义的vector(维度)一致。2. 检查law_articles表是否有数据embedding字段是否为非NULL。3. 调整RETRIEVAL_TOP_K参数如从3调到5。4. 为embedding列创建合适的索引如HNSW。CREATE INDEX ON law_articles USING hnsw (embedding vector_cosine_ops);WebSocket连接立即断开1. 前端连接地址/协议错误wsvswss。2. 后端CORS配置未覆盖WebSocket。3. 后端WebSocket端点路径错误。1. 前端确保使用ws://开发或wss://生产。2. CORS主要针对HTTPWebSocket协议不同。检查防火墙或反向代理如Nginx的WebSocket代理配置。3. 确认后端app.websocket(“/ws”)路径与前端的连接路径匹配。LLM回答出现幻觉或未引用法条1. Prompt指令不够严格。2. 检索到的上下文contexts未有效传递给LLM。3. LLM的temperature参数过高。1. 强化system_prompt明确要求“严格根据提供的条文回答”。2. 在日志中打印contexts确认其内容正确且已拼接到user_prompt中。3. 将temperature调低如0.1使输出更确定性。插入大量法律条文时速度慢1. 每条记录单独提交事务。2. 向量化过程是同步的阻塞主线程。1. 使用批量插入session.add_all()并定期提交。2. 将向量化任务放入后台队列如Celery或使用异步批处理。错误pgvector扩展未安装Docker镜像可能不包含pgvector或扩展未在目标数据库创建。使用明确包含pgvector的镜像如ankane/pgvector。进入容器执行psql -U law_admin -d law_consultation -c “CREATE EXTENSION IF NOT EXISTS vector;”。前端收不到流式响应后端未实现流式输出或前端未正确处理StreamingResponse或分块消息。1. 后端使用OpenAI的流式响应并将streamTrue的chunk通过WebSocket逐个发送。2. 前端监听WebSocket的onmessage事件并拼接消息。6. 最佳实践与工程建议将原型系统投入生产环境需要考虑更多工程化因素。6.1 知识库构建与管理数据质量法律文本必须来源权威如政府官网并进行严格的清洗、去重和格式化。分块策略法律条文长短不一。过长的文本如整部法律嵌入效果差过短则信息不足。建议按“条”或“款”进行分块并保留章节上下文信息如“《民法典》第六编 第三章 第667条”。元数据丰富除了title,content,source可添加law_type民法、刑法、effected_date、revision_info等字段便于做混合检索向量过滤。增量更新建立知识库版本管理和增量更新管道。当法律修订时能快速更新相关条文的向量。6.2 检索优化索引选择pgvector支持ivfflat和hnsw索引。对于千万级以下的数据hnsw在性能和召回率上通常更优。创建命令CREATE INDEX ON law_articles USING hnsw (embedding vector_cosine_ops);。混合检索结合关键词BM25和向量检索语义取长补短。例如先用关键词筛出一部分候选集再用向量检索做精排。重排序初次向量检索返回Top K如20条结果后可以使用更精细的交叉编码器模型进行重排序选出最精准的Top N如3条提升最终效果。6.3 LLM提示词与安全防御性提示在system_prompt中明确限制LLM的行为例如“你只能回答中国大陆法律相关问题”、“不得提供具体个案的法律行动建议”、“所有回答必须以‘仅供参考不构成法律意见’结尾”。上下文管理WebSocket对话需维护历史上下文。但法律咨询中过往对话可能干扰当前问题。建议策略是将当前问题检索结果作为主要上下文选择性附带最近1-2轮历史。审核与日志所有用户问题和AI回答必须落盘存储并考虑引入人工审核流程用于发现模型偏差和迭代优化。6.4 系统性能与可扩展性异步化如上所述使用FastAPI和asyncpg充分利用异步IO防止数据库或LLM API调用阻塞。服务解耦将向量检索、LLM调用等重CPU/IO操作拆分为独立微服务通过消息队列如RabbitMQ, Redis Stream通信提高整体吞吐和弹性。缓存策略对常见问题如“诉讼时效”的检索结果或最终答案进行缓存Redis可大幅降低响应延迟和LLM API成本。限流与熔断对WebSocket连接和LLM API调用实施限流防止滥用。为外部API如OpenAI配置熔断器防止因其不稳定导致服务雪崩。6.5 前端用户体验连接稳定性实现WebSocket自动重连机制并在UI上给予连接状态提示。流式输出实现LLM回答的流式输出打字机效果提升体验。这需要后端支持流式生成并分块发送。引用展示在UI中将AI回答中引用的法律条文高亮或折叠展示点击可查看原文增强可信度。对话历史提供对话历史列表和继续上次对话的功能。6.6 部署与监控容器化使用Docker Compose或Kubernetes编排所有服务后端、数据库、向量化Worker等。配置分离所有敏感信息API密钥、数据库密码必须通过环境变量或配置中心管理绝不入代码库。健康检查为每个服务提供/health端点并配置就绪性和存活探针。全面监控接入APM工具如SkyWalking, Sentry监控应用性能、错误日志监控数据库连接数、向量索引性能监控LLM API的调用延迟、费用和错误率。通过以上步骤你不仅能够搭建一个可运行的AI法律咨询平台原型更能理解其背后的核心原理与工程化考量。从检索增强的架构设计到向量数据库的实操再到实时通讯的集成这套技术栈的组合拳能够广泛应用于知识密集型、高准确性要求的智能问答场景如医疗咨询、金融合规、教育辅导等。
1. 项目背景与核心概念在传统的法律援助服务中用户常常面临咨询渠道不畅、专业律师资源有限、等待时间长以及信息检索效率低下等问题。随着人工智能技术的飞速发展特别是大语言模型LLM的涌现为构建智能化、即时化的在线法律服务平台提供了全新的可能。然而LLM本身存在“幻觉”问题可能生成不准确或虚构的法律条文这对于严谨的法律咨询场景是致命的。因此一个理想的解决方案是将LLM的通用对话能力与精准的法律知识库相结合。本项目“基于AILLM的法律援助在线咨询平台”正是为此而生。它通过整合PostgreSQL 向量数据库、LLM大模型和WebSocket即时通讯三大核心技术构建了一个能够提供专业、实时、有据可查的智能法律咨询系统。其核心工作流程是当用户提出法律问题时系统首先利用向量检索技术从结构化的法律知识库中精准找到相关法条和案例然后将这些“证据”与用户问题一同提交给LLM由LLM生成基于事实、引用准确的回答并通过WebSocket实时推送给用户。接下来我们逐一拆解这几个核心概念LLM大语言模型如 GPT、ChatGLM、通义千问等它们拥有强大的自然语言理解和生成能力是本平台的“大脑”负责理解用户意图并组织语言生成最终回复。检索增强生成RAG这是本项目的核心架构思想。RAG通过“先检索后生成”的方式有效解决了LLM的幻觉和知识更新滞后问题。具体到本项目就是先从法律知识库中检索出最相关的法律条文、司法解释或判例再将它们作为上下文提供给LLM约束其生成内容确保回答的专业性和准确性。PostgreSQL 与 pgvectorPostgreSQL 是一个功能强大的开源关系型数据库。pgvector是其一个扩展插件使其具备了存储和高效检索向量数据的能力。我们将法律文本通过嵌入模型转换为高维向量即语义编码存入 PostgreSQL从而构建起本项目的“法律知识大脑”。当用户提问时将问题同样转换为向量并在数据库中进行相似度搜索快速找到语义上最相关的法律知识。WebSocket一种在单个TCP连接上进行全双工通信的协议。相比于传统的HTTP轮询WebSocket能实现服务器主动向客户端推送消息非常适合在线聊天、实时通知等场景。在本平台中它用于建立用户与AI助手之间稳定、低延迟的对话通道实现流畅的即时咨询体验。简单来说这个平台就像一个拥有海量法律典籍且过目不忘的“AI律师助理”。它不仅能和你实时对话WebSocket还能随时翻查最权威的法律条文PostgreSQL 向量检索并用自己的话LLM清晰、准确地为你解答。2. 技术栈选型与环境准备为了成功复现本项目你需要准备好以下开发环境与技术组件。版本号以当前稳定版为例你可以根据实际情况调整。2.1 核心环境与工具操作系统Ubuntu 20.04 LTS / Windows 10 / macOS (建议在Linux环境下部署生产环境)Python: 3.9 或 3.10 (这是多数AI框架兼容性较好的版本)Java(可选)如果后端使用Spring Boot需 JDK 11 或 17Node.js(可选)如果前端使用Vue/React需 Node.js 16Docker Docker Compose(强烈推荐)用于快速部署数据库等中间件保证环境一致性。2.2 后端技术栈我们将以一个Python FastAPI后端为例进行讲解它轻量、异步非常适合AI应用和WebSocket。Web框架FastAPI(用于构建RESTful API和WebSocket端点)数据库与向量检索PostgreSQL15主数据库存储用户信息、对话记录等。pgvector扩展为PostgreSQL添加向量能力。ORM/数据库工具SQLAlchemyPython SQL工具包和ORM。asyncpg或psycopg2PostgreSQL适配器。alembic数据库迁移工具。大模型接入openai库 (如果使用OpenAI API) 或zhipuai(智谱AI) 等。也可使用本地模型如通过transformers库加载ChatGLM3、Qwen等。文本嵌入模型sentence-transformers用于将法律文本和用户问题转换为向量。推荐模型all-MiniLM-L6-v2(轻量英文) 或paraphrase-multilingual-MiniLM-L12-v2(多语言)。其他工具库langchain(可选)用于简化RAG流程的编排但为了理解原理本文会从底层实现开始。uvicornASGI服务器用于运行FastAPI应用。python-dotenv管理环境变量。2.3 前端技术栈 (示例)框架Vue 3 TypeScriptUI库Element Plus 或 Ant Design VueWebSocket客户端原生WebSocketAPI 或vue-use-websocketHTTP客户端axios2.4 项目初始化与环境配置首先创建项目目录并设置Python虚拟环境。# 创建项目目录 mkdir ai-law-consultation-platform cd ai-law-consultation-platform # 创建后端目录 mkdir backend cd backend # 创建Python虚拟环境 (Linux/macOS) python3 -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 创建 requirements.txt 并安装核心依赖 cat requirements.txt EOF fastapi0.104.1 uvicorn[standard]0.24.0 sqlalchemy2.0.23 asyncpg0.29.0 alembic1.12.1 python-dotenv1.0.0 sentence-transformers2.2.2 openai1.3.0 # 如果使用OpenAI API # transformers4.36.0 # 如果使用本地模型 # torch # 根据transformers需求安装 EOF pip install -r requirements.txt2.5 使用 Docker 启动 PostgreSQL pgvector这是最关键的一步。我们使用Docker Compose来定义和运行服务。在项目根目录 (ai-law-consultation-platform/) 创建docker-compose.yml文件version: 3.8 services: postgres: image: ankane/pgvector:latest # 这个镜像已包含pgvector扩展 container_name: law_pg_vector environment: POSTGRES_USER: law_admin POSTGRES_PASSWORD: your_secure_password_here POSTGRES_DB: law_consultation ports: - 5432:5432 volumes: - postgres_data:/var/lib/postgresql/data - ./init.sql:/docker-entrypoint-initdb.d/init.sql # 初始化脚本 restart: unless-stopped volumes: postgres_data:同时创建数据库初始化脚本init.sql用于创建我们需要的扩展和表结构初版-- 启用 pgvector 扩展 CREATE EXTENSION IF NOT EXISTS vector; -- 创建一个用于测试的法律知识表后续会丰富 CREATE TABLE IF NOT EXISTS law_articles ( id BIGSERIAL PRIMARY KEY, title TEXT NOT NULL, -- 法律条文标题 content TEXT NOT NULL, -- 法律条文内容 source TEXT, -- 来源如《民法典》第xxx条 embedding vector(384) -- 向量字段维度需与嵌入模型匹配例如 all-MiniLM-L6-v2 是384维 ); -- 为向量字段创建索引以加速检索使用IVFFlat或HNSW这里用IVFFlat示例 CREATE INDEX ON law_articles USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);重要说明vector(384)中的维度384必须与你选用的sentence-transformers模型输出维度一致。all-MiniLM-L6-v2模型输出维度就是384。启动数据库服务# 在项目根目录执行 docker-compose up -d执行后一个包含了pgvector扩展的PostgreSQL数据库就在本地的5432端口运行起来了。3. 核心原理与模块拆解在开始编码前我们需要深入理解系统是如何协同工作的。整个平台可以划分为四个核心模块其交互流程如下图所示概念图用户提问 | v [前端] --(WebSocket)-- [后端API/WebSocket服务] | | | v | [问题向量化] | | | v | [向量检索 (PostgreSQL pgvector)] | | | v | [检索结果 (相关法条)] | | | v | [构造Prompt 调用LLM] | | | v | [生成最终答案] | | | v [前端] --(WebSocket)-- [流式/非流式返回答案]3.1 知识库构建与向量化模块这是RAG的“知识底座”。法律条文、案例等非结构化文本需要被转换成计算机能理解的“语义向量”并存储起来。文本预处理清洗法律文本去除无关字符进行分段例如按法条拆分。向量化嵌入使用sentence-transformers模型将每一段文本转换为一个固定维度的浮点数向量。这个向量在数学空间中的“位置”代表了文本的语义。向量存储将(文本内容, 向量)对存入PostgreSQL的law_articles表。pgvector扩展允许我们直接存储vector类型的数据。为什么是向量检索传统的数据库关键词检索如LIKE或全文索引依赖于词汇匹配无法理解“借款合同”和“借贷协议”的语义相似性。向量检索通过比较向量之间的“距离”如余弦相似度能找出语义上最相近的内容即使它们没有相同的字词。3.2 检索增强生成RAG流程模块这是本项目的“智能引擎”。当用户提问时查询向量化将用户问题Q通过相同的嵌入模型转换为向量V_q。语义检索在PostgreSQL中执行近似最近邻搜索找出与V_q余弦相似度最高的前k条法律条文。SQL语句类似于SELECT id, title, content, source, 1 - (embedding ‘[V_q]‘) AS similarity FROM law_articles ORDER BY embedding ‘[V_q]‘ LIMIT 5;是pgvector提供的余弦距离运算符。提示词工程将检索到的法律条文{doc1, doc2, ...}和用户问题Q组合成一个结构化的提示词Prompt提交给LLM。你是一个专业的法律AI助手请严格根据以下提供的法律条文来回答问题。 如果提供的条文不足以回答问题请明确告知“根据现有资料无法回答”。 【相关法律条文】 1. 《民法典》第xxx条... 2. 《合同法》第yyy条... 【用户问题】 Q 【请回答】LLM生成LLM基于这个充满“证据”的上下文生成最终答案。这极大地减少了幻觉提高了答案的准确性和可信度。3.3 实时通讯模块为了提供类聊天的体验我们采用WebSocket。连接建立前端通过ws://your-domain/ws/{session_id}与后端建立持久化连接。对话管理后端需要维护会话状态可能包括关联用户ID、保存对话历史上下文等。消息路由前端发送一个包含问题的JSON消息。后端接收后触发RAG流程并将生成的结果通过同一条WebSocket连接推回前端。流式输出高级为了体验更佳可以让LLM以流式stream方式生成文本后端每生成一段就立刻推送给前端实现打字机效果。3.4 数据持久化模块使用SQLAlchemy ORM来管理用户信息注册、登录本文不展开认证细节。对话会话每个咨询会话的元数据。消息记录用户与AI的每一轮问答关联到会话。这对于后续分析、模型优化和上下文管理至关重要。法律知识库即前面提到的law_articles表。4. 完整实战从零搭建核心后端服务现在我们将一步步实现上述核心模块。我们将创建一个简单的后端包含知识库入库、RAG检索和WebSocket对话接口。4.1 项目结构backend/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── database.py # 数据库连接与ORM定义 │ ├── models.py # SQLAlchemy数据模型 │ ├── schemas.py # Pydantic数据验证模型 │ ├── services/ │ │ ├── __init__.py │ │ ├── embedding_service.py # 文本向量化服务 │ │ ├── retrieval_service.py # 向量检索服务 │ │ └── llm_service.py # LLM调用服务 │ ├── api/ │ │ ├── __init__.py │ │ ├── endpoints.py # RESTful API (如知识库管理) │ │ └── websocket.py # WebSocket处理逻辑 │ └── core/ │ └── config.py # 配置管理 ├── alembic/ # 数据库迁移目录后续生成 ├── requirements.txt ├── .env.example └── docker-compose.yml # 位于项目根目录4.2 配置与数据库模型首先设置配置。创建app/core/config.pyfrom pydantic_settings import BaseSettings class Settings(BaseSettings): # 数据库配置 DATABASE_URL: str postgresqlasyncpg://law_admin:your_secure_password_herelocalhost:5432/law_consultation # 嵌入模型配置 EMBEDDING_MODEL: str sentence-transformers/all-MiniLM-L6-v2 EMBEDDING_DEVICE: str cpu # 或 cuda # LLM配置 (以OpenAI为例) OPENAI_API_KEY: str OPENAI_BASE_URL: str https://api.openai.com/v1 # 或代理地址 OPENAI_MODEL: str gpt-3.5-turbo # 检索配置 RETRIEVAL_TOP_K: int 3 # 每次检索返回的最相关条文数量 class Config: env_file .env settings Settings()创建.env文件参考.env.example并填入你的实际配置尤其是数据库密码和API密钥。定义数据模型app/models.pyfrom sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql import func from pgvector.sqlalchemy import Vector Base declarative_base() class LawArticle(Base): __tablename__ law_articles id Column(Integer, primary_keyTrue, indexTrue) title Column(String(500), nullableFalse) content Column(Text, nullableFalse) source Column(String(300)) embedding Column(Vector(384)) # 维度与模型匹配 created_at Column(DateTime(timezoneTrue), server_defaultfunc.now()) class Conversation(Base): __tablename__ conversations id Column(Integer, primary_keyTrue, indexTrue) user_id Column(String(100), indexTrue) # 简化处理实际应关联用户表 title Column(String(255)) # 会话标题可由首条问题生成 created_at Column(DateTime(timezoneTrue), server_defaultfunc.now()) class Message(Base): __tablename__ messages id Column(Integer, primary_keyTrue, indexTrue) conversation_id Column(Integer, ForeignKey(conversations.id, ondeleteCASCADE), nullableFalse) role Column(String(20), nullableFalse) # user or assistant content Column(Text, nullableFalse) created_at Column(DateTime(timezoneTrue), server_defaultfunc.now())创建数据库连接会话app/database.pyfrom sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from app.core.config import settings # 创建异步引擎 engine create_async_engine( settings.DATABASE_URL, echoTrue, # 开发时显示SQL日志生产环境应关闭 futureTrue ) # 创建异步会话工厂 AsyncSessionLocal sessionmaker( engine, class_AsyncSession, expire_on_commitFalse ) # 依赖注入用的获取会话函数 async def get_db(): async with AsyncSessionLocal() as session: try: yield session finally: await session.close()4.3 实现核心服务层1. 嵌入服务app/services/embedding_service.pyfrom sentence_transformers import SentenceTransformer import numpy as np from app.core.config import settings import logging logger logging.getLogger(__name__) class EmbeddingService: _model None classmethod async def get_model(cls): 懒加载嵌入模型 if cls._model is None: logger.info(fLoading embedding model: {settings.EMBEDDING_MODEL}) cls._model SentenceTransformer(settings.EMBEDDING_MODEL, devicesettings.EMBEDDING_DEVICE) return cls._model classmethod async def embed_texts(cls, texts: list[str]) - list[list[float]]: 将文本列表转换为向量列表 model await cls.get_model() # 模型.encode是同步的在异步环境中使用run_in_executor避免阻塞事件循环 import asyncio from functools import partial loop asyncio.get_event_loop() embeddings await loop.run_in_executor(None, partial(model.encode, texts, convert_to_numpyTrue, normalize_embeddingsTrue)) return embeddings.tolist() classmethod async def embed_single(cls, text: str) - list[float]: 将单条文本转换为向量 embeddings await cls.embed_texts([text]) return embeddings[0]2. 检索服务app/services/retrieval_service.pyfrom sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from pgvector.sqlalchemy import cosine_distance from app.models import LawArticle from app.services.embedding_service import EmbeddingService from app.core.config import settings import logging logger logging.getLogger(__name__) class RetrievalService: staticmethod async def retrieve_relevant_articles(query: str, db: AsyncSession, top_k: int None) - list[LawArticle]: 根据用户问题检索最相关的法律条文。 返回LawArticle 对象列表 if top_k is None: top_k settings.RETRIEVAL_TOP_K # 1. 将查询文本向量化 query_embedding await EmbeddingService.embed_single(query) # 2. 构建向量检索SQL查询 (使用余弦距离距离越小越相似) stmt ( select(LawArticle) .order_by(LawArticle.embedding.cosine_distance(query_embedding)) .limit(top_k) ) # 3. 执行查询 result await db.execute(stmt) articles result.scalars().all() logger.info(fRetrieved {len(articles)} articles for query: {query[:50]}...) return articles3. LLM服务app/services/llm_service.py这里以OpenAI API为例。如果你使用本地模型需要调整调用方式。from openai import AsyncOpenAI from app.core.config import settings import logging logger logging.getLogger(__name__) class LLMService: def __init__(self): self.client AsyncOpenAI( api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_BASE_URL, ) async def generate_answer(self, query: str, contexts: list[str]) - str: 基于用户问题和检索到的上下文调用LLM生成答案。 Args: query: 用户问题 contexts: 检索到的法律条文内容列表 Returns: LLM生成的答案文本 if not settings.OPENAI_API_KEY: return LLM服务未配置。请检查API密钥设置。 # 构造Prompt context_str \n\n.join([f{i1}. {ctx} for i, ctx in enumerate(contexts)]) system_prompt 你是一个专业的中国法律AI助手必须严格根据用户提供的法律条文来回答问题。 你的回答应当准确、清晰、有条理并引用相关条文。 如果提供的条文不足以回答用户的问题请诚实地告知“根据提供的资料我无法给出确切答案”并建议用户咨询专业律师。 禁止编造法律条文或案例。 user_prompt f【用户问题】 {query} 【相关法律条文】 {context_str} 请根据以上条文回答用户的问题。 try: response await self.client.chat.completions.create( modelsettings.OPENAI_MODEL, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt} ], temperature0.1, # 低温度使输出更确定、更少创造性 max_tokens1000, ) answer response.choices[0].message.content return answer.strip() except Exception as e: logger.error(f调用LLM API失败: {e}) return f生成回答时出现错误{str(e)}4.4 实现WebSocket端点与RAG流程整合创建app/api/websocket.py这是整个实时咨询功能的核心。from fastapi import WebSocket, WebSocketDisconnect, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.services.retrieval_service import RetrievalService from app.services.llm_service import LLMService from app.models import Conversation, Message import json import logging import uuid logger logging.getLogger(__name__) class ConnectionManager: 管理WebSocket连接简化版单机可用 def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) async def send_personal_message(self, message: str, websocket: WebSocket): await websocket.send_text(message) manager ConnectionManager() async def websocket_endpoint(websocket: WebSocket, db: AsyncSession Depends(get_db)): await manager.connect(websocket) llm_service LLMService() # 为当前会话创建一个简单的会话记录实际应关联用户 conversation Conversation(user_idfanon_{uuid.uuid4().hex[:8]}, title新咨询) db.add(conversation) await db.commit() await db.refresh(conversation) logger.info(fNew WebSocket connection established for conversation {conversation.id}) try: while True: # 1. 接收用户消息 data await websocket.receive_text() user_message json.loads(data) query_text user_message.get(text, ).strip() if not query_text: await manager.send_personal_message(json.dumps({error: 消息内容为空}), websocket) continue # 保存用户消息 user_msg_obj Message(conversation_idconversation.id, roleuser, contentquery_text) db.add(user_msg_obj) # 2. 检索增强获取相关法律条文 await manager.send_personal_message(json.dumps({status: retrieving, message: 正在检索相关法律条文...}), websocket) relevant_articles await RetrievalService.retrieve_relevant_articles(query_text, db) contexts [f{art.source}: {art.content} for art in relevant_articles] if not contexts: await manager.send_personal_message(json.dumps({status: warning, message: 未找到直接相关的法律条文将尝试基于通用知识回答。}), websocket) contexts [未检索到特定法律条文。] # 3. 调用LLM生成答案 await manager.send_personal_message(json.dumps({status: generating, message: 正在生成回答...}), websocket) answer await llm_service.generate_answer(query_text, contexts) # 4. 发送答案并保存 await manager.send_personal_message(json.dumps({status: completed, message: answer}), websocket) assistant_msg_obj Message(conversation_idconversation.id, roleassistant, contentanswer) db.add(assistant_msg_obj) await db.commit() logger.info(fConversation {conversation.id}: Q: {query_text[:30]}... A: {answer[:30]}...) except WebSocketDisconnect: manager.disconnect(websocket) logger.info(fWebSocket disconnected for conversation {conversation.id}) except Exception as e: logger.exception(fWebSocket error: {e}) await manager.send_personal_message(json.dumps({error: 服务器内部错误}), websocket) manager.disconnect(websocket)4.5 创建FastAPI主应用并注册路由创建app/main.pyfrom fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api import websocket from app.api import endpoints # 假设有RESTful API import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(titleAI法律援助咨询平台API, version1.0.0) # 配置CORS前端跨域 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应指定具体前端地址 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 注册WebSocket路由 app.websocket(/ws)(websocket.websocket_endpoint) # 注册RESTful API路由 # app.include_router(endpoints.router, prefix/api/v1) app.get(/) async def root(): return {message: AI法律援助咨询平台后端服务已启动, docs: /docs} app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: import uvicorn uvicorn.run(app.main:app, host0.0.0.0, port8000, reloadTrue)4.6 运行与测试启动后端服务cd backend uvicorn app.main:app --reload --host 0.0.0.0 --port 8000服务将在http://localhost:8000启动。访问http://localhost:8000/docs可以看到自动生成的API文档。测试WebSocket连接 可以使用websocat命令行工具或在线WebSocket测试客户端如https://www.piesocket.com/websocket-tester。连接地址ws://localhost:8000/ws发送消息{text: 借款合同纠纷的诉讼时效是多久}观察日志你将在后端控制台看到完整的检索和生成日志。5. 常见问题与排查思路在开发和部署过程中你可能会遇到以下典型问题。问题现象可能原因排查步骤与解决方案启动失败数据库连接错误1. Docker容器未运行。2. 数据库URL配置错误密码、端口。3. PostgreSQL未启用pgvector扩展。1.docker-compose ps检查容器状态docker-compose logs postgres查看日志。2. 检查DATABASE_URL环境变量或.env文件。3. 进入数据库执行CREATE EXTENSION vector;。向量检索结果不相关1. 嵌入模型与向量维度不匹配。2. 法律知识库数据质量差或未向量化。3. 检索的top_k值太小或太大。4. 未对向量列创建索引导致全表扫描精度差。1. 确认EMBEDDING_MODEL输出维度与表定义的vector(维度)一致。2. 检查law_articles表是否有数据embedding字段是否为非NULL。3. 调整RETRIEVAL_TOP_K参数如从3调到5。4. 为embedding列创建合适的索引如HNSW。CREATE INDEX ON law_articles USING hnsw (embedding vector_cosine_ops);WebSocket连接立即断开1. 前端连接地址/协议错误wsvswss。2. 后端CORS配置未覆盖WebSocket。3. 后端WebSocket端点路径错误。1. 前端确保使用ws://开发或wss://生产。2. CORS主要针对HTTPWebSocket协议不同。检查防火墙或反向代理如Nginx的WebSocket代理配置。3. 确认后端app.websocket(“/ws”)路径与前端的连接路径匹配。LLM回答出现幻觉或未引用法条1. Prompt指令不够严格。2. 检索到的上下文contexts未有效传递给LLM。3. LLM的temperature参数过高。1. 强化system_prompt明确要求“严格根据提供的条文回答”。2. 在日志中打印contexts确认其内容正确且已拼接到user_prompt中。3. 将temperature调低如0.1使输出更确定性。插入大量法律条文时速度慢1. 每条记录单独提交事务。2. 向量化过程是同步的阻塞主线程。1. 使用批量插入session.add_all()并定期提交。2. 将向量化任务放入后台队列如Celery或使用异步批处理。错误pgvector扩展未安装Docker镜像可能不包含pgvector或扩展未在目标数据库创建。使用明确包含pgvector的镜像如ankane/pgvector。进入容器执行psql -U law_admin -d law_consultation -c “CREATE EXTENSION IF NOT EXISTS vector;”。前端收不到流式响应后端未实现流式输出或前端未正确处理StreamingResponse或分块消息。1. 后端使用OpenAI的流式响应并将streamTrue的chunk通过WebSocket逐个发送。2. 前端监听WebSocket的onmessage事件并拼接消息。6. 最佳实践与工程建议将原型系统投入生产环境需要考虑更多工程化因素。6.1 知识库构建与管理数据质量法律文本必须来源权威如政府官网并进行严格的清洗、去重和格式化。分块策略法律条文长短不一。过长的文本如整部法律嵌入效果差过短则信息不足。建议按“条”或“款”进行分块并保留章节上下文信息如“《民法典》第六编 第三章 第667条”。元数据丰富除了title,content,source可添加law_type民法、刑法、effected_date、revision_info等字段便于做混合检索向量过滤。增量更新建立知识库版本管理和增量更新管道。当法律修订时能快速更新相关条文的向量。6.2 检索优化索引选择pgvector支持ivfflat和hnsw索引。对于千万级以下的数据hnsw在性能和召回率上通常更优。创建命令CREATE INDEX ON law_articles USING hnsw (embedding vector_cosine_ops);。混合检索结合关键词BM25和向量检索语义取长补短。例如先用关键词筛出一部分候选集再用向量检索做精排。重排序初次向量检索返回Top K如20条结果后可以使用更精细的交叉编码器模型进行重排序选出最精准的Top N如3条提升最终效果。6.3 LLM提示词与安全防御性提示在system_prompt中明确限制LLM的行为例如“你只能回答中国大陆法律相关问题”、“不得提供具体个案的法律行动建议”、“所有回答必须以‘仅供参考不构成法律意见’结尾”。上下文管理WebSocket对话需维护历史上下文。但法律咨询中过往对话可能干扰当前问题。建议策略是将当前问题检索结果作为主要上下文选择性附带最近1-2轮历史。审核与日志所有用户问题和AI回答必须落盘存储并考虑引入人工审核流程用于发现模型偏差和迭代优化。6.4 系统性能与可扩展性异步化如上所述使用FastAPI和asyncpg充分利用异步IO防止数据库或LLM API调用阻塞。服务解耦将向量检索、LLM调用等重CPU/IO操作拆分为独立微服务通过消息队列如RabbitMQ, Redis Stream通信提高整体吞吐和弹性。缓存策略对常见问题如“诉讼时效”的检索结果或最终答案进行缓存Redis可大幅降低响应延迟和LLM API成本。限流与熔断对WebSocket连接和LLM API调用实施限流防止滥用。为外部API如OpenAI配置熔断器防止因其不稳定导致服务雪崩。6.5 前端用户体验连接稳定性实现WebSocket自动重连机制并在UI上给予连接状态提示。流式输出实现LLM回答的流式输出打字机效果提升体验。这需要后端支持流式生成并分块发送。引用展示在UI中将AI回答中引用的法律条文高亮或折叠展示点击可查看原文增强可信度。对话历史提供对话历史列表和继续上次对话的功能。6.6 部署与监控容器化使用Docker Compose或Kubernetes编排所有服务后端、数据库、向量化Worker等。配置分离所有敏感信息API密钥、数据库密码必须通过环境变量或配置中心管理绝不入代码库。健康检查为每个服务提供/health端点并配置就绪性和存活探针。全面监控接入APM工具如SkyWalking, Sentry监控应用性能、错误日志监控数据库连接数、向量索引性能监控LLM API的调用延迟、费用和错误率。通过以上步骤你不仅能够搭建一个可运行的AI法律咨询平台原型更能理解其背后的核心原理与工程化考量。从检索增强的架构设计到向量数据库的实操再到实时通讯的集成这套技术栈的组合拳能够广泛应用于知识密集型、高准确性要求的智能问答场景如医疗咨询、金融合规、教育辅导等。
