简介面向企业AI平台建设者与技术决策者顺丰科技的企业应用PPT完整呈现了Agentic AI从平台搭建到业务落地全过程的实战路径。全篇围绕Agent生态发展、Agentic工具与实际应用场景展开重点介绍自研EGPU池化、混合云推理优化、弹性资源调度以及模型广场对文本、视觉、语音等多模态大模型的统一接入。内容同时还涵盖基于Langfuse的LLMOps全链路跟踪、OTEL协议对齐、性能与效果测评、模块化开发、内部安全鉴权及全流程合规管控。压缩包内包含1个pptx演示文档大小约5.41MB已有258人浏览学习。读者可从中全面了解NL2SQL、客服意图识别、DeepSeek私有化部署等典型场景的方案细节获得模型选型、资源降本、API统一网关与多环境适配的落地经验为构建企业级智能体生态提供直观参考。1. 企业 Agentic AI 和单轮 RAG 的边界在哪最近很多团队手里都有一份叫做Agentic AI 企业应用.pptx的规划可真要动手时才发现单轮 RAG 只能做“检索—拼接—回答”而业务想要的却是模型自己判断接下来查哪个库、调哪个工具甚至在答案不可靠时主动纠正。我把这条链路收敛成 FastAPI LangChain LangGraph pgvector 形态的 Agentic RAG用 LangGraph 控制流程状态用 pgvector 存向量和业务元数据再用 FastAPI 暴露成企业系统可调用的服务。下面按这条思路展开后端工程师可以照着搭建架构师可以直接跳到参数与排障部分。2. 用 LangGraph 把 Agentic AI 的流程状态固定下来在企业场景里“Agentic”往往意味着模型可以选择先做什么、后做什么但这并不代表流程可以失控。LangGraph 把整个 Agent 表达成一张有向图图的节点是动作边是条件转移状态对象就是当前这轮会话的记忆。它和手写 while 循环最大的差别在于图的每一步都能显式挂载回调、持久化 checkpoint并且可以在任意节点上做权限和审计。2.1 为什么 LangGraph 比手写 while 循环更适合企业流程LangChain 的 LCEL 只适合无环的链式调用适合做简单的文档问答。但 Agentic AI 场景里模型可能先检索一次发现信息不足再改写查询重试甚至中途调用一个 SQL 工具。这种循环如果用 while 写一旦并发上来很难控制工具调用次数和终止条件。LangGraph 用StateGraph把这个循环变成“从 route 回到 retrieve”的显式边。我一般会按“检索—判断—再检索—回答”的最小模型起步而不是一开始就上多 Agent 编排。组件在企业里的职责选型理由LangGraph流程状态、循环、分支、checkpoint显式状态机可挂回调适合审计pgvector向量检索 业务字段过滤复用现有 PostgreSQL多租户隔离容易FastAPI对外 API、并发、限流异步性能好生态成熟LangChainLLM 封装、工具协议、检索抽象工具和回调模型统一开发效率高2.2 pgvector 在 Agent 里的作用是“可检索的企业记忆”Agent 的记忆分两层。短记忆是状态里的question、retrieved这些字段它们在一轮推理里临时存在长记忆是企业文档的向量索引。pgvector 是 PostgreSQL 的扩展可以存向量但更重要的是能同时存content、dept、doc_name这些业务字段。Agent 在改写查询时可以在检索前直接过滤部门或文档类型这正是企业做权限控制最常用的手段。CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS doc_chunks ( id bigserial PRIMARY KEY, content text, doc_name text, dept text, embedding vector(1536) ); CREATE INDEX ON doc_chunks USING hnsw (embedding vector_cosine_ops);这里的vector(1536)必须和 embedding 模型输出的维度一致常见的有 768、1024、1536。索引先按 1536 建好如果后面换 embedding 模型需要重建列和索引。文本字段单独保留是为了让 Agent 在检索后还能读取原始内容也方便后续做来源追溯。2.3 最小依赖与状态图骨架langchain-postgres是 LangChain 官方适配 pgvector 的包langchain-openai则用来统一接入 OpenAI 兼容的 embedding 和 chat 模型。依赖不需要一上来追求最新版LangChain 生态的 Breaking Change 比较多锁一个大版本段能省下很多调试时间。fastapi uvicorn langchain langchain-openai langgraph langchain-postgres pgvector接着定义一个最小的 LangGraph 状态图。节点函数只负责返回需要更新的状态字段路由函数负责决定下一步走向。from typing import TypedDict, Literal from langgraph.graph import StateGraph, END class AgentState(TypedDict): question: str plan: list[str] retrieved: list[str] answer: str def retrieve_node(state: AgentState) - dict: # 节点返回的 dict 会并回整体状态不要返回整个 state return {retrieved: []} def route_node(state: AgentState) - Literal[retrieve, answer]: # 实际判断可以交给 LLM但这步先留成规则 if state[retrieved]: return answer return retrieve def answer_node(state: AgentState) - dict: return {answer: ...} graph StateGraph(AgentState) graph.add_node(retrieve, retrieve_node) graph.add_node(route, route_node) graph.add_node(answer, answer_node) graph.set_entry_point(retrieve) graph.add_edge(retrieve, route) graph.add_conditional_edges( route, lambda s: route_node(s), {retrieve: retrieve, answer: answer} ) graph.add_edge(answer, END) compiled graph.compile()add_conditional_edges第一个参数是当前节点第二个参数是路由函数第三个参数是返回值到后继节点的映射。如果没有匹配到 keyLangGraph 会直接报错。这里route_node用的是简单规则实际可以换成 LLM 对“信息是否足够”的评分。3. 用 FastAPI 和 pgvector 把 Agentic RAG 拉通运行状态图只是骨架真正要面向企业交付还需要一个稳定 API 层和可替换的检索实现。FastAPI 在这里承担两个任务一是把 LangGraph 的调用封装成业务方方便对接的 POST 接口二是统一处理超时、限流和异常返回。3.1 在 FastAPI 端点里只暴露一个稳定的会话入口我给外部业务方暴露的路径永远是/v1/agent接口参数只保留question和session_id。session_id 对应 LangGraph 配置里的thread_id同一个会话再次调用时可以保留之前的状态。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(titleagentic-rag-service) class AgentRequest(BaseModel): question: str session_id: str default class AgentResponse(BaseModel): answer: str app.post(/v1/agent) def run_agent(req: AgentRequest) - AgentResponse: config {configurable: {thread_id: req.session_id}} output compiled.invoke({question: req.question}, configconfig) return AgentResponse(answeroutput[answer])如果不想让同一会话的状态被下一轮污染每次请求都传一个随机session_id即可。企业多轮对话场景里thread_id 通常会带上用户 ID 或工单 ID比如user_123:ticket_456方便按会话维度做日志检索。3.2 把 pgvector 检索器注入 retrieve 节点pgvector 接入 LangChain 的常见做法是用langchain_postgres.PGVector适配器。它负责把向量表抽象成 LangChain 的 VectorStore再用as_retriever转成检索器。from langchain_postgres import PGVector from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vector_store PGVector( embeddingsembeddings, collection_namedoc_chunks, connectionpostgresql://agent_admin:passlocalhost:5432/agentdb, ) retriever vector_store.as_retriever(search_kwargs{k: 6})这里的collection_name会映射到 pgvector 的集合管理表connection是标准 PostgreSQL 连接串。k6是召回条数不要一开始就取 20否则多轮检索叠加后上下文太长LLM 容易判断失焦。Agent 在规划阶段如果拆出多个子查询还需要在检索节点里循环执行查询并去重。def retrieve_node(state: AgentState) - dict: if not state.get(plan): state[plan] [state[question]] docs [] for q in state[plan]: docs.extend(retriever.invoke(q)) seen set() uniq [] for d in docs: if d.page_content not in seen: seen.add(d.page_content) uniq.append(d) return {retrieved: [d.page_content for d in uniq[:10]]}循环多个 query 是 Agentic RAG 和普通 RAG 的最大区别。不一定在第一轮就检索而是在 route 节点判断“是否缺某些事实”后动态把新 query 塞进plan。去重很关键同一个内容被两个 query 带回来后如果不去重会重复占用上下文窗口。3.3 让工具调用出现在 Agent 循环里企业 Agent 除了检索文档往往还要查客户 ID、查订单状态。LangGraph 不直接约定义 tool而是复用 LangChain 的 tool 协议由 LLM 决定是否调用。from langchain_core.tools import tool tool def lookup_customer_id(user_name: str) - int: 根据客户全名返回企业内部客户 ID # 这里接内部 CRM 接口 return 10086当模型认为需要调用工具时会返回一个 tool_calls 列表。Graph 里可以加一个execute_tool节点遍历 tool_calls 并执行然后把结果写回状态。这样工具和 RAG 检索是平行的不会互相阻塞。加上异常处理是必须的否则一个工具超时会拖垮整条 Agent 流程。from fastapi import HTTPException from langgraph.errors import GraphRecursionError try: output compiled.invoke( {question: req.question}, config{configurable: {thread_id: req.session_id}} ) except GraphRecursionError as exc: raise HTTPException(status_code504, detailagent recursion limit exceeded)GraphRecursionError是最常见的线上故障来源后面会专门讲怎么设置步数阈值。4. 给 Agentic AI 定阈值召回、步数与缓存边界运行起来只是第一步企业应用要看的是稳定性和成本。Agentic RAG 的链路比普通 RAG 长每个环节多一层循环成本和延迟都会成倍上涨。建议按“向量召回 → 图执行步数 → 缓存命中”的顺序调优。4.1 先调 pgvector 索引再调 LLM 温度很多人在跑通 POC 后第一反应是调 prompt 或温度但企业环境里最先应该调的是向量库。pgvector 默认创建索引后查询质量直接决定后续 LLM 判断准不准。线上数据量到十万级就必须明确使用 HNSW 索引。CREATE INDEX ON doc_chunks USING hnsw (embedding vector_cosine_ops) WITH (m 16, ef_construction 64); SET hnsw.ef_search 40;m越大召回越高内存占用也随之上升ef_construction决定索引构建质量太大会拖慢入库速度ef_search是每次查询时的动态召回范围。经验值如下。参数经验值说明m16 ~ 32超过 32 提升有限内存涨得快ef_construction64 ~ 128100 万级文档建议 128ef_search40 ~ 100对延迟敏感的接口先从 40 试k5 ~ 10最终进入上下文的条数别超过 154.2 LangGraph 递归上限要和业务路径层级对齐LangGraph 把每个节点调用都计入递归深度所以光跳两次检索就会消耗 45 步。我一般设recursion_limit20如果 Agent 链路里有工具调用、再检索、再判断20 步足以覆盖多数场景。超过 30 步的请求大概率是 Model 陷入循环了不如直接终止。compiled.invoke( {question: req.question}, config{ recursion_limit: 20, timeout: 60_000, configurable: {thread_id: req.session_id} } )这里的timeout单位是毫秒60 秒是单次请求的最长等待时间。企业内部接口一般要求 35 秒内返回但如果 Agent 要查多个数据源可以把超时放宽到 60 秒同时在前端加轮询状态。4.3 缓存不是整个 graph而是检索和最终答案我见过很多 Agent 应用死在高成本上因为每次请求都会把整条图完整跑一遍。正确的做法是按question dept做缓存键检索结果缓存 1060 秒最终答案在业务允许的情况下缓存更久。工具调用结果不要缓存因为你不知道内部系统的数据什么时候更新。cache_key f{req.question}:{req.session_id} cached cache.get(cache_key) if cached: return AgentResponse(answercached)缓存不能包住所有节点否则 Agent 无法感知新文档。对时效性敏感场景比如查库存或查订单最终答案的 TTL 建议设 30 秒或不缓存。只缓存检索片段能让 LLM 在保留事实来源的同时每次重新生成表达。5. 把 Agentic AI 的每一步变成可审计日志线上问题最难的不是改代码而是复现。Agent 是状态机状态一旦流失下一次请求可能走上完全不同的路径。所以我在生产环境里一定做一件事自定义回调处理器把每次 LLM 调用、工具调用、检索动作全部写成结构化日志。5.1 在 graph.invoke 上挂一个自记录回调LangGraph 的config支持直接传callbacks不需要改图结构。自定义回调继承BaseCallbackHandler只覆写自己关心的钩子。from langchain_core.callbacks import BaseCallbackHandler import json, time class TraceHandler(BaseCallbackHandler): def __init__(self): self.trace [] def on_llm_end(self, response, **kwargs): self.trace.append({ ts: time.time(), kind: llm, data: response.llm_output }) def on_tool_end(self, output, **kwargs): self.trace.append({ ts: time.time(), kind: tool, data: output }) handler TraceHandler() compiled.invoke( {question: req.question}, config{callbacks: [handler]} )这里的trace会跟着单次请求走请求结束后把trace连同session_id、run_id一起写入日志系统。llm_output里有 token 用量tool_end里能看到工具返回结果这两个值拼起来就能还原一次完整的失败路径。5.2 用日志回放错误而不是重新跑一遍遇到线上失败请求我会先把日志中的run_id过滤出来看它经过哪些节点、每一步的 prompt 是什么、工具返回了什么。这样能快速判断是规划错了、检索没召回还是工具参数传错了。jq select(.run_id9f3a...) | .prompt trace.log | head -20把这段回放脚本固化进 CI每次发版前自动跑通三个历史故障用例Agent 应用的可维护性就能往前推一大步。本文还有配套的精品资源点击获取