不做大哥好多年
xiaonaiqiang
2025-06-10
目录

300.整体设计

# 00.架构说明

AI Agent 的服务平台四层架构

  • 用户入口层（API / UI）
  • Agent 执行编排层（Core）
  • 能力支撑层（LLM / Tool / Memory / RAG）
  • 基础设施层（存储 / 调度 / 日志 / 权限）
层级 关键职责 技术核心
用户入口层 请求接入、身份鉴权、上下文绑定 FastAPI + Redis + WebSocket + JWT
Agent 编排层 Agent 任务流程、调用决策、执行控制 LangChain / LangGraph + asyncio + Prompt 工程
能力支撑层 实际执行 LLM、工具、记忆检索、RAG 补充 LLM API + 工具封装 + Redis Memory + 向量库
基础设施层 存储、调度、日志、权限、配置等平台支撑 PostgreSQL + MinIO + Loki + Celery + Consul

# 1、用户入口层（API / UI）

  • 示例请求场景：用户在 Chat UI 输入一句话：「帮我总结一下这篇文章的重点」，并上传一段 PDF 内容

  • 职责：提供统一入口，识别用户身份、会话状态，转发到 Agent 核心逻辑

  • 技术栈：

    • React + WebSocket 用于 UI 实时响应

    • FastAPI 接收消息，统一路由处理

    • Redis 存储会话状态、上下文信息

    • JWT 认证 + OAuth 鉴权

处理流程

  • 用户通过 Web 前端 UI（React + Tailwind）输入请求 + 上传文件

  • 浏览器通过 WebSocket 向后端发送 JSON 请求

    • {
  "session_id": "sess-xyz",
  "user_input": "帮我总结一下这篇文章的重点",
  "document": "xxx.pdf"
}
      1
      2
      3
      4
      5

  • FastAPI WebSocket 接收后

    • 解析 JWT token，提取用户身份
    • 确认 session_id 是否存在于 Redis，如果没有则初始化 session
    • 上传的文档先缓存（本地或 MinIO），并生成预处理 ID
    • 构造结构化调用参数转入下一层：agent_executor.invoke(input_dict)

  • 用户入口层的 API Handler 确实需要负责选择 Agent

    • 用户主动选择：比如 UI 上选择 客服助手、文档助手、数据分析 Agent
    • 系统智能分发：根据请求内容自动匹配合适的 Agent（类似 Intent Router）
      • 实现思路，调用一个小型 LLM，判断请求属于哪个 Agent

# 2、Agent 执行编排层

  • 职责：对接 LangChain/LangGraph，负责执行计划、调用 Memory、决定工具是否使用等

  • 技术栈：

    • LangChain AgentExecutor（ReAct / Function Agent）

    • LangGraph（用于多阶段 Agent 状态流转）

    • Pydantic 构造请求结构

    • asyncio 控制执行流程

# 1）websocket 设计

  • 收到上层传入的 input_dict（使用 http 请求）

    • 后端 → 立即生成唯一task_id并返回

    • {
  "input": "帮我总结一下这篇文章的重点",
  "file_id": "doc-123",
  "chat_history": [...],
  "user_id": "u-001"
}
      1
      2
      3
      4
      5
      6

    • 结果获取阶段（WebSocket）

      • 前端 → 使用task_id建立WebSocket连接
      • 后端 → 通过同一WS连接持续推送：排队状态 → 执行进度 → 最终结果/错误信息

# 2）动态创建 Agent

  • 判断是否需要 Tool

    • 显式判断：用户是否上传了 PDF/图片/表格？
    • 意图识别：通过 LLM 判断意图，如“总结、翻译、提取数据” → 绑定相应 Tool

  • 判断是否需要开启 Memory

    • 检查请求是否带有 session_id 和 user_id

    • 读取 Redis，恢复为 ConversationBufferMemory

    • Memory 会自动记录过去对话，并注入到 Prompt 中

    • 用户的历史对话存在 Redis 中，结构示意

    • Key: "chat_memory:{user_id}:{session_id}"
