LangChain+LangGraph实现企业级Agent：从Demo到工作流编排与可观测部署

这两年 AI Agent 已经从概念变成很多团队 KPI 里的关键词。但如果你真的用 LangChain 写过 Agent，大概率会遇到这样一组问题：本地 Demo 调得很顺，模型回答也很聪明，一接真实业务就发现流程不可控、结果不可复现、出了问题查不到上下文。这不是模型不够强，而是编排层没跟上。

这篇内容想聊清楚一件事：用 LangChain + LangGraph 做企业级 Agent，真正的关键点不是“多会调 Prompt”，而是把 Agent 从“自由发挥的对话模型”改造成“有状态、可路由、可观察、可回滚的工程系统”。本文将沿着一条完整路线展开：先讲 LangChain 与 LangGraph 的分工与区别，再带读者从 V1 版本的最小 Agent 骨架出发，逐步重构为工作流编排的 StateGraph，接着用一个 TextToSQL Agent 作为综合实战，最后落到可观测部署与运维排查。

读完你应该能回答三个问题：第一，LangGraph 到底解决 LangChain 的哪些短板；第二，如何把一个随手能跑的 Agent 改造成适合生产的工作流；第三，企业级 Agent 上线前，哪些监控和验证环节不能跳过。

1. 这篇文章真正要解决的问题

现在网上关于 LangChain 和 LangGraph 的教程很多，但大多数停留在“怎么装、怎么跑”的层面。很多团队照教程写完一个 Agent，进入真实业务后遇到的第一波问题往往不是模型回答错误，而是流程本身出了问题：某个工具调用超时了，整个对话卡死；Agent 在某个节点反复循环，Token 费用飙升；一次回答依赖好几个中间状态，但程序一重启状态就丢了。这些问题有一个共同根源：Agent 没有明确的流程边界和状态管理。

企业级 Agent 和普通 Demo 最大的区别，体现在四个维度上：

• 可控性：节点是否可编程、可跳转、可终止，而不是完全交给模型自由发挥。
• 可观测性：每轮对话、每次工具调用、每次模型请求是否都有完整 Trace。
• 状态持久化：多轮对话、长任务、异步执行时，状态能否被持久化和恢复。
• 安全性：Agent 访问数据、触发外部操作时，有没有权限边界和风控手段。

从 V1 到工作流再到 TextToSQL 和可观测部署，正好覆盖了这四个维度。V1 解决的是“能不能跑”，工作流解决的是“能不能控”，TextToSQL 解决的是“能不能做真实的业务动作”，可观测部署解决的是“上线后能不能维护”。

这篇文章适合三类读者：一是已经在用 LangChain 写 Agent，但总觉得项目推进不下去的开发者；二是准备在公司内部搭建 Agent 服务，需要向团队解释技术选型的架构师；三是刚接触 LLM 应用开发，想把概念一次性理清、少走弯路的后端工程师。如果你属于其中一类，读完之后可以沿着文中的代码结构直接改造自己的项目。

2. LangChain 与 LangGraph：先搞清楚分工再动手

很多初学者第一次看到“LangChain 和 LangGraph 的区别”这个问题时，第一反应是 LangGraph 是 LangChain 的升级版。这个理解不够准确。从官方定位来看，LangChain 是一整套围绕 LLM 应用开发的基础工具集合，包括模型封装、提示词模板、检索增强（RAG）、工具调用、输出解析和记忆组件；而 LangGraph 是其中的有状态编排引擎，专门用来把大模型、工具和外部系统组织成可控的图结构。

可以把 LangChain 理解成一个“工具箱”，LangGraph 则是“流水线控制系统”。工具箱里有很多好用的工具：ChatOpenAI 封装了大模型调用，DocumentLoader 负责文档加载，有很多现成的检索器。但“每一步先做什么、再做什么、出错时怎么办”，在传统 LangChain 代码里是没什么体现的。最常见的写法是这样：用 AgentExecutor 挂上一堆工具，把任务丢给模型，让模型自己决定调用顺序。

# 传统 LangChain Agent 写法（简化示例） from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.agents import Tool llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) tools = [ Tool(name="search", func=search_func, description="搜索公开信息"), Tool(name="calculator", func=calc_func, description="数学计算"), ] agent = create_react_agent(llm, tools, prompt) executor = AgentExecutor(agent=agent, tools=tools) result = executor.invoke({"input": "帮我查一下上海今天的天气"})

