企业级 Agent 开发现在绕不开两个框架LangChain 和 LangGraph。LangChain 解决的是 LLM 应用开发的组件化问题把模型对话、提示词模板、工具调用变成标准化模块LangGraph 则把 Agent 从“一个循环调函数”升级成“一张可编排、可追踪、可恢复的状态图”。这次我们沿着一条完整项目主线走一遍先写一个最朴素的 Agent V1验证工具调用和记忆再做 LangGraph 工作流化处理条件路由、子图和并行分支接着落地一个 TextToSQL 智能体实现自然语言查询数据库最后把服务用 FastAPI 包起来加上日志、追踪和监控形成一整套可观测的企业级 Agent 开发方案。这篇文章适合三类读者已经在用 LangChain 写原型、想往生产环境收敛的开发者正在做企业级 AI 应用、需要把 Agent 流程结构化的架构师以及准备面试 LangChain/LangGraph Agent 相关岗位、想把自己项目讲清楚的同学。文章不追求面面俱到重点是给出一条从 V1 到可观测部署的完整路径代码可以直接复制到项目里改。1. 核心能力速览在动手之前先把这套技术栈的关键规格整理出来。能力项说明项目类型LangChain LangGraph 企业级 Agent 开发框架组合核心功能多节点工作流编排、工具调用、条件路由、记忆管理、TextToSQL、可观测部署语言要求Python 3.9建议 3.10 或 3.11硬件要求框架本身不占显存取决于 LLM 是 API 服务还是本地推理模型显存占用本地部署大模型时需要独立评估纯框架代码不涉及显存支持平台Windows / Linux / macOS启动方式脚本启动、docker compose启动、FastAPI 服务启动接口 API支持可封装为 REST API 供业务系统调用批量任务支持可基于 asyncio 或消息队列实现并发处理可观测性支持 LangSmith 追踪、OpenTelemetry 指标、结构化日志适合场景企业内部知识问答、数据分析助手、业务系统智能体集成这套技术栈的定位不是某个开箱即用的“一键包”而是给开发者一套标准的工程骨架。LangChain 负责组件层LangGraph 负责流程层FastAPI 负责服务层监控链路负责稳定性。四者组合起来才是一个企业级 Agent 项目该有的形态。2. LangChain 与 LangGraph框架分工与选择很多刚接触 Agent 的同学会问LangChain 和 LangGraph 到底有什么区别是不是 LangGraph 要取代 LangChain答案都不是。LangChain 提供的是 LLM 应用开发的基础组件模型封装、提示词模板、输出解析器、向量存储、工具抽象、文档加载器。它解决的是“怎么让模型能力变成可复用的代码模块”。比如ChatOpenAI把 OpenAI 的接口封装成统一模型接口tool装饰器把普通 Python 函数变成模型可以理解的工具 Schema。LangGraph 提供的是 Agent 的运行时骨架。它把 Agent 流程抽象成一张有向图节点是“做某件事”的函数边是“流程怎么走”的控制逻辑。核心特色包括显式的状态管理Agent 运行过程中的所有中间结果都保存在 State 中支持条件路由模型可以决定下一步走哪个分支支持循环解决 ReAct 框架中“工具调用 - 观察结果 - 继续思考”的循环问题支持子图可以把一个复杂的业务模块封装成独立的图再嵌入主图支持持久化可以把运行状态保存到数据库实现断点续跑。2.1 什么时候只用 LangChain如果业务场景只是“单轮问答 一个文档检索”不需要复杂的多步任务那么只使用 LangChain 就足够了。比如一个简单的 RAG 问答先通过向量检索拿到文档片段然后把片段拼到 Prompt 里让模型回答。这个过程是线性的用 LangChain 的LCEL表达式就能完成。from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.runnable import RunnablePassthrough prompt ChatPromptTemplate.from_template( 根据以下资料回答问题\n{context}\n\n问题{question} ) chain ( {context: retriever, question: RunnablePassthrough()} | prompt | ChatOpenAI(modelgpt-4o-mini, temperature0) )这种链路是静态的、可预测的不需要 LangGraph 参与。2.2 什么时候必须用 LangGraph当 Agent 需要自主决策“先做什么、再做什么”并且执行路径可能发生变化时就应该用 LangGraph。典型场景包括客服 Agent 先判断用户意图再调用不同业务系统数据分析 Agent 先识别问题类型再决定是否查询数据库多智能体协作一个调度 Agent 把任务分发给多个子 Agent。判断标准很简单如果流程里存在“根据模型输出决定下一步”的分支或者需要多次循环调用工具LangGraph 就是更合适的载体。“智能体”和“工作流”的边界也在这里工作流是固定路径智能体是模型参与决策的动态路径LangGraph 恰好同时支持这两种形态。3. 环境准备与项目初始化先把项目环境搭起来。这里给出一套通用的初始化方式实际项目根据团队技术栈调整。3.1 创建虚拟环境建议使用uv或conda管理 Python 环境。以uv为例uv venv agent-env --python 3.11 source agent-env/bin/activate如果使用 condaconda create -n agent-env python3.11 -y conda activate agent-env3.2 安装核心依赖pip install langchain langchain-openai langgraph langchain-core pip install fastapi uvicorn sqlalchemy psycopg2-binary pip install opentelemetry-api opentelemetry-sdk pip install opentelemetry-exporter-otlp pip install langsmith依赖版本需要锁定。LangChain 和 LangGraph 更新节奏很快生产环境建议把requirements.txt或pyproject.toml中的版本号固定下来避免出现“昨天还能跑今天升级完报错”的情况。3.3 环境变量创建.env文件统一管理配置OPENAI_API_KEYsk-xxxxxxxx OPENAI_API_BASEhttps://api.your-proxy.com/v1 LANGSMITH_API_KEYls-xxxxxxxx LANGSMITH_PROJECTagent-project DATABASE_URLpostgresql://user:passwordlocalhost:5432/businessLangSmith 负责链路追踪生产环境建议开启如果只做本地调试可以先不配置。3.4 验证环境写一个最简的 LangGraph 图确认安装无误from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): message: str def hello_node(state: State): return {message: hello langgraph} graph StateGraph(State) graph.add_node(hello, hello_node) graph.add_edge(START, hello) graph.add_edge(hello, END) app graph.compile() print(app.invoke({message: }))如果能输出{message: hello langgraph}环境就算跑通了。这一步很重要先排除环境问题再进入业务开发。4. Agent V1 实战工具调用与记忆先写一个最原始的 Agent V1。这个版本用 ReAct 模式模型决定调什么工具拿到工具结果再生成最终回答。4.1 定义一个业务工具假设要给业务方做一个订单查询智能体先定义一个查询订单状态的工具from langchain_core.tools import tool tool def query_order_status(order_id: str) - str: 根据订单ID查询订单状态返回订单当前状态信息。 # 这里替换成真实业务系统调用 order_data { SO20240101: 已发货预计3天内送达, SO20240102: 已支付待发货, } return order_data.get(order_id, 未找到该订单)tool装饰器会自动根据函数名、docstring 和参数生成模型的工具 Schema所以 docstring 必须写清楚“这个工具能做什么、参数是什么意思”。4.2 构建带工具能力的 Agent V1from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([query_order_status]) def agent_node(state): response llm_with_tools.invoke(state[messages]) return {messages: [response]}这里的关键是bind_tools模型在收到问题时如果判断需要查询订单会在返回内容里带上tool_calls字段而不会直接输出最终回答。4.3 工具执行节点from langchain_core.messages import ToolMessage from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] def tools_node(state): last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: result query_order_status.invoke(tool_call[args]) results.append(ToolMessage(contentresult, tool_call_idtool_call[id])) return {messages: results}add_messages是 LangGraph 内置的 reducer作用是自动把新的消息追加到已有的 messages 列表中而不是覆盖。4.4 编译并运行from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(START, agent) # 条件路由有工具调用就去 tools 节点否则结束 def should_continue(state): last_message state[messages][-1] return tools if last_message.tool_calls else END graph.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) graph.add_edge(tools, agent) app graph.compile() result app.invoke({messages: [(human, 查询订单 SO20240101 的状态)]}) print(result[messages][-1].content)这段代码就是 Agent V1 的完整形态模型负责决策工具负责执行循环由 LangGraph 的条件边控制。整个过程中每一次“模型调用 - 工具调用 - 模型再调用”都是一轮完整的循环LangGraph 会把这个过程记录在状态里。4.5 记忆让 Agent 记住上下文V1 版本虽然能跑通但有个明显问题多轮对话中模型无法记住之前的会话内容。解决办法是引入消息持久化。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user-order-001}} # 第一轮 result1 app.invoke( {messages: [(human, 我是张三)]}, configconfig, ) # 第二轮模型能记住上一轮的对话 result2 app.invoke( {messages: [(human, 我刚刚说了我是谁)]}, configconfig, ) print(result2[messages][-1].content)thread_id是会话标识同一个会话内的消息会累积保存。企业级场景中MemorySaver只适合开发调试生产环境建议使用基于 PostgreSQL 或 Redis 的持久化 checkpoint这样服务重启后会话不会丢。5. LangGraph 工作流化状态机、条件路由与子图Agent V1 跑通之后下一步是把它改造成企业级工作流。企业级系统的特点是流程不是一条线走完而是有分支、有校验、有回退、有多模块协作。LangGraph 的图化设计正好对应这个需求。5.1 需求场景设计假设我们要做一个“智能客服工单处理”工作流流程如下接收用户问题判断问题类型订单查询 / 退款售后 / 商品咨询根据类型调用不同业务工具判断工具结果是否需要人工复核如需复核转到人工节点否则直接返回用户。5.2 状态定义from typing import Annotated, TypedDict, Literal from langgraph.graph.message import add_messages class WorkflowState(TypedDict): messages: Annotated[list, add_messages] intent: str need_human: bool final_answer: str每个节点都可以读取和更新 State这是 LangGraph 的核心机制。intent和need_human就是我们自定义的业务状态用来在不同节点之间传递决策结果。5.3 节点函数def intent_node(state: WorkflowState): 判断用户意图 prompt f判断以下用户问题的意图只回答order/refund/product 中的一个。 用户问题: {state[messages][-1].content} response llm.invoke(prompt) return {intent: response.content.strip()} def order_node(state: WorkflowState): 订单查询处理 # 实际业务中解析订单号并调用订单服务 return {messages: [(ai, 您的订单已发货预计3天内送达)], need_human: False} def refund_node(state: WorkflowState): 退款售后处理 return {messages: [(ai, 退款申请已提交预计1-3个工作日处理)], need_human: False} def product_node(state: WorkflowState): 商品咨询处理 return {messages: [(ai, 根据商品信息这款产品支持7天无理由退货)], need_human: False} def human_review_node(state: WorkflowState): 人工复核节点 return {messages: [(ai, 您的问题已转接人工客服请稍候)], need_human: True}5.4 条件路由条件路由用add_conditional_edges实现from langgraph.graph import StateGraph, START, END graph StateGraph(WorkflowState) graph.add_node(intent, intent_node) graph.add_node(order, order_node) graph.add_node(refund, refund_node) graph.add_node(product, product_node) graph.add_node(human_review, human_review_node) graph.add_edge(START, intent) # 根据意图路由到不同处理节点 def route_by_intent(state: WorkflowState): return state[intent] graph.add_conditional_edges( intent, route_by_intent, { order: order, refund: refund, product: product, }, ) # 三个业务节点都可以路由到人工复核或结束 for node_name in [order, refund, product]: graph.add_conditional_edges( node_name, lambda state: human_review if state[need_human] else END, {human_review: human_review, END: END}, ) graph.add_edge(human_review, END) app graph.compile()这里的关键点是路由函数返回的是字符串映射表把字符串映射到具体节点。这种设计让路由逻辑和节点逻辑完全解耦企业级项目里可以根据线上数据随时调整路由策略。5.5 子图当业务模块变大把所有节点塞在一个图里会难以维护。LangGraph 支持子图可以把退款流程封装成一个独立的图再嵌入主图from langgraph.graph import StateGraph, START, END # 子图退款流程 refund_subgraph StateGraph(WorkflowState) refund_subgraph.add_node(validate_refund, validate_refund_node) refund_subgraph.add_node(process_refund, process_refund_node) refund_subgraph.add_edge(START, validate_refund) refund_subgraph.add_edge(validate_refund, process_refund) refund_subgraph.add_edge(process_refund, END) refund_subgraph_compiled refund_subgraph.compile() # 主图直接把子图作为节点 main_graph StateGraph(WorkflowState) main_graph.add_node(intent, intent_node) main_graph.add_node(refund_flow, refund_subgraph_compiled) main_graph.add_edge(START, intent) def route_by_intent(state): return refund_flow if state[intent] refund else other_flow main_graph.add_conditional_edges( intent, route_by_intent, {refund_flow: refund_flow, other_flow: other_flow}, )子图内部怎么改不影响主图这种封装方式在企业级多团队协作中非常有用。5.6 并行分支LangGraph 也支持并行执行。如果一个问题需要同时查订单系统、库存系统和物流系统可以用扇出结构def fan_out_node(state): return state # 并行执行三个节点 graph.add_node(fan_out, fan_out_node) graph.add_node(query_order, query_order_node) graph.add_node(query_stock, query_stock_node) graph.add_node(query_logistics, query_logistics_node) graph.add_edge(START, fan_out) graph.add_edge(fan_out, query_order) graph.add_edge(fan_out, query_stock) graph.add_edge(fan_out, query_logistics) graph.add_edge(query_order, aggregate) graph.add_edge(query_stock, aggregate) graph.add_edge(query_logistics, aggregate)并行节点有多个入边LangGraph 会在所有前置节点完成后自动汇合到aggregate节点。实际并发度取决于底层运行时的线程池配置需要在性能测试阶段重点观察。6. TextToSQL Agent 落地实战TextToSQL 是企业级 Agent 中很实用的场景业务人员用自然语言提问系统自动生成 SQL 并返回查询结果。但这里有一个非常重要的前提生产环境必须使用只读账号并且对查询做严格的白名单和长度限制。6.1 工具定义import json import sqlalchemy as sa from langchain_core.tools import tool from typing import List tool def execute_query_sql(sql: str) - str: 执行只读SQL查询。仅支持SELECT语句禁止修改数据。 sql sql.strip().lower() if not sql.startswith(select): return 错误仅允许执行SELECT查询语句。 if len(sql) 500: return 错误SQL语句过长请简化查询条件。 engine sa.create_engine(DATABASE_URL) with engine.connect() as conn: result conn.execute(sa.text(sql)) rows result.mappings().all() if not rows: return 查询结果为空。 return json.dumps([dict(row) for row in rows], ensure_asciiFalse, defaultstr)注意三个安全点只允许 SELECT、限制 SQL 长度、连接使用只读数据库账号。企业级环境中还应该加一层 SQL 解析校验拦截information_schema之外的系统表访问。6.2 表结构提示TextToSQL 的准确性很大程度上取决于模型是否知道表结构。把表结构注入到系统提示词中SYSTEM_PROMPT 你是一个数据分析助手。数据库包含以下表 1. orders 表订单表 - order_id: 订单ID - customer_id: 客户ID - product_name: 商品名称 - amount: 订单金额 - create_time: 下单时间 2. customers 表客户表 - customer_id: 客户ID - customer_name: 客户名称 - register_time: 注册时间 根据用户的问题生成对应的 SQLSQL 必须是只读查询。 如果问题不明确需要追问用户确认。 6.3 Agent 构建from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from typing import Annotated, TypedDict from langgraph.graph.message import add_messages llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([execute_query_sql]) class SQLState(TypedDict): messages: Annotated[list, add_messages] final_answer: str def sql_agent_node(state: SQLState): messages [{role: system, content: SYSTEM_PROMPT}] state[messages] response llm_with_tools.invoke(messages) return {messages: [response]} def sql_tools_node(state: SQLState): last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: result execute_query_sql.invoke(tool_call[args]) results.append(ToolMessage(contentresult, tool_call_idtool_call[id])) return {messages: results} def should_continue(state: SQLState): last_message state[messages][-1] return tools if last_message.tool_calls else END graph StateGraph(SQLState) graph.add_node(agent, sql_agent_node) graph.add_node(tools, sql_tools_node) graph.add_edge(START, agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) graph.add_edge(tools, agent) sql_app graph.compile()6.4 测试用例test_cases [ 上个月销售额最高的前5个商品是什么, 最近7天新增了多少客户, 删除所有客户的订单记录, # 必须被拒绝 ] for question in test_cases: result sql_app.invoke({messages: [(human, question)]}) print(f问题: {question}) print(f回答: {result[messages][-1].content}) print(---)第三类测试用例很关键验证系统是否能拒绝危险操作。如果模型直接生成了DELETE语句即使工具层拦截了也说明系统提示词还需要调整。7. 可观测部署日志、追踪与监控部署是企业级 Agent 落地中最容易被忽视的一环。Agent 不是普通的 HTTP 接口它的执行路径是动态的同一个问题可能走不同的工具调用链。如果没有可观测性线上出问题时基本只能靠猜。7.1 LangSmith 集成LangSmith 是 LangChain 官方的可观测平台打开追踪后每条 Agent 执行记录都会展示完整的调用链export LANGSMITH_TRACINGtrue export LANGSMITH_API_KEYls-xxxxxxxx export LANGSMITH_PROJECTagent-project代码层面不需要额外改动LangChain 和 LangGraph 会自动上报。通过 LangSmith 可以看到模型调用的 token 消耗、每个节点的执行时间、工具调用的输入输出以及条件路由实际走到了哪个分支。7.2 结构化日志LangSmith 记录的是单条调用但企业运维还会关心整体日志。建议使用 Python 标准logging模块输出 JSON 格式的结构化日志import json import logging class JsonFormatter(logging.Formatter): def format(self, record): log_entry { timestamp: self.formatTime(record, %Y-%m-%d %H:%M:%S), level: record.levelname, logger: record.name, message: record.getMessage(), } if hasattr(record, request_id): log_entry[request_id] record.request_id return json.dumps(log_entry, ensure_asciiFalse) logger logging.getLogger(agent-service) handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO)在关键节点埋点def sql_agent_node(state): logger.info(开始处理SQL意图, extra{request_id: state.get(request_id)}) response llm_with_tools.invoke(state[messages]) logger.info(模型返回完成, extra{request_id: state.get(request_id), tool_calls: len(response.tool_calls)}) return {messages: [response]}7.3 OpenTelemetry 指标监控LangSmith 解决单链路追踪OpenTelemetry 解决系统级指标。可以把每次调用的延迟和 token 消耗暴露成 Prometheus 指标from opentelemetry import metrics from opentelemetry.exporter.prometheus import PrometheusMetricExporter from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader reader PeriodicExportingMetricReader(PrometheusMetricExporter()) provider MeterProvider(metric_readers[reader]) metrics.set_meter_provider(provider) meter metrics.get_meter(agent-service) request_counter meter.create_counter( nameagent.requests.total, descriptionAgent请求总数, unit1, ) request_latency meter.create_histogram( nameagent.request.latency, descriptionAgent请求耗时, unitms, )在每个请求完成时记录指标。这样 Grafana 上就能看到 QPS、P95 延迟、错误率、Token 消耗等关键指标配合告警规则线上问题可以提前暴露。7.4 部署方式对比langgraph dev与自建 FastAPILangGraph 官方推荐用langgraph dev在本地启动开发服务器连接 LangGraph Studio 可视化调试工作流langgraph dev这个命令会启动一个带热更新的调试服务适合开发阶段。但它需要额外的运行时依赖并且通常需要配置 Redis、PostgreSQL 等外部服务不适合直接作为生产入口。生产环境更稳妥的方式是自建 FastAPI 服务把编译好的图挂上去uv run uvicorn app.main:app --host 0.0.0.0 --port 8000两者区别总结对比项langgraph dev自建 FastAPI 服务定位本地开发调试生产对外服务可视化内置 LangGraph Studio需要自定义可接 Grafana灵活性受框架默认配置限制完全自主控制中间件、鉴权、日志稳定性适合开发环境适合线上多实例部署扩展点较少可集成任意 Python 中间件从实际项目经验看开发阶段用langgraph dev可以快速验证图逻辑进入测试和上线阶段必须切到自建 FastAPI 服务否则很难做标准化的发布和监控。7.5 Docker 部署示例version: 3.8 services: agent-api: build: . ports: - 8000:8000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} - DATABASE_URL${DATABASE_URL} - LANGSMITH_API_KEY${LANGSMITH_API_KEY} - LANGSMITH_TRACINGtrue deploy: resources: limits: memory: 1G restart: always healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 30s timeout: 5s retries: 3配置文件写好后执行docker compose up -d即可拉起服务。生产环境建议在容器前面加一层 Nginx 或云负载均衡统一处理 TLS 和流量分发。8. 接口 API 与批量任务设计Agent 服务最终要对外提供接口设计时必须考虑三个方面同步还是异步、批量任务怎么跑、失败怎么处理。8.1 FastAPI 接口封装from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(titleAgent Service) class QueryRequest(BaseModel): question: str thread_id: str default user_id: str | None None class QueryResponse(BaseModel): answer: str thread_id: str app.post(/agent/query) async def agent_query(req: QueryRequest): config {configurable: {thread_id: req.thread_id}} try: result await app_graph.ainvoke( {messages: [(human, req.question)]}, configconfig, ) return QueryResponse( answerresult[messages][-1].content, thread_idreq.thread_id, ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: ok}thread_id的语义很重要同一个thread_id的多轮请求会共享上下文不同用户必须使用不同的thread_id避免串数据。8.2 curl 调用示例curl -X POST http://localhost:8000/agent/query \ -H Content-Type: application/json \ -d { question: 查询订单 SO20240101 的状态, thread_id: user-001 }预期返回{ answer: 您的订单已发货预计3天内送达, thread_id: user-001 }8.3 批量任务批量业务的复杂度不在于调用多少次而在于任务状态管理。简单场景用并发协程生产环境建议引入消息队列。import asyncio async def run_batch(questions: list[str], thread_ids: list[str]): configs [ {configurable: {thread_id: tid}} for tid in thread_ids ] tasks [ app_graph.ainvoke( {messages: [(human, q)]}, configcfg, ) for q, cfg in zip(questions, configs) ] results await asyncio.gather(*tasks, return_exceptionsTrue) answers [] for r in results: if isinstance(r, Exception): answers.append({error: str(r)}) else: answers.append({answer: r[messages][-1].content}) return answers批量任务有几个常见坑LLM API 有并发限制需要做限流控制单个问题超时会影响整批任务建议加超时result await asyncio.wait_for(task, timeout30)批量执行要写结果日志方便失败后重跑。8.4 失败重试from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10) ) def call_agent_with_retry(question: str, thread_id: str): result app_graph.invoke( {messages: [(human, question)]}, config{configurable: {thread_id: thread_id}}, ) return result[messages][-1].content这里tenacity会自动做指数退避重试第一次失败等 2 秒第二次失败等 4 秒最多重试 3 次。注意要区分哪些错误可以重试API 超时可以重试SQL 语法错误重试 10 次也没用。9. 性能观察、常见问题与最佳实践9.1 性能观察方法Agent 服务的耗时主要来自三块LLM 推理、工具执行、网络 IO。排查性能问题不要上来就优化代码先看链路追踪数据。LangSmith 上能直接看到每个节点耗时先用数据定位瓶颈。观察项查看方式优化方向LLM 调用耗时LangSmith 单节点耗时换更快模型、减少上下文长度工具执行耗时日志中工具节点耗时优化 SQL、加缓存、并行调用SQL 查询耗时数据库慢查询日志加索引、限制返回行数总耗时LangSmith Trace 总时长优先优化最长节点Token 消耗LangSmith Token 统计截断历史消息、精简提示词9.2 常见问题排查问题现象可能原因排查方式解决方案启动后接口 404FastAPI 路由没注册或服务没跑起来日志中找 startup 记录访问 /docs 验证确认路由注册代码检查端口模型返回空内容API Key 失效或上下文超长单独调用 LLM 测试检查 API Key 和请求长度Agent 无限循环工具返回内容不符合模型预期模型反复调用同一个工具LangSmith 查看循环路径增加工具返回内容约束或设置最大递归次数TextToSQL 生成错误 SQL表结构提示不完整检查系统提示词中的表结构信息补充字段注释和示例查询批量任务部分失败LLM API 并发限流查看错误码和日志增加限流和失败重试追踪数据不显示环境变量没配置或 LangSmith 网络不通确认环境变量查看 SDK 日志重新配置环境变量端口冲突9000/8000 端口被占用lsof -i:8000查看占用进程换端口或杀进程多轮对话上下文串问题thread_id 使用错误检查请求日志中的 thread_id按用户维度分配独立 thread_id9.3 生产环境最佳实践第一数据库权限最小化。TextToSQL 场景必须使用只读账号禁止使用业务主账号连接数据库。可以定期检查数据库连接池监控确认没有 DDL/DML 操作。第二过程数据和敏感信息脱敏。日志、追踪系统里可能包含用户输入的个人信息生产环境要配置脱敏规则避免把手机号、身份证号、业务密钥写进日志。第三版本锁定。LangChain 和 LangGraph 每周都在更新发布前把依赖版本固定到pyproject.toml或requirements.txt升级时单独做回归测试。第四保留最小可运行配置。团队协作时经常出现“别人机器上跑得好好的我这跑不起来”建议维护一个examples/minimal示例目录只依赖少量第三方库用于新环境快速验证。第五建立测试用例集。Agent 测试不能只看“能不能跑”要维护一组固定测试用例覆盖正常问题、边界问题、恶意 SQL、无结果查询、模型幻觉场景。每次调整都要回归这组用例。10. 总结与下一步从 Agent V1 到 LangGraph 工作流再到 TextToSQL 落地和可观测部署这条链路覆盖了一个企业级智能体从原型到生产环境的全部关键环节。最值得先验证的是 LangGraph 的条件路由和状态管理这两个能力决定了 Agent 能否从“演示项目”变成“线上业务系统”。最容易踩的坑则是版本漂移和数据安全问题前者靠锁定依赖解决后者靠只读账号和安全校验解决。如果继续深入下一步可以做三件事把 TextToSQL 的表结构信息做成动态加载支持更多业务库引入 Redis 队列替代批量任务的并发协程提升任务吞吐量把 LangSmith 的数据同步到自建监控大盘形成统一可观测体系。企业级 Agent 开发的本质不是跑通一个功能而是让这个功能可以长期、稳定、可追溯地在业务环境里运行这套技术栈恰好提供了完整的解决方案。