Value: JSON list [
  {"role": "human", "content": "你好"},
  {"role": "ai", "content": "你好，我能帮你什么？"},
  ...
]
      1
      2
      3
      4
      5
      6

  • 动态构建 AgentExecutor（整合 LLM + Tool + Memory）

    • 由于每次请求，需要的 tools 和 memory 等不同，所以每次可以动态创建 agent

    • 动态Agent创建的作用

      • 用户隔离：每个请求独立生成 AgentExecutor，保证
        • 工具权限隔离（用户A/B的工具列表不同）
        • 记忆隔离（会话历史独立）
      • 灵活配置：根据请求参数动态选择
        • LLM模型（如GPT-3.5→GPT-4）
        • 工具组合（如是否启用联网搜索）
    • from functools import lru_cache

# 缓存工具初始化（避免重复加载工具定义）
@lru_cache(maxsize=100)
def get_tools(user_id: str) -> List[Tool]:
    if check_permission(user_id, "pdf"):
        return [pdf_tool, search_tool]
    else:
        return [search_tool]

def handle_request(user_input, user_id, session_id):
    # 动态获取工具（带缓存优化）
    tools = get_tools(user_id)
    
    # 每次新建Agent（保证工具隔离）
    agent = initialize_agent(
        tools=tools,  # 动态注入
        llm=llm,
        memory=load_memory(session_id)
    )
    return agent.run(user_input)
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21

# 3）动态创建Agent + 静态LangGraph

  • 触发 LangGraph 执行（如配置为状态流程图）
    • 第一阶段：文档解析（Tool 触发）
    • 第二阶段：提取内容传入 LLM 总结
    • 第三阶段：回传输出

# 4）实时推送至前端

  • 执行期间，每个阶段的 token 响应通过 LangChain callback + FastAPI WebSocket 实时推送至前端

# 3、能力支撑层（LLM / Tool / Memory / RAG）

  • 职责：完成实际智能体能力调用，如语言模型、工具插件、记忆检索、RAG 文档增强等

  • 技术栈：

    • LLM：OpenAI / DeepSeek / AzureOpenAI（通过 LangChain 封装）

    • Tool：PDF Summarizer Tool / Search Tool（注册至 Agent 工具列表）

    • Memory：Redis-based ChatMessageHistory（按 user_id/session_id 分区）

    • RAG：FAISS / Qdrant / Weaviate 等向量库

处理流程

  • Tool 工具模块调用：

    • 如果 Agent 判断需要处理文档，则调用 pdf_summary_tool.run(file_id)
    • 工具内部调用 LangChain DocumentLoader + TextSplitter + LLM 总结链

  • 如果绑定了向量库（RAG）

    • 对用户请求向量化
    • 检索相关文章内容，加入上下文增强提示模板

  • 如果启用了记忆系统（Memory）：

    • 从 Redis 中读取该用户的最近 N 条历史对话
    • 一并传入 LLM Prompt Template 中

  • 统一调用 LLM，如

    • llm_chain.invoke({
  "context": "...",
  "question": "请总结本文重点"
})
      1
      2
      3
      4

  • 每生成一个 token，通过 LangChain callback 回传给 WebSocket 输出接口

# 4、基础设施层（调度 / 存储 / 日志 / 权限）

  • 职责：支撑整个智能体平台的运行时基础能力，包括持久化、权限控制、异步调度等

  • 技术栈：

    • 存储：MinIO（文档上传存储） + PostgreSQL（用户行为、调用日志）

    • 调度：Celery / Redis Queue（如长任务、异步分析任务）

    • 日志监控：Grafana + Loki + Promtail（接入所有 API 调用日志和 token 使用量）

    • 配置管理：Pydantic / Consul（用于切换 LLM、RAG、功能灰度）

  • 处理流程：

    • 上传的文档存储在 MinIO，对应路径映射为 file_id

    • 日志系统记录：

      • 每次用户请求内容、响应耗时、所调用模型、工具使用情况等

    • 如果任务耗时超长，调度系统将其抛入队列异步处理，用户稍后接收结果

    • 所有模型调用计费行为，异步写入 PostgreSQL 供后期审计和分账

# 01.用户入口层（API / UI）

# ✅ 核心职责

  • 提供人机交互、程序调用等统一入口
  • 负责请求鉴权、会话上下文绑定、用户态数据注入
  • 支持 Chat UI、Workflow UI、API/SDK、多轮交互等形态

# 🛠 技术实现要点

维度 内容
接口设计 提供 RESTful / WebSocket / GraphQL 接口，封装 Agent 调用入口
上下文维护 请求中注入用户信息、上下文标识符（用于 memory/RAG）
会话管理 管理会话生命周期（session_id、user_id 绑定 memory）
UI设计 简单场景可用 LangChainHub 的 Chat UI，复杂的用 React + 状态引擎