这种方式写起来非常快，但它的核心逻辑是模型在循环里自主决策。模型说“调用搜索”就调用搜索，说“结束”就结束。在简单场景下这没问题，但你会发现：流程无法固定、状态无法持久化、中间步骤无法被外部系统观测和控制。模型返回出错时，你很难知道它到底经过了哪些节点。

LangGraph 改变了这个思路。它把一次 Agent 任务建模成一个带状态的图：

• State（状态）：在图执行过程中共享的数据结构，类似 Web 服务里的请求上下文，每一轮节点更新后都写入新的状态。
• Node（节点）：一个具体的处理步骤，可以是大模型调用、工具执行、规则判断或数据库读写。
• Edge（边）：节点之间的连接关系，表示执行顺序。
• Conditional Edge（条件边）：根据当前状态决定走哪个分支的路由器。
• Subgraph（子图）：一个大 Agent 内部嵌套的独立小图，用于模块化复用。

和传统 AgentExecutor 最大的区别是，LangGraph 把执行路径从“模型自由决定”改成了“开发者定义路由规则，模型在规则内做选择”。这意味着你可以在关键节点上加上校验、分支和熔断逻辑，Agent 的每一步都变得可预期、可追踪。

下表可以帮你快速区分两者：

所以更稳妥的判断是：如果你只是做原型或内部小工具，LangChain 的 AgentExecutor 仍然够用；如果你的 Agent 要接真实业务、要服务多用户、要做多轮长任务，LangGraph 才是那个把工程问题交给开发者的关键组件。

3. 环境准备与项目基础结构

进入实操之前，先准备好运行环境。下面是本文建议的最小环境：

• 操作系统：macOS / Linux / Windows（WSL 更推荐）
• Python 版本：建议 3.10 及以上
• 包管理：pip 或 poetry，推荐创建独立虚拟环境
• 大模型 API：OpenAI 兼容接口即可，也可以用国内大模型厂商的 OpenAI 兼容服务
• 关键依赖：langchain、langchain-openai、langgraph、langsmith

建议先创建虚拟环境，再安装依赖：

# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate # 安装核心依赖 pip install --upgrade pip pip install langchain langchain-openai langgraph langsmith

如果你的服务器无法访问外部模型服务，也可以在本地用 vLLM 或 Ollama 启动 OpenAI 兼容的服务，然后在环境变量中替换 base_url。本文示例代码集中在 LangGraph 编排逻辑上，所以模型 API 选用不影响核心结构。

项目目录建议如下：

agent_project/ ├── .env # API Key 与环境配置 ├── requirements.txt # 依赖锁定文件 ├── app/ │ ├── __init__.py │ ├── state.py # Agent 状态定义 │ ├── nodes.py # 图节点实现 │ ├── router.py # 条件路由函数 │ ├── graph.py # 图构建与编译 │ └── tools.py # 工具函数与工具注册 ├── examples/ │ └── v1_simple_agent.py # V1 最小示例 └── tests/ └── test_basic_flow.py

在项目根目录准备.env文件，主要包含模型服务配置：

# .env OPENAI_API_KEY=sk-xxxx OPENAI_BASE_URL=https://api.openai.com/v1 LANGCHAIN_TRACING_V2=true LANGCHAIN_API_KEY=lsv2-xxxx LANGCHAIN_PROJECT=enterprise-agent-demo

这里要特别提醒一句：不要把 API Key 提交到 Git 仓库。生产环境的密钥管理应该放到密钥管理服务或 CI/CD 的 Secret 中。

4. V1 Agent：先跑通最小可用的 Agent 骨架

先写一个最朴素的 V1 Agent。目标是：模型能调用工具、能返回最终答案，代码尽量短，先跑通链路。后面再基于这个版本逐步改造。

4.1 定义一个简单的计算工具

# 文件路径：app/tools.py def add(a: float, b: float) -> float: """两个数字相加""" return a + b def multiply(a: float, b: float) -> float: """两个数字相乘""" return a * b TOOLS = [ { "type": "function", "function": { "name": "add", "description": "计算两个数字的和", "parameters": { "type": "object", "properties": { "a": {"type": "number", "description": "第一个数字"}, "b": {"type": "number", "description": "第二个数字"} }, "required": ["a", "b"] } } }, { "type": "function", "function": { "name": "multiply", "description": "计算两个数字的乘积", "parameters": { "type": "object", "properties": { "a": {"type": "number", "description": "第一个数字"}, "b": {"type": "number", "description": "第二个数字"} }, "required": ["a", "b"] } } } ]

