三层架构设计前端业务AI层的职责划分前言在构建企业级AI Agent系统时清晰的架构分层至关重要。本文将详细介绍本项目采用的三层架构设计以及各层的职责划分和通信机制。适合读者架构师、全栈工程师、技术Leader一、为什么需要三层架构1.1 单体架构的问题传统单体架构 Frontend Backend AI 混在一起 ├── 代码耦合严重 ├── 难以独立扩展 ├── 技术栈受限 └── 团队协作困难1.2 三层架构的优势三层架构 Frontend ← → Server ← → Agent ├── 职责清晰 ├── 独立部署 ├── 技术栈自由 ├── 易于扩展 └── 团队并行开发二、架构全景图2.1 系统架构┌─────────────────────────────────────────────────────────┐ │ 用户层 │ │ (浏览器/移动端) │ └────────────────────┬────────────────────────────────────┘ │ HTTPS ▼ ┌─────────────────────────────────────────────────────────┐ │ Frontend Layer │ │ Next.js React TailwindCSS │ │ (Port: 3000) │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 聊天界面 │ │ 用户认证 │ │ 对话管理 │ │ │ └──────────┘ └──────────┘ └──────────┘ │ └────────────────────┬────────────────────────────────────┘ │ HTTP/SSE ▼ ┌─────────────────────────────────────────────────────────┐ │ Server Layer │ │ FastAPI PostgreSQL Redis │ │ (Port: 8000) │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 业务逻辑 │ │ 用户管理 │ │ 数据存储 │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ JWT认证 │ │ 缓存管理 │ │ 日志记录 │ │ │ └──────────┘ └──────────┘ └──────────┘ │ └────────────────────┬────────────────────────────────────┘ │ HTTP/SSE ▼ ┌─────────────────────────────────────────────────────────┐ │ Agent Layer │ │ LangChain Ollama Weaviate │ │ (Port: 8001) │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ RAG检索 │ │ Prompt组装│ │ 流式生成 │ │ │ └──────────┘ └──────────┘ └──────────┘ │ └────────────────────┬────────────────────────────────────┘ │ ┌────────────┼────────────┐ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌──────────┐ │Weaviate│ │ Ollama │ │PostgreSQL│ │ :8080 │ │:11434 │ │ :5432 │ └────────┘ └────────┘ └──────────┘三、Frontend Layer前端层3.1 核心职责主要职责用户界面- 提供友好的交互体验身份认证- Token管理和自动刷新实时通信- SSE流式接收AI回复状态管理- 对话历史和用户状态路由管理- 页面导航和权限控制3.2 技术栈// 技术选型Frontend Stack:├── Next.js13.5.6// React框架支持SSR├── React18.2.0// UI组件库├── TypeScript5.2.2// 类型安全├── TailwindCSS3.3.5// 原子化CSS├── Axios1.6.0// HTTP客户端└──microsoft/fetch-event-source2.0.1// SSE支持3.3 核心代码示例// frontend/app/chat/page.tsxuse clientimport{useState}fromreactimport{streamChat}from/lib/api-clientexportdefaultfunctionChatPage(){const[messages,setMessages]useStateMessage[]([])const[inputValue,setInputValue]useState()const[isStreaming,setIsStreaming]useState(false)const[currentAnswer,setCurrentAnswer]useState()consthandleSendasync(){if(!inputValue.trim()||isStreaming)return// 添加用户消息constuserMessage{id:Date.now().toString(),role:user,content:inputValue}setMessages(prev[...prev,userMessage])setInputValue()setIsStreaming(true)setCurrentAnswer()try{// 调用Server API接收流式响应awaitstreamChat(userMessage.content,{onToken:(token){setCurrentAnswer(prevprevtoken)},onDone:(data){constaiMessage{id:Date.now().toString(),role:assistant,content:data.answer}setMessages(prev[...prev,aiMessage])setCurrentAnswer()setIsStreaming(false)},onError:(error){console.error(错误:,error)setIsStreaming(false)}})}catch(error){console.error(发送失败:,error)setIsStreaming(false)}}return(div classNameflex flex-col h-screen{/* 消息列表 */}div classNameflex-1 overflow-y-auto p-4{messages.map((msg)(MessageBubble key{msg.id}message{msg}/))}{/* 实时流式消息 */}{currentAnswer(div classNamebg-gray-100 rounded-lg p-4{currentAnswer}span classNameanimate-pulse▊/span/div)}/div{/* 输入框 */}div classNamep-4 border-tinput value{inputValue}onChange{(e)setInputValue(e.target.value)}onKeyPress{(e){if(e.keyEnter!e.shiftKey){e.preventDefault()handleSend()}}}placeholder输入问题...disabled{isStreaming}/button onClick{handleSend}disabled{!inputValue.trim()||isStreaming}{isStreaming?生成中...:发送}/button/div/div)}四、Server Layer业务层4.1 核心职责主要职责认证授权- JWT双Token机制业务逻辑- 用户管理、对话管理️数据持久化- PostgreSQL存储⚡缓存管理- Redis加速服务编排- 协调Frontend和Agent日志监控- Loguru结构化日志4.2 技术栈# 技术选型Server Stack:├── FastAPI0.100.0# 现代化异步Web框架├── Uvicorn0.31.1# ASGI服务器├── SQLAlchemy2.0.18# 异步ORM├── PostgreSQL# 关系型数据库├── Redis4.6.0# 缓存和会话├── Alembic1.11.1# 数据库迁移├── Pydantic2.0.0# 数据验证├── Python-Jose3.3.0# JWT认证├── Passlib[bcrypt]1.7.4# 密码加密└── Loguru0.7.0# 结构化日志4.3 核心代码示例# server/api/agent.pyfromfastapiimportAPIRouter,Depends,HTTPExceptionfromfastapi.responsesimportStreamingResponseimporthttpximportjson routerAPIRouter(prefix/api/agent,tags[agent])AGENT_URLhttp://localhost:8001router.post(/chat/stream)asyncdefagent_stream(question:str,current_userDepends(get_current_user)): 代理Agent流式接口 Server作为中间层 1. 验证用户身份JWT 2. 记录请求日志 3. 转发到Agent服务 4. 保存对话历史 # 记录请求logger.info(f用户{current_user.username}发起问答:{question})asyncdefevent_generator():asyncwithhttpx.AsyncClient()asclient:asyncwithclient.stream(POST,f{AGENT_URL}/stream,json{question:question},timeout60.0)asresponse:asyncforlineinresponse.aiter_lines():ifline:yieldf{line}\nreturnStreamingResponse(event_generator(),media_typetext/event-stream)# server/api/auth.pyfromdatetimeimporttimedeltafromfastapiimportAPIRouter,Depends,HTTPExceptionfromsqlalchemy.ext.asyncioimportAsyncSession routerAPIRouter(prefix/api/auth,tags[auth])router.post(/login)asyncdeflogin(username:str,password:str,db:AsyncSessionDepends(get_db)):用户登录# 验证用户userawaitauthenticate_user(db,username,password)ifnotuser:raiseHTTPException(status_code401,detail用户名或密码错误)# 生成Tokenaccess_tokencreate_access_token(data{sub:user.username},expires_deltatimedelta(minutes30))refresh_tokencreate_refresh_token(data{sub:user.username},expires_deltatimedelta(days7))return{code:0,msg:登录成功,data:{access_token:access_token,refresh_token:refresh_token,token_type:bearer}}# server/api/conversations.pyrouter.post()asyncdefcreate_conversation(title:str,current_userDepends(get_current_user),db:AsyncSessionDepends(get_db)):创建新对话conversationConversation(user_idcurrent_user.id,titletitle)db.add(conversation)awaitdb.commit()awaitdb.refresh(conversation)return{code:0,msg:创建成功,data:conversation}五、Agent LayerAI层5.1 核心职责主要职责向量检索- Weaviate相似度搜索AI推理- Ollama LLM生成Prompt工程- 动态组装上下文流式生成- 实时返回Token结果格式化- 统一响应格式5.2 技术栈# 技术选型Agent Stack:├── LangChain0.1.0# AI应用开发框架├── LangChain-Community0.0.10# 社区集成├── Ollama0.1.0# 本地LLM服务├── Weaviate1.27.1# 向量数据库├── FastAPI0.104.0# HTTP服务框架└── Pandas2.0.0# 数据处理AI模型:├── llama3.2:latest# 对话生成模型└── nomic-embed-text# 文本向量化模型5.3 核心代码示例# agent/http_service.pyfromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponsefrompydanticimportBaseModelimportjsonfromagentimportServiceTicketAgent appFastAPI(titleService Ticket Agent)classChatRequest(BaseModel):question:stragent_instanceNonedefget_agent():globalagent_instanceifagent_instanceisNone:agent_instanceServiceTicketAgent()returnagent_instanceapp.post(/stream)asyncdefstream_chat(request:ChatRequest): 流式对话接口SSE 返回SSE格式的流式数据 - event: thinking (思考状态) - event: sources (检索来源) - event: token (逐个token) - event: done (完成) - event: error (错误) asyncdefevent_generator():agentget_agent()try:asyncforeventinagent.ask_stream(request.question):event_typeevent.get(type)event_dataevent.get(data,{})# 格式化为SSEyieldfevent:{event_type}\nyieldfdata:{json.dumps(event_data,ensure_asciiFalse)}\n\nifevent_typein[done,error]:breakexceptExceptionase:error_event{code:500,msg:fAgent错误:{str(e)}}yieldfevent: error\nyieldfdata:{json.dumps(error_event,ensure_asciiFalse)}\n\nreturnStreamingResponse(event_generator(),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,X-Accel-Buffering:no,})# agent/ticket_agent.pyfromlangchain_core.runnablesimportRunnablePassthroughfromlangchain_core.output_parsersimportStrOutputParserclassServiceTicketAgent:def_setup_qa_chain(self):设置问答链LCELdefretrieve_and_format(question):docsself._search_similar_documents(question)return\n\n.join(doc.page_contentfordocindocs)# LCEL链式调用self.qa_chain({context:retrieve_and_format,question:RunnablePassthrough()}|self.prompt_template|self.llm|StrOutputParser())returnself.qa_chainasyncdefask_stream(self,question:str):流式问答# 1. 检索相关工单yield{type:thinking,data:{status:retrieving,message:正在检索相关工单...}}source_docsself._search_similar_documents(question)# 2. 返回检索结果yield{type:sources,data:{sources:[...],count:len(source_docs)}}# 3. 流式生成答案qa_chainself._setup_qa_chain()full_answerasyncforchunkinqa_chain.astream(question):tokenstr(chunk)full_answertokenyield{type:token,data:{token:token}}# 4. 完成yield{type:done,data:{answer:full_answer,metadata:{...}}}六、层间通信协议6.1 Frontend ↔ Server// HTTP请求POST/api/auth/login Content-Type:application/json{username:user,password:password}// HTTP响应{code:0,msg:success,data:{access_token:eyJ...,refresh_token:eyJ...}}// SSE流式响应POST/api/agent/chat/stream Authorization:Bearer eyJ...event:token data:{token:根据}event:token data:{token:历史}event:done data:{answer:...,metadata:{...}}6.2 Server ↔ Agent# HTTP代理POST http://localhost:8001/stream Content-Type:application/json{question:物流信息5天没更新怎么处理}# SSE响应event:thinking data:{status:retrieving,message:正在检索相关工单...}event:sources data:{sources:[...],count:5}event:token data:{token:根据}event:done data:{answer:...,metadata:{...}}七、独立部署与扩展7.1 独立部署# docker-compose.ymlversion:3.8services:frontend:build:./frontendports:-3000:3000depends_on:-serverserver:build:./serverports:-8000:8000depends_on:-postgres-redis-agentagent:build:./agentports:-8001:8001depends_on:-ollama-weaviatepostgres:image:postgres:15ports:-5432:5432redis:image:redis:7ports:-6379:6379weaviate:image:semitechnologies/weaviate:1.27.1ports:-8080:8080ollama:image:ollama/ollama:latestports:-11434:114347.2 水平扩展# 扩展Agent层services:agent:build:./agentdeploy:replicas:3# 3个Agent实例ports:-8001-8003:8001# 负载均衡nginx:image:nginx:latestports:-80:80volumes:-./nginx.conf:/etc/nginx/nginx.conf八、团队协作8.1 职责分工Frontend团队: ├── UI/UX设计 ├── React组件开发 ├── 状态管理 └── 前端性能优化 Server团队: ├── 业务逻辑开发 ├── 数据库设计 ├── API接口设计 └── 认证授权 Agent团队: ├── AI模型选型 ├── Prompt工程 ├── RAG优化 └── 向量数据库管理8.2 并行开发阶段1: 接口定义1天 ├── Frontend定义需要的API ├── Server定义Agent接口 └── 三方达成一致 阶段2: 并行开发2周 ├── Frontend: Mock数据开发UI ├── Server: 实现业务逻辑 └── Agent: 实现AI功能 阶段3: 联调测试3天 ├── Frontend Server联调 ├── Server Agent联调 └── 端到端测试九、总结三层架构的核心优势✅职责清晰- 每层专注自己的领域✅独立部署- 可以单独扩展和升级✅技术自由- 每层选择最适合的技术栈✅团队协作- 多团队并行开发✅易于维护- 降低系统复杂度下一篇预告《技术选型背后的思考为什么选择Next.jsFastAPILangChain》我们将深入分析每个技术栈的选型理由和替代方案对比。作者简介资深开发者创业者。专注于视频通讯技术领域。国内首本Flutter著作《Flutter技术入门与实战》作者,另著有《Dart语言实战》及《WebRTC音视频开发》等书籍。多年从事视频会议、远程教育等技术研发对于Android、iOS以及跨平台开发技术有比较深入的研究和应用作为主要程序员开发了多个应用项目涉及医疗、交通、银行等领域。学习资料项目地址作者GitHub欢迎交流如有问题欢迎在评论区讨论