# 🔗 LangChain / LangGraph 应用点

  • 不直接负责 UI，但需要配合前端实现如流式响应（Streaming Output）
  • Agent 执行的入口函数由 API 控制调用：如 agent_executor.invoke(input_dict)

# 02.Agent 执行编排层（Core）

# ✅ 核心职责

  • 执行 Agent 的对话、推理、工具调用、RAG 等推理链路
  • 负责流程状态控制、角色切换、多 Agent 协同
  • 可支持复杂多阶段流程，如销售线索、产品推荐、任务调度等

# 🛠 技术实现要点

能力 实现要点
执行编排 使用 LangChain 的 AgentExecutor 进行基础编排，或使用 LangGraph 做状态流程控制
状态流转 LangGraph 提供基于 StateGraph 的状态机逻辑，适合多阶段对话、任务
多 Agent 协作 使用 LangGraph 支持多个节点（不同 Agent/角色）协作处理
容错机制 自定义节点重试逻辑、Fallback 节点、异常中断处理

# 🔗 LangChain / LangGraph 应用点

  • 简单流程：initialize_agent() → AgentExecutor（支持 ReAct/SelfAsk）

  • 复杂编排：LangGraph 构建状态图 → 每个 Node 映射一个业务意图或子 Agent

    python


复制编辑
from langgraph.graph import StateGraph

graph = StateGraph(state_schema=AgentState)
graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)
graph.set_entry_point("plan")
graph.set_finish_point("done")
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11

# 03.能力支撑层（LLM / Tool / Memory / RAG）

# ✅ 核心职责

  • 封装基础智能能力模块，供 Agent 执行层调用
  • 提供 LLM 模型、工具能力、记忆、知识库等支持
  • 所有模块需接口化、可复用、易插拔

# 🛠 技术实现要点

能力模块 实现要点
LLM 封装统一 LLM 接入层（OpenAI、Azure、DeepSeek、Qwen 等），支持流式、token限制
Tool 使用 LangChain Tool 注册，支持同步/异步调用、结构化 schema
Memory LangChain 内置多种 memory（Buffer、Summary、VectorStore 等），支持 session_id 多会话
RAG 向量化文档（OpenAIEmbeddings / HuggingFace）→ FAISS / Qdrant / Milvus → retriever → chain
多模型支持 抽象模型适配器接口，基于用途绑定不同模型（聊天、提问、生成 code）

# 🔗 LangChain 应用点示例

  • LLM：ChatOpenAI(model="gpt-4")

  • Tool：继承 BaseTool 自定义注册

  • Memory：

    python


复制编辑
ConversationBufferMemory(memory_key="chat_history", return_messages=True)
    1
    2
    3
    4
    5

  • RAG：

    python


复制编辑
from langchain.chains import RetrievalQA
chain = RetrievalQA.from_chain_type(llm=llm, retriever=vectorstore.as_retriever())
    1
    2
    3
    4
    5
    6

# 04.基础设施层（存储 / 调度 / 日志 / 权限）

# ✅ 核心职责

  • 提供 Agent 运行所依赖的系统基础能力
  • 支持任务调度、上下文持久化、日志采集、权限管理
  • 保证平台的性能、可靠性、安全性

# 🛠 技术实现要点

模块 实现要点
存储层 - Redis/MongoDB 管理 Memory 数据 - MinIO/OSS 存文档 - ClickHouse 存埋点日志
调度队列 - Celery / Arq / FastQueue 管理异步 Agent 调用 - 支持重试、超时、优先级
日志系统 - LangChain CallbackManager 对推理过程做细粒度日志采集 - 接入 Kafka / ELK / Loki 做观测
权限管理 - OAuth / JWT 鉴权用户 - 定义 Token 限额 / Tool 调用白名单 / API 调用审计
配置管理 - 使用 ETCD / Consul 做可动态热更新的配置平台（如开启哪些 Agent、控制哪些功能）
202.AI销售智能体

←

最近更新
01
06.LangGraph
06-09
02
202.AI销售智能体
06-07
03
05.快递Agent智能体
06-04
更多文章>
Theme by Vdoing | Copyright © 2019-2025 逍遥子 技术博客 京ICP备2021005373号