工具定义采用 OpenAI Function Calling 风格，LangChain 0.2 以上版本可以直接把这种格式传给模型，模型会在需要时返回工具名称和参数。

4.2 用 LangChain 封装 AgentExecutor

# 文件路径：examples/v1_simple_agent.py import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate from app.tools import TOOLS llm = ChatOpenAI( model="gpt-4o-mini", temperature=0, ) tools = [ { "name": "add", "description": "计算两个数字的和", "func": lambda a, b: a + b, }, { "name": "multiply", "description": "计算两个数字的乘积", "func": lambda a, b: a * b, }, ] # 这里使用工具描述，让模型知道有哪些函数可调用 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个能调用计算工具的助手。请根据用户问题选择合适的工具。"), ("human", "{input}"), ]) # 为了简化示例，我们直接把工具名称映射到本地函数 def call_tool(tool_name: str, args: dict): fn_map = { "add": lambda: tools[0]["func"](args["a"], args["b"]), "multiply": lambda: tools[1]["func"](args["a"], args["b"]), } return fn_map[tool_name]() model_with_tools = llm.bind_tools(TOOLS) # 自定义 Agent 循环：调用模型 -> 如果返回工具则执行 -> 继续调用模型 def simple_agent(user_input: str, max_iterations: int = 5): messages = [ {"role": "system", "content": "你是一个能调用计算工具的助手。请根据用户问题选择合适的工具。"}, {"role": "user", "content": user_input}, ] for _ in range(max_iterations): response = model_with_tools.invoke(messages) # 如果模型没有返回工具调用，说明已经给出最终答案 if not response.tool_calls: return response.content # 逐个执行工具调用 for tool_call in response.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] result = call_tool(tool_name, tool_args) # 把工具执行结果追加到对话历史 messages.append({ "role": "assistant", "content": response.content, "tool_calls": [tool_call], }) messages.append({ "role": "tool", "tool_call_id": tool_call["id"], "content": str(result), }) return "达到最大迭代次数，未得到最终结果。" if __name__ == "__main__": print(simple_agent("请计算 12 和 5 的和，再乘以 3"))

这个示例没有直接依赖 AgentExecutor，而是手动实现了一个最小的“模型-工具-模型”循环，目的是把 Agent 的内部机制暴露出来。你会发现，V1 Agent 的问题非常明显：

1. 循环条件固定：只能靠 max_iterations 硬限制，模型失败后没有分支兜底。
2. 状态不透明：整个 messages 列表是隐式状态，外部看不到中间过程。
3. 无结构化输出：如果某个工具调用失败，你只能在字符串里等待模型自我纠错。
4. 不可观测：没有 Trace，出问题只能靠 print 猜测。

这个版本适合用来验证模型能力和工具逻辑，但不适合进生产。接下来的章节，我们就用它做基底，改造成 LangGraph 工作流。

5. 从 V1 到工作流：StateGraph 条件路由与子图编排

LangGraph 的核心价值在于把 Agent 执行过程变成一张图，开发者可以定义状态、节点、边和条件路由。这里我们用同一套“计算工具”场景，演示怎么把 V1 改造成 StateGraph。

5.1 定义状态结构

# 文件路径：app/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class AgentState(TypedDict): # 消息列表，add_messages 是 LangGraph 提供的增量合并器 messages: Annotated[list, add_messages] # 当前已经尝试过的工具执行次数，用于循环检测 tool_attempts: int # 最终结果 answer: str

这里Annotated[list, add_messages]是一个关键设计：它告诉 LangGraph，当多个节点更新 messages 字段时，不是直接覆盖，而是把新消息追加到旧消息后面。这个机制对应 Agent 多轮对话的上下文累积需求。

5.2 编写节点函数

# 文件路径：app/nodes.py from app.state import AgentState from app.tools import TOOLS def llm_node(state: AgentState): """调用大模型，返回最终回答或工具调用指令""" from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) model_with_tools = llm.bind_tools(TOOLS) response = model_with_tools.invoke(state["messages"]) return {"messages": [response]} def tool_node(state: AgentState): """执行模型请求的工具调用，并把结果写回状态""" from langchain_core.messages import ToolMessage tool_map = { "add": lambda args: args["a"] + args["b"], "multiply": lambda args: args["a"] * args["b"], } messages = [] attempts = state.get("tool_attempts", 0) for tool_call in state["messages"][-1].tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] try: result = tool_map[tool_name](tool_args) content = str(result) except Exception as e: content = f"工具执行失败: {str(e)}" messages.append(ToolMessage( content=content, tool_call_id=tool_call["id"], )) attempts += 1 return {"messages": messages, "tool_attempts": attempts}

关键逻辑说明：

• llm_node负责推理，它会基于当前所有历史消息决定是返回最终答案还是调用工具。
• tool_node负责执行，它读取模型返回的最后一个消息中的 tool_calls 字段，执行对应工具，并把结果作为 ToolMessage 追加到消息列表。
• 如果工具执行出错，我们不是让程序崩溃，而是把错误信息作为 ToolMessage 返回给模型，让模型自行决定如何修正。这是 Agent 容错的关键设计。

5.3 条件路由：模型下一步该去哪里

# 文件路径：app/router.py def should_continue(state: AgentState) -> str: """ 根据模型返回判断走工具节点还是结束 返回字符串是路由分支的名称 """ last_message = state["messages"][-1] if last_message.tool_calls: # 还有工具调用，进入工具执行节点 return "tool" # 没有工具调用，说明模型已经给出最终回答 return "end"

条件路由是 LangGraph 最重要的能力。它并不复杂，本质就是一个普通 Python 函数，输入当前状态，输出一个分支名称。这个函数可以包含任意业务规则，比如：当工具执行次数超过阈值时强制结束，当某个关键工具失败时进入人工审核节点。

5.4 组装图结构

# 文件路径：app/graph.py from langgraph.graph import StateGraph, START, END from app.state import AgentState from app.nodes import llm_node, tool_node from app.router import should_continue builder = StateGraph(AgentState) # 添加节点 builder.add_node("llm", llm_node) builder.add_node("tool", tool_node) # 设置入口 builder.add_edge(START, "llm") # 条件路由：llm 节点执行后决定下一个节点 builder.add_conditional_edges( "llm", should_continue, { "tool": "tool", "end": END, } ) # 工具节点执行完成后回到 llm，让模型看到工具结果后继续推理 builder.add_edge("tool", "llm") # 编译成可执行的图 agent_graph = builder.compile()

这段代码展示了一个经典的 Agent 循环：START -> llm -> (条件路由) -> tool -> llm -> ... -> END。模型每次回答都会先经过条件路由判断，如果返回工具调用就执行工具并把结果喂回去，直到模型给出最终答案为止。

调用方式如下：

from app.graph import agent_graph result = agent_graph.invoke( {"messages": [{"role": "user", "content": "请计算 12 和 5 的和，再乘以 3"}], "tool_attempts": 0} ) print(result["messages"][-1].content)

执行过程可以通过result["messages"]完整复盘：第一步模型调用add(12, 5)，返回 17；第二步模型调用multiply(17, 3)，返回 51；第三步模型给出最终答案“结果是 51”。

5.5 循环检测与子图

LangGraph 图结构本身允许环的存在，但这也会带来失控风险。实际工程中，一般会在路由函数里加上循环次数限制：

# 在 router.py 中加入次数控制 def should_continue_with_limit(state: AgentState) -> str: last_message = state["messages"][-1] # 超过 5 次工具调用，强制结束，避免无限循环和费用失控 if state.get("tool_attempts", 0) >= 5: return "end" if last_message.tool_calls: return "tool" return "end"

子图则是另一种常见需求。当 Agent 的业务变复杂，比如一个客服 Agent 内部还包含“订单查询 Agent”“退款流程 Agent”，可以把内部流程封装成子图，作为一个节点挂进主图。在 LangGraph 中，子图的编译结果可以直接作为父图的节点使用：

# 子图封装为节点示例 order_subgraph = order_builder.compile() chat_builder.add_node("order_query", order_subgraph)

这种分层设计能有效控制代码复杂度，让主图只承担业务路由，子图承担细化执行。团队协作时，不同子图可以由不同成员独立开发和测试。

6. TextToSQL Agent 实战：从自然语言到可执行 SQL

TextToSQL 是 Agent 在企业场景中落地价值很高的方向之一。业务人员不写 SQL，用自然语言提问：“上个月华东区销售额排名前五的产品是什么”，系统自动生成 SQL、查询数据库、返回答案。听起来很美好，但真实项目里有几个绕不开的坑。

6.1 难点拆解

第一是Schema 上下文。数据库表可能几十张，每张表几十个字段，全塞进 Prompt 会让模型失去注意力，也浪费 Token。合理的做法是动态选择相关表，只把相关表结构发给模型。

第二是SQL 正确性。模型生成的 SQL 可能语法正确但逻辑错误，比如 join 条件写错、聚合函数用错。所以不能只让模型生成 SQL 就结束，必须实际执行或至少做语法校验。

第三是安全性。TextToSQL 涉及数据库访问，必须约束为只读连接、限制可访问的表和列、设置查询超时，防止模型生成危险操作。

6.2 架构设计

用户输入 -> 1. 选择相关表（根据用户问题从表清单中筛选） -> 2. 构建 Schema Prompt -> 3. LLM 生成 SQL -> 4. SQL 校验与执行（只读） -> 5. 结果格式化 -> 6. 返回给用户

用 LangGraph 实现，可以设计四个节点：select_tables、generate_sql、execute_sql、format_answer，并在generate_sql之后加入一个条件路由：SQL 校验失败时回到生成节点重试，重试超过两次则进入人工处理。

6.3 核心代码示例

# 文件路径：app/text2sql_nodes.py import sqlite3 # 模拟数据库连接信息 DB_PATH = "./demo.db" # 表清单和字段说明，实际项目中可以从 information_schema 动态获取 TABLE_METADATA = { "products": { "description": "产品信息表", "columns": { "id": "主键", "name": "产品名称", "category": "产品分类", "price": "单价（元）", } }, "orders": { "description": "订单表", "columns": { "id": "订单ID", "product_id": "产品ID，关联 products.id", "quantity": "数量", "order_date": "下单日期", "region": "区域", } } } def select_tables_node(state): """根据用户问题选择相关表，简化示例：直接返回所有表""" user_question = state["messages"][-1].content # 实际项目中可以用 embedding 或关键词匹配筛选相关表 related_tables = list(TABLE_METADATA.keys()) return {"related_tables": related_tables} def generate_sql_node(state): """调用模型生成 SQL""" from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage user_question = state["messages"][-1].content related_tables = state["related_tables"] schema_text = "" for table in related_tables: meta = TABLE_METADATA[table] schema_text += f"表 {table}（{meta['description']}）字段：\n" for col, desc in meta["columns"].items(): schema_text += f" - {col}: {desc}\n" llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) prompt = f"""你是一个专业的 SQL 工程师。根据用户问题生成 SQLite 查询 SQL。 可用表结构： {schema_text} 要求： 1. 只做 SELECT 查询，不能生成 INSERT/UPDATE/DELETE 等语句 2. 只使用上述表结构中的表和字段 3. 如果有时间条件，使用订单表的 order_date 字段 4. 结果只输出 SQL 语句本身，不要额外解释 用户问题：{user_question} """ response = llm.invoke([HumanMessage(content=prompt)]) sql = response.content.strip() # 简单提取 SQL，去掉可能出现的 markdown 代码块标记 if sql.startswith("```"): sql = sql.split("```")[1] if sql.startswith("sql"): sql = sql[2:] sql = sql.strip() return {"sql": sql, "sql_attempts": state.get("sql_attempts", 0) + 1} def execute_sql_node(state): """执行 SQL（只读模式）并返回结果""" import sqlite3 sql = state["sql"] # 生产环境请使用只读账号，这里演示打开只读连接 conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) cursor = conn.execute(sql) # 只取前 20 行，防止返回数据过大 columns = [desc[0] for desc in cursor.description] rows = cursor.fetchmany(20) conn.close() return {"query_result": {"columns": columns, "rows": rows}} def validate_sql_node(state): """校验 SQL 是否安全""" sql = state["sql"].strip().lower() # 黑名单机制：强制禁止非 SELECT 语句 forbidden_keywords = ["insert", "update", "delete", "drop", "alter", "create"] for kw in forbidden_keywords: if sql.startswith(kw) or f" {kw} " in sql: return "invalid" return "valid"

6.4 组装 TextToSQL 图

# 文件路径：app/text2sql_graph.py from langgraph.graph import StateGraph, START, END from app.text2sql_nodes import ( select_tables_node, generate_sql_node, validate_sql_node, execute_sql_node, ) from app.state import AgentState builder = StateGraph(AgentState) builder.add_node("select_tables", select_tables_node) builder.add_node("generate_sql", generate_sql_node) builder.add_node("validate_sql", validate_sql_node) builder.add_node("execute_sql", execute_sql_node) builder.add_edge(START, "select_tables") builder.add_edge("select_tables", "generate_sql") builder.add_edge("generate_sql", "validate_sql") # SQL 校验失败时重新生成，最多重试 2 次 def route_after_validate(state): if state.get("sql_attempts", 0) >= 2: return "execute_sql" # 实际项目中可改为进入人工处理节点 if state["validation_result"] == "invalid": return "generate_sql" return "execute_sql" builder.add_conditional_edges( "validate_sql", route_after_validate, { "generate_sql": "generate_sql", "execute_sql": "execute_sql", } ) builder.add_edge("execute_sql", END) text2sql_graph = builder.compile()

这里的重点是validate_sql之后的条件路由，它实现了“生成 SQL -> 校验 -> 不合格就重新生成”的闭环。同时也体现了 LangGraph 的价值：模型负责生成和修正，规则负责兜底和拦截。

6.5 安全提示

TextToSQL 的安全性再怎么强调都不过分。生产环境至少要做到：

• 使用独立数据库账号，只授予 SELECT 权限。
• 配置查询超时时间，比如单条 SQL 超过 10 秒直接终止。
• 敏感表、敏感字段做脱敏或禁止访问。
• SQL 关键字黑名单只是最低要求，更可靠的是用 SQL Parser 解析出 AST，确认只包含查询语句。
• 所有生成过的 SQL 都写入审计日志，便于事后追踪。

7. 可观测部署：Agent 上生产前必须补的运维课

Agent 比普通 API 服务更难运维，因为一次用户请求会触发多次模型调用、工具调用和路由决策，任何一个环节出问题都可能导致最终回答质量下降。如果没有 Trace，排查问题几乎等于大海捞针。

7.1 可观测的三个层次

企业级 Agent 可观测性至少要覆盖三层：

• 服务层：请求量、响应时间、错误率、Token 用量，这是传统监控就能覆盖的。
• 执行层：一次 Agent 请求经过了哪些节点、每个节点耗时多少、路由到了哪个分支，这就是 Trace。
• 模型层：每次 LLM 调用的 Prompt 是什么、响应是什么、延迟多少、消耗了多少 Token。

本地开发时，最简单的可观测配置是开启 LangSmith。LangSmith 是 LangChain 生态中的可观测平台，可以自动捕获每次 Agent 运行的完整 Trace，包括节点调用顺序、工具输入输出、模型耗时和 Token 统计。

# 在代码中启用 LangSmith import os from dotenv import load_dotenv load_dotenv() # 确保 .env 中已经配置： # LANGCHAIN_TRACING_V2=true # LANGCHAIN_API_KEY=lsv2-xxxx # LANGCHAIN_PROJECT=enterprise-agent-demo from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 之后所有 LangChain 调用和 LangGraph 执行都会自动上报 Trace

如果公司内网环境不允许使用云端服务，也可以考虑私有化部署 LangSmith，或者直接把执行日志写成结构化 JSON。

7.2 结构化日志：自己的兜底方案

不是所有团队都愿意把数据上报到第三方平台，更通用的是在代码里埋点，输出结构化日志。LangGraph 节点本身就是普通 Python 函数，所以非常容易在节点入口和出口打印关键信息：

# 文件路径：app/observability.py import json import time from datetime import datetime def log_node_start(node_name: str, state: dict): print(json.dumps({ "event": "node_start", "node": node_name, "ts": datetime.utcnow().isoformat(), "message_count": len(state.get("messages", [])), }, ensure_ascii=False)) def log_node_end(node_name: str, state: dict, duration_ms: float, extra: dict = None): payload = { "event": "node_end", "node": node_name, "ts": datetime.utcnow().isoformat(), "duration_ms": round(duration_ms, 2), } if extra: payload.update(extra) print(json.dumps(payload, ensure_ascii=False))

在实际项目中，可以把这些日志发送到 ELK、Loki 或 ClickHouse 等日志平台，再配合 Grafana 展示。相比直接看 LangSmith，结构化作法更可控，适合对数据敏感的企业环境。

7.3 部署形态与成本控制

LangGraph 应用本身只是 Python 服务，可以按普通后端服务方式部署。LangGraph 平台也提供了一键部署、任务队列、持久化会话等能力，但如果你已经有一套成熟的微服务基础设施，也可以直接把graph编译结果封装成 HTTP API，跑在 Kubernetes 上。

这里需要提醒一个成本问题：Agent 的 Token 消耗比普通 RAG 问答高得多，因为一次任务可能要经历“模型-工具-模型-工具”多轮调用。生产环境必须设置预算上限，比如：单次请求最大工具调用次数、单次请求最大 Token 数、单用户每日配额。LangGraph 的条件路由和recursion_limit就是最直接的兜底工具。

8. 常见问题与排查思路

以下是 LangChain + LangGraph 开发中最高频的问题，按经验整理了排查路径：

除了表格里的问题，还有一个非常隐蔽的坑需要单独说：不要在生产代码里把大模型请求放到用户请求的同步链路上。LangGraph 本身支持异步执行和持久化，如果你的 Agent 任务比较耗时，应该设计成“提交任务 -> 后台执行 -> 通过回调或轮询获取结果”的异步模式，而不是让用户一直等待 HTTP 响应。

9. 最佳实践与工程建议

结合前文的项目经验，这里整理一份企业级 Agent 的工程建议清单，按代码开发、流程设计、运维部署三个维度展开。

代码与状态设计

State 结构不要在开发后期才定义。它是 LangGraph 应用的“接口协议”，建议在项目一开始就和团队成员评审。状态里只放需要跨节点共享的数据，临时变量不要塞进 State，否则会污染 Trace 和调试体验。消息合并器add_messages很好用，但请注意它只适合消息列表字段，其他业务字段还是要显式控制合并逻辑。

所有节点函数建议保持“纯函数”风格：输入 State，返回新的字段增量，不修改外部全局变量。这样每个节点都可以单独测试，也方便 mock。

流程设计

企业级 Agent 不鼓励把所有业务判断都交给模型。通用原则是：需要灵活推理的地方用模型，需要确定性结果的地方用规则。比如 SQL 生成让模型做，SQL 校验就交给解析器；工具选择让模型做，工具输入参数校验就交给 JSON Schema。

条件路由函数是业务逻辑最容易失控的地方，建议为每个分支编写单元测试。比如“工具执行失败时是否走了降级分支”“超过重试次数后是否正确进入人工处理”。这些测试不需要真实调用大模型，可以用 mock 数据验证路由逻辑。

运维与安全

Agent 上线前，建议建立最小回归集：准备 20 条典型业务问题，每次模型版本或提示词变更后自动跑一遍，记录回答质量和执行路径。这一步能显著降低“改了一个 Prompt，另外九个场景全部受影响”的风险。

数据库类 Agent 必须遵循最小权限原则。TextToSQL 用只读账号，限制可访问的表和列，配置超时和返回行数上限，所有 SQL 记录审计日志。Agent 触发外部写操作时，必须经过人工确认或额外的二次验证机制。

后续学习方向

如果这篇内容你已经完全消化，下一步可以沿着三个方向深入：

1. 学习 LangGraph 的持久化能力，实现多轮会话状态恢复和长任务异步执行。
2. 为你的 Agent 接入评估体系，用 LLM 作为裁判自动评估回答质量，建立 Prompt 回归测试。
3. 把工作流从单 Agent 扩展到多 Agent 协作，设计主管 Agent 和多个专用子 Agent 之间的任务分发与结果汇总。

最后留一个建议：不要一上来就追求复杂的图结构。先从 V1 骨架跑通，用 StateGraph 替换循环，再逐步增加条件路由和子图。Agent 复杂度的每一次上升，都应该有明确的业务收益来支撑。工程上真正重要的，不是把图画得多复杂，而是让每条路径都可解释、可控制、可回滚。把这套基本功打扎实，LangChain + LangGraph 会成为你在企业级 AI 应用落地时非常趁手的工具组合。