前言
在 LLM 应用开发中，如何构建一个可靠、可扩展、可维护的 Agent 系统，是每个 AI 工程师必须面对的挑战。LangGraph 作为 LangChain 生态中的状态机编排框架，提供了一套优雅的解决方案。
本文将从零开始，带你构建一个生产级 LangGraph Agent，涵盖：
核心概念与架构设计
模块化项目结构
多工具调用与状态管理
持久化与人工干预
错误处理与可观测性
测试、部署与监控
全文约 8000 字，建议收藏阅读。
一、为什么选择 LangGraph？
1.1 传统 Agent 的痛点
在 LangGraph 出现之前，构建 Agent 通常面临以下问题：
┌─────────────────────────────────────────────────────────────────┐
│ 传统 Agent 架构的问题 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 问题 1: 状态管理混乱 │
│ ┌────────────────────────────────────────────────────────┐ │
│ │ 用户输入 → LLM → 工具调用 → LLM → 回复 │ │
│ │ │ │
│ │ 问题：状态散落在各处，难以追踪和调试 │ │
│ └────────────────────────────────────────────────────────┘ │
│ │
│ 问题 2: 流程控制困难 │
│ - 多轮对话的上下文管理复杂 │
│ - 条件分支逻辑难以维护 │
│ - 循环调用容易失控 │
│ │
│ 问题 3: 缺乏可观测性 │
│ - 无法追踪 Agent 的决策过程 │
│ - 调试困难，出错后难以定位 │
│ - 无法回溯历史状态 │
│ │
│ 问题 4: 扩展性差 │
│ - 添加新工具需要修改核心逻辑 │
│ - 难以实现人工干预 │
│ - 无法支持复杂的多 Agent 协作 │
│ │
└─────────────────────────────────────────────────────────────────┘1.2 LangGraph 的核心优势
✅ 显式状态管理
TypedDict 定义状态结构
状态在节点间传递，清晰可追踪
支持状态快照与回溯
✅ 图结构编排
节点（Node）：执行单元
边（Edge）：流程控制
条件边：动态路由
✅ 内置持久化
MemorySaver：内存存储
SqliteSaver：数据库存储
支持断点续传
✅ 人工干预（Human-in-the-loop）
interrupt 节点暂停执行
等待人工审核后继续
适用于高风险操作
✅ 流式输出
支持 token 级流式
支持事件流
实时反馈用户
二、核心概念详解
2.1 状态（State）
状态是 LangGraph 的核心，所有节点共享同一个状态对象。
from typing import Annotated, List
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
"""
Agent 状态定义
messages: 对话历史（自动合并）
error_count: 错误计数
step: 当前步骤
user_id: 用户 ID
session_id: 会话 ID
"""
messages: Annotated[List[BaseMessage], add_messages]
error_count: int
step: str
user_id: str
session_id: str
# 可扩展更多字段
intermediate_results: dict
retry_count: int关键点：
Annotated[List[BaseMessage], add_messages]：自动合并消息，避免重复状态在节点间传递，每个节点返回部分更新
支持嵌套结构和复杂类型
2.2 节点（Node）
节点是执行单元，接收状态并返回状态更新。
def agent_node(state: AgentState) -> dict:
"""
Agent 节点：调用 LLM 决策
Args:
state: 当前状态
Returns:
状态更新字典
"""
messages = state["messages"]
model = get_model()
try:
response = model.invoke(messages)
return {
"messages": [response],
"step": "agent_decision"
}
except Exception as e:
return {
"messages": [AIMessage(content=f"错误：{e}")],
"error_count": state.get("error_count", 0) + 1,
"step": "error"
}节点类型：
普通节点：执行业务逻辑
工具节点：调用外部工具（
ToolNode）条件节点：根据状态决定下一步
2.3 边（Edge）
边定义节点间的流转关系。
# 1. 普通边：固定流转
graph.add_edge("node_a", "node_b")
# 2. 条件边：动态路由
graph.add_conditional_edges(
"agent",
should_continue, # 路由函数
{
"tools": "tools", # 调用工具
"human": "human", # 人工干预
END: END # 结束
}
)
# 3. 入口点
graph.set_entry_point("agent")2.4 执行模式
# 1. 同步执行
result = app.invoke(initial_state)
# 2. 流式执行（事件流）
for event in app.stream(initial_state, stream_mode="values"):
print(event)
# 3. 流式执行（更新流）
for chunk in app.stream(initial_state, stream_mode="updates"):
print(chunk)
# 4. 异步执行
result = await app.ainvoke(initial_state)三、项目结构设计
3.1 生产级项目结构
langgraph_agent/
├── .env # 环境变量
├── .env.example # 环境变量示例
├── .gitignore
├── README.md
├── requirements.txt
├── setup.py
│
├── config/ # 配置文件
│ ├── __init__.py
│ ├── settings.py # 配置管理
│ └── logging.yaml # 日志配置
│
├── src/
│ ├── __init__.py
│ │
│ ├── core/ # 核心模块
│ │ ├── __init__.py
│ │ ├── state.py # 状态定义
│ │ ├── graph.py # 图构建
│ │ └── nodes.py # 节点函数
│ │
│ ├── tools/ # 工具模块
│ │ ├── __init__.py
│ │ ├── weather.py # 天气工具
│ │ ├── calculator.py # 计算器工具
│ │ └── time_tool.py # 时间工具
│ │
│ ├── models/ # 模型封装
│ │ ├── __init__.py
│ │ └── llm.py # LLM 工厂
│ │
│ ├── utils/ # 工具函数
│ │ ├── __init__.py
│ │ ├── logger.py # 日志工具
│ │ └── validators.py # 验证器
│ │
│ └── api/ # API 接口
│ ├── __init__.py
│ ├── app.py # FastAPI 应用
│ └── routes.py # 路由定义
│
├── tests/ # 测试
│ ├── __init__.py
│ ├── test_tools.py
│ ├── test_graph.py
│ └── test_api.py
│
├── data/ # 数据存储
│ ├── checkpoints/ # 状态检查点
│ └── logs/ # 日志文件
│
└── scripts/ # 脚本
├── run_agent.py # 运行脚本
└── benchmark.py # 性能测试3.2 配置管理
config/settings.py
#!/usr/bin/env python3
"""
配置管理模块
支持环境变量、配置文件、默认值三级配置
"""
import os
from typing import Optional
from pydantic import BaseSettings, Field
from dotenv import load_dotenv
load_dotenv()
class Settings(BaseSettings):
"""应用配置"""
# LLM 配置
openai_api_key: str = Field(..., env="OPENAI_API_KEY")
openai_model: str = Field("gpt-4", env="OPENAI_MODEL")
openai_temperature: float = Field(0.0, env="OPENAI_TEMPERATURE")
openai_timeout: int = Field(30, env="OPENAI_TIMEOUT")
openai_max_retries: int = Field(3, env="OPENAI_MAX_RETRIES")
# 应用配置
app_name: str = Field("LangGraph Agent", env="APP_NAME")
app_version: str = Field("1.0.0", env="APP_VERSION")
debug: bool = Field(False, env="DEBUG")
log_level: str = Field("INFO", env="LOG_LEVEL")
# 持久化配置
checkpoint_dir: str = Field("./data/checkpoints", env="CHECKPOINT_DIR")
use_sqlite: bool = Field(True, env="USE_SQLITE")
sqlite_path: str = Field("./data/agent.db", env="SQLITE_PATH")
# API 配置
api_host: str = Field("0.0.0.0", env="API_HOST")
api_port: int = Field(8000, env="API_PORT")
api_workers: int = Field(4, env="API_WORKERS")
# 限流配置
rate_limit_enabled: bool = Field(True, env="RATE_LIMIT_ENABLED")
rate_limit_requests: int = Field(100, env="RATE_LIMIT_REQUESTS")
rate_limit_period: int = Field(60, env="RATE_LIMIT_PERIOD")
# 监控配置
enable_tracing: bool = Field(False, env="ENABLE_TRACING")
langsmith_api_key: Optional[str] = Field(None, env="LANGSMITH_API_KEY")
langsmith_project: str = Field("langgraph-agent", env="LANGSMITH_PROJECT")
class Config:
env_file = ".env"
case_sensitive = False
# 全局配置实例
settings = Settings()3.3 状态定义
src/core/state.py
#!/usr/bin/env python3
"""
状态定义模块
"""
from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from datetime import datetime
class AgentState(TypedDict):
"""
Agent 状态结构
Attributes:
messages: 对话消息列表（自动合并）
user_id: 用户 ID
session_id: 会话 ID
step: 当前执行步骤
error_count: 错误计数
retry_count: 重试次数
intermediate_results: 中间结果存储
metadata: 元数据
created_at: 创建时间
updated_at: 更新时间
"""
messages: Annotated[List[BaseMessage], add_messages]
user_id: str
session_id: str
step: str
error_count: int
retry_count: int
intermediate_results: Dict[str, Any]
metadata: Dict[str, Any]
created_at: str
updated_at: str
def create_initial_state(
user_input: str,
user_id: str = "default",
session_id: Optional[str] = None
) -> AgentState:
"""
创建初始状态
Args:
user_input: 用户输入
user_id: 用户 ID
session_id: 会话 ID
Returns:
初始化的 AgentState
"""
from langchain_core.messages import HumanMessage
import uuid
if session_id is None:
session_id = str(uuid.uuid4())
now = datetime.now().isoformat()
return AgentState(
messages=[HumanMessage(content=user_input)],
user_id=user_id,
session_id=session_id,
step="start",
error_count=0,
retry_count=0,
intermediate_results={},
metadata={},
created_at=now,
updated_at=now
)四、完整代码实现
4.1 工具定义（模块化）
src/tools/weather.py
#!/usr/bin/env python3
"""天气查询工具"""
import logging
from langchain_core.tools import tool
logger = logging.getLogger(__name__)
@tool
def get_weather(city: str) -> str:
"""
获取指定城市的天气（模拟）。
Args:
city: 城市名称，例如"北京"
Returns:
天气信息字符串
"""
logger.info(f"调用工具 get_weather，参数: city={city}")
if not city or not city.strip():
return "错误：城市名不能为空"
# 模拟天气数据库
weather_db = {
"北京": "晴朗，25°C，空气质量良好",
"上海": "多云，28°C，湿度 65%",
"深圳": "雷阵雨，30°C，注意防雷",
"广州": "晴转多云，32°C，紫外线强",
"杭州": "小雨，22°C，适合出行"
}
result = weather_db.get(
city.strip(),
f"暂时无法获取{city}的天气，请稍后重试"
)
logger.debug(f"get_weather 返回: {result}")
return resultsrc/tools/calculator.py
#!/usr/bin/env python3
"""计算器工具"""
import logging
import ast
import operator
from langchain_core.tools import tool
logger = logging.getLogger(__name__)
# 安全的运算符白名单
SAFE_OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
}
def safe_eval(expression: str) -> float:
"""
安全的表达式求值
Args:
expression: 数学表达式
Returns:
计算结果
"""
try:
node = ast.parse(expression, mode='eval')
def _eval(node):
if isinstance(node, ast.Expression):
return _eval(node.body)
elif isinstance(node, ast.Num):
return node.n
elif isinstance(node, ast.BinOp):
op = SAFE_OPERATORS.get(type(node.op))
if op is None:
raise ValueError(f"不支持的运算符: {type(node.op).__name__}")
return op(_eval(node.left), _eval(node.right))
elif isinstance(node, ast.UnaryOp):
op = SAFE_OPERATORS.get(type(node.op))
if op is None:
raise ValueError(f"不支持的运算符: {type(node.op).__name__}")
return op(_eval(node.operand))
else:
raise ValueError(f"不支持的节点类型: {type(node).__name__}")
return _eval(node)
except Exception as e:
raise ValueError(f"表达式解析失败: {str(e)}")
@tool
def calculate(expression: str) -> str:
"""
计算数学表达式的值（安全模式）。
Args:
expression: 数学表达式字符串，如"2+3*4"
Returns:
计算结果字符串
"""
logger.info(f"调用工具 calculate，参数: expression={expression}")
try:
result = safe_eval(expression)
return f"计算结果: {result}"
except Exception as e:
error_msg = f"计算错误: {str(e)}"
logger.error(error_msg)
return error_msgsrc/tools/init.py
"""工具模块"""
from .weather import get_weather
from .calculator import calculate
from .time_tool import get_current_time
# 导出所有工具
TOOLS = [get_weather, calculate, get_current_time]
__all__ = ["TOOLS", "get_weather", "calculate", "get_current_time"]4.2 图构建
src/core/graph.py
#!/usr/bin/env python3
"""
LangGraph 图构建模块
"""
import logging
from typing import Dict, Any
from langchain_core.messages import AIMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.memory import MemorySaver
from .state import AgentState
from .nodes import agent_node, error_handler_node
from ..tools import TOOLS
from config.settings import settings
logger = logging.getLogger(__name__)
def should_continue(state: AgentState) -> str:
"""
条件路由函数
Args:
state: 当前状态
Returns:
下一步节点名称
"""
# 检查错误次数
if state.get("error_count", 0) >= 3:
logger.warning("错误次数过多，终止执行")
return "error_handler"
# 使用内置的 tools_condition
return tools_condition(state)
def build_agent_graph(use_checkpointer: bool = True):
"""
构建 LangGraph 工作流
Args:
use_checkpointer: 是否启用持久化
Returns:
编译后的图
"""
logger.info("开始构建 Agent 图")
# 创建图
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(TOOLS))
graph.add_node("error_handler", error_handler_node)
# 设置入口
graph.set_entry_point("agent")
# 添加条件边
graph.add_conditional_edges(
"agent",
should_continue,
{
"tools": "tools",
"error_handler": "error_handler",
END: END
}
)
# 工具执行后回到 agent
graph.add_edge("tools", "agent")
# 错误处理后结束
graph.add_edge("error_handler", END)
# 配置持久化
checkpointer = None
if use_checkpointer:
if settings.use_sqlite:
logger.info(f"使用 SQLite 持久化: {settings.sqlite_path}")
checkpointer = SqliteSaver.from_conn_string(settings.sqlite_path)
else:
logger.info("使用内存持久化")
checkpointer = MemorySaver()
# 编译图
app = graph.compile(checkpointer=checkpointer)
logger.info("Agent 图编译完成")
return app4.3 节点实现
src/core/nodes.py
#!/usr/bin/env python3
"""
节点函数模块
"""
import logging
from typing import Dict, Any
from datetime import datetime
from langchain_core.messages import AIMessage
from .state import AgentState
from ..models.llm import get_model
logger = logging.getLogger(__name__)
def agent_node(state: AgentState) -> Dict[str, Any]:
"""
Agent 节点：调用模型决定下一步动作
Args:
state: 当前状态
Returns:
状态更新
"""
logger.info(f"--- Agent 节点开始 (step: {state.get('step', 'unknown')}) ---")
messages = state["messages"]
model = get_model()
try:
# 调用模型
response = model.invoke(messages)
logger.debug(f"模型响应: {response}")
return {
"messages": [response],
"step": "agent_decision",
"updated_at": datetime.now().isoformat()
}
except Exception as e:
error_msg = f"模型调用失败: {str(e)}"
logger.error(error_msg)
return {
"messages": [AIMessage(content=f"抱歉，我遇到技术问题：{error_msg}")],
"error_count": state.get("error_count", 0) + 1,
"step": "error",
"updated_at": datetime.now().isoformat()
}
def error_handler_node(state: AgentState) -> Dict[str, Any]:
"""
错误处理节点
Args:
state: 当前状态
Returns:
状态更新
"""
logger.error(f"进入错误处理节点，错误次数: {state.get('error_count', 0)}")
error_message = AIMessage(
content="抱歉，系统遇到多次错误，已终止执行。请稍后重试或联系管理员。"
)
return {
"messages": [error_message],
"step": "error_handled",
"updated_at": datetime.now().isoformat()
}五、高级特性
5.1 持久化与断点续传
使用 SQLite 持久化
#!/usr/bin/env python3
"""
持久化示例
"""
from langgraph.checkpoint.sqlite import SqliteSaver
from src.core.graph import build_agent_graph
from src.core.state import create_initial_state
def demo_persistence():
"""演示持久化功能"""
# 构建带持久化的图
app = build_agent_graph(use_checkpointer=True)
# 创建初始状态
config = {"configurable": {"thread_id": "user-123-session-1"}}
initial_state = create_initial_state(
user_input="北京天气怎么样？",
user_id="user-123",
session_id="session-1"
)
# 第一次执行
print("=== 第一次执行 ===")
result1 = app.invoke(initial_state, config=config)
print(f"回复: {result1['messages'][-1].content}\n")
# 继续对话（使用相同的 thread_id）
print("=== 继续对话 ===")
from langchain_core.messages import HumanMessage
# 获取当前状态
current_state = app.get_state(config)
# 添加新消息
current_state.values["messages"].append(
HumanMessage(content="那上海呢？")
)
# 继续执行
result2 = app.invoke(current_state.values, config=config)
print(f"回复: {result2['messages'][-1].content}\n")
# 查看历史状态
print("=== 历史状态 ===")
for state in app.get_state_history(config):
print(f"Step: {state.values.get('step')}")
print(f"Messages: {len(state.values.get('messages', []))}")
print(f"Updated: {state.values.get('updated_at')}")
print("-" * 50)
if __name__ == "__main__":
demo_persistence()5.2 人工干预（Human-in-the-loop）
#!/usr/bin/env python3
"""
人工干预示例
"""
from typing import Dict, Any
from langchain_core.messages import AIMessage
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from src.core.state import AgentState
def risky_operation_node(state: AgentState) -> Dict[str, Any]:
"""高风险操作节点"""
return {
"messages": [AIMessage(content="准备执行高风险操作，等待人工审核...")],
"step": "awaiting_approval"
}
def execute_operation_node(state: AgentState) -> Dict[str, Any]:
"""执行操作节点"""
return {
"messages": [AIMessage(content="操作已执行完成")],
"step": "completed"
}
def build_human_in_loop_graph():
"""构建带人工干预的图"""
graph = StateGraph(AgentState)
graph.add_node("risky_operation", risky_operation_node)
graph.add_node("execute", execute_operation_node)
graph.set_entry_point("risky_operation")
# 添加中断点
graph.add_edge("risky_operation", "execute")
graph.add_edge("execute", END)
# 编译时指定中断节点
app = graph.compile(
checkpointer=MemorySaver(),
interrupt_before=["execute"] # 在 execute 前暂停
)
return app
def demo_human_in_loop():
"""演示人工干预"""
app = build_human_in_loop_graph()
config = {"configurable": {"thread_id": "approval-demo"}}
from src.core.state import create_initial_state
initial_state = create_initial_state("执行删除操作")
# 第一次执行（会在 execute 前暂停）
print("=== 第一次执行（暂停等待审核）===")
result = app.invoke(initial_state, config=config)
print(f"状态: {result['step']}")
print(f"消息: {result['messages'][-1].content}\n")
# 模拟人工审核
print("=== 人工审核中... ===")
approval = input("是否批准执行？(y/n): ")
if approval.lower() == 'y':
# 继续执行
print("\n=== 继续执行 ===")
result = app.invoke(None, config=config)
print(f"状态: {result['step']}")
print(f"消息: {result['messages'][-1].content}")
else:
print("\n操作已取消")
if __name__ == "__main__":
demo_human_in_loop()5.3 并发与异步
#!/usr/bin/env python3
"""
并发执行示例
"""
import asyncio
from typing import List
from src.core.graph import build_agent_graph
from src.core.state import create_initial_state
async def process_query_async(query: str, user_id: str):
"""异步处理单个查询"""
app = build_agent_graph(use_checkpointer=False)
initial_state = create_initial_state(
user_input=query,
user_id=user_id
)
# 使用异步调用
result = await app.ainvoke(initial_state)
return result["messages"][-1].content
async def batch_process(queries: List[str]):
"""批量处理查询"""
tasks = [
process_query_async(query, f"user-{i}")
for i, query in enumerate(queries)
]
results = await asyncio.gather(*tasks)
return results
async def demo_async():
"""演示异步并发"""
queries = [
"北京天气怎么样？",
"计算 100 + 200",
"现在几点了？",
"上海天气如何？",
"计算 50 * 3"
]
print(f"开始处理 {len(queries)} 个查询...")
import time
start = time.time()
results = await batch_process(queries)
elapsed = time.time() - start
print(f"\n处理完成，耗时: {elapsed:.2f}s")
print(f"平均每个查询: {elapsed/len(queries):.2f}s\n")
for i, (query, result) in enumerate(zip(queries, results)):
print(f"{i+1}. 查询: {query}")
print(f" 回复: {result}\n")
if __name__ == "__main__":
asyncio.run(demo_async())六、API 服务化
6.1 FastAPI 集成
src/api/app.py
#!/usr/bin/env python3
"""
FastAPI 应用
"""
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import Optional, List
import logging
import asyncio
import json
from src.core.graph import build_agent_graph
from src.core.state import create_initial_state
from config.settings import settings
logger = logging.getLogger(__name__)
# 创建应用
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="LangGraph Agent API"
)
# CORS 配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 全局图实例
agent_graph = build_agent_graph(use_checkpointer=True)
# 请求模型
class ChatRequest(BaseModel):
"""对话请求"""
message: str
user_id: str = "default"
session_id: Optional[str] = None
stream: bool = False
class ChatResponse(BaseModel):
"""对话响应"""
message: str
session_id: str
step: str
error_count: int
@app.get("/")
async def root():
"""根路径"""
return {
"name": settings.app_name,
"version": settings.app_version,
"status": "running"
}
@app.get("/health")
async def health():
"""健康检查"""
return {"status": "healthy"}
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""
对话接口（非流式）
"""
try:
# 创建初始状态
initial_state = create_initial_state(
user_input=request.message,
user_id=request.user_id,
session_id=request.session_id
)
# 配置
config = {
"configurable": {
"thread_id": f"{request.user_id}-{initial_state['session_id']}"
}
}
# 执行
result = await agent_graph.ainvoke(initial_state, config=config)
# 提取响应
last_message = result["messages"][-1]
return ChatResponse(
message=last_message.content,
session_id=result["session_id"],
step=result["step"],
error_count=result["error_count"]
)
except Exception as e:
logger.error(f"对话处理失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
"""
对话接口（流式）
"""
async def generate():
try:
# 创建初始状态
initial_state = create_initial_state(
user_input=request.message,
user_id=request.user_id,
session_id=request.session_id
)
# 配置
config = {
"configurable": {
"thread_id": f"{request.user_id}-{initial_state['session_id']}"
}
}
# 流式执行
async for event in agent_graph.astream(initial_state, config=config):
# 提取最后一条消息
if "messages" in event:
last_message = event["messages"][-1]
if hasattr(last_message, "content") and last_message.content:
chunk = {
"type": "message",
"content": last_message.content,
"step": event.get("step", "unknown")
}
yield f"data: {json.dumps(chunk)}\n\n"
# 发送结束标记
yield "data: [DONE]\n\n"
except Exception as e:
logger.error(f"流式处理失败: {e}")
error_chunk = {
"type": "error",
"content": str(e)
}
yield f"data: {json.dumps(error_chunk)}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream"
)
@app.get("/sessions/{session_id}/history")
async def get_session_history(session_id: str, user_id: str = "default"):
"""
获取会话历史
"""
try:
config = {
"configurable": {
"thread_id": f"{user_id}-{session_id}"
}
}
# 获取状态历史
history = []
for state in agent_graph.get_state_history(config):
history.append({
"step": state.values.get("step"),
"messages": [
{
"type": msg.__class__.__name__,
"content": msg.content
}
for msg in state.values.get("messages", [])
],
"updated_at": state.values.get("updated_at")
})
return {"session_id": session_id, "history": history}
except Exception as e:
logger.error(f"获取历史失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(
app,
host=settings.api_host,
port=settings.api_port,
workers=settings.api_workers
)6.2 客户端示例
#!/usr/bin/env python3
"""
API 客户端示例
"""
import requests
import json
class AgentClient:
"""Agent API 客户端"""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url.rstrip("/")
def chat(self, message: str, user_id: str = "default", session_id: str = None):
"""同步对话"""
response = requests.post(
f"{self.base_url}/chat",
json={
"message": message,
"user_id": user_id,
"session_id": session_id
}
)
response.raise_for_status()
return response.json()
def chat_stream(self, message: str, user_id: str = "default"):
"""流式对话"""
response = requests.post(
f"{self.base_url}/chat/stream",
json={
"message": message,
"user_id": user_id,
"stream": True
},
stream=True
)
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = line[6:]
if data == "[DONE]":
break
try:
chunk = json.loads(data)
yield chunk
except json.JSONDecodeError:
continue
def get_history(self, session_id: str, user_id: str = "default"):
"""获取会话历史"""
response = requests.get(
f"{self.base_url}/sessions/{session_id}/history",
params={"user_id": user_id}
)
response.raise_for_status()
return response.json()
# 使用示例
if __name__ == "__main__":
client = AgentClient()
# 同步对话
print("=== 同步对话 ===")
result = client.chat("北京天气怎么样？")
print(f"回复: {result['message']}\n")
# 流式对话
print("=== 流式对话 ===")
for chunk in client.chat_stream("计算 100 + 200"):
if chunk["type"] == "message":
print(chunk["content"], end="", flush=True)
print("\n")七、测试
7.1 单元测试
tests/test_tools.py
#!/usr/bin/env python3
"""
工具测试
"""
import pytest
from src.tools import get_weather, calculate, get_current_time
def test_get_weather():
"""测试天气工具"""
# 正常情况
result = get_weather.invoke({"city": "北京"})
assert "北京" in result or "晴" in result or "°C" in result
# 空城市
result = get_weather.invoke({"city": ""})
assert "错误" in result
# 未知城市
result = get_weather.invoke({"city": "火星"})
assert "无法获取" in result or "火星" in result
def test_calculate():
"""测试计算器工具"""
# 正常计算
result = calculate.invoke({"expression": "2+3"})
assert "5" in result
result = calculate.invoke({"expression": "10*5"})
assert "50" in result
# 除零错误
result = calculate.invoke({"expression": "10/0"})
assert "错误" in result
# 非法字符
result = calculate.invoke({"expression": "import os"})
assert "错误" in result
def test_get_current_time():
"""测试时间工具"""
result = get_current_time.invoke({})
assert "当前时间" in result
# 自定义格式
result = get_current_time.invoke({"format": "%Y-%m-%d"})
assert "-" in result7.2 集成测试
tests/test_graph.py
#!/usr/bin/env python3
"""
图执行测试
"""
import pytest
from src.core.graph import build_agent_graph
from src.core.state import create_initial_state
@pytest.fixture
def agent_app():
"""创建测试用图"""
return build_agent_graph(use_checkpointer=False)
def test_simple_query(agent_app):
"""测试简单查询"""
initial_state = create_initial_state("你好")
result = agent_app.invoke(initial_state)
assert len(result["messages"]) > 1
assert result["error_count"] == 0
def test_tool_call(agent_app):
"""测试工具调用"""
initial_state = create_initial_state("北京天气怎么样？")
result = agent_app.invoke(initial_state)
# 检查是否调用了工具
messages = result["messages"]
has_tool_call = any(
hasattr(msg, "tool_calls") and msg.tool_calls
for msg in messages
)
assert has_tool_call or "北京" in messages[-1].content
def test_error_handling(agent_app):
"""测试错误处理"""
# 模拟错误场景
initial_state = create_initial_state("计算 abc + def")
result = agent_app.invoke(initial_state)
# 应该有错误提示
last_message = result["messages"][-1].content
assert "错误" in last_message or "无法" in last_message
@pytest.mark.asyncio
async def test_async_execution(agent_app):
"""测试异步执行"""
initial_state = create_initial_state("现在几点了？")
result = await agent_app.ainvoke(initial_state)
assert len(result["messages"]) > 1
assert "时间" in result["messages"][-1].content7.3 API 测试
tests/test_api.py
#!/usr/bin/env python3
"""
API 测试
"""
import pytest
from fastapi.testclient import TestClient
from src.api.app import app
client = TestClient(app)
def test_root():
"""测试根路径"""
response = client.get("/")
assert response.status_code == 200
assert "name" in response.json()
def test_health():
"""测试健康检查"""
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
def test_chat():
"""测试对话接口"""
response = client.post(
"/chat",
json={
"message": "你好",
"user_id": "test-user"
}
)
assert response.status_code == 200
data = response.json()
assert "message" in data
assert "session_id" in data
def test_chat_with_tool():
"""测试工具调用"""
response = client.post(
"/chat",
json={
"message": "北京天气怎么样？",
"user_id": "test-user"
}
)
assert response.status_code == 200
data = response.json()
assert "北京" in data["message"] or "天气" in data["message"]八、监控与可观测性
8.1 LangSmith 集成
#!/usr/bin/env python3
"""
LangSmith 追踪集成
"""
import os
from langsmith import Client
from config.settings import settings
def setup_langsmith():
"""配置 LangSmith 追踪"""
if settings.enable_tracing and settings.langsmith_api_key:
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = settings.langsmith_api_key
os.environ["LANGCHAIN_PROJECT"] = settings.langsmith_project
print(f"✅ LangSmith 追踪已启用")
print(f" 项目: {settings.langsmith_project}")
else:
print("⚠️ LangSmith 追踪未启用")
def get_langsmith_client() -> Client:
"""获取 LangSmith 客户端"""
return Client(api_key=settings.langsmith_api_key)
def query_traces(project_name: str = None, limit: int = 10):
"""查询追踪记录"""
client = get_langsmith_client()
project = project_name or settings.langsmith_project
runs = client.list_runs(
project_name=project,
limit=limit
)
for run in runs:
print(f"Run ID: {run.id}")
print(f"Name: {run.name}")
print(f"Status: {run.status}")
print(f"Duration: {run.end_time - run.start_time if run.end_time else 'N/A'}")
print("-" * 50)8.2 Prometheus 指标
#!/usr/bin/env python3
"""
Prometheus 指标导出
"""
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response
import time
# 定义指标
request_count = Counter(
'agent_requests_total',
'Total number of agent requests',
['user_id', 'status']
)
request_duration = Histogram(
'agent_request_duration_seconds',
'Agent request duration in seconds',
['user_id']
)
active_sessions = Gauge(
'agent_active_sessions',
'Number of active sessions'
)
tool_calls = Counter(
'agent_tool_calls_total',
'Total number of tool calls',
['tool_name', 'status']
)
error_count = Counter(
'agent_errors_total',
'Total number of errors',
['error_type']
)
class MetricsMiddleware:
"""指标中间件"""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
start_time = time.time()
# 执行请求
await self.app(scope, receive, send)
# 记录指标
duration = time.time() - start_time
path = scope["path"]
if path == "/chat":
request_duration.labels(user_id="default").observe(duration)
request_count.labels(user_id="default", status="success").inc()
# 添加到 FastAPI
from src.api.app import app
@app.get("/metrics")
async def metrics():
"""Prometheus 指标端点"""
return Response(
content=generate_latest(),
media_type="text/plain"
)九、部署
9.1 Docker 部署
Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 创建数据目录
RUN mkdir -p /app/data/checkpoints /app/data/logs
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["python", "-m", "uvicorn", "src.api.app:app", "--host", "0.0.0.0", "--port", "8000"]docker-compose.yml
version: '3.8'
services:
agent-api:
build: .
container_name: langgraph-agent
restart: unless-stopped
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- OPENAI_MODEL=gpt-4
- LOG_LEVEL=INFO
- USE_SQLITE=true
- SQLITE_PATH=/app/data/agent.db
volumes:
- ./data:/app/data
- ./logs:/app/logs
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 39.2 Kubernetes 部署
k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: langgraph-agent
labels:
app: langgraph-agent
spec:
replicas: 3
selector:
matchLabels:
app: langgraph-agent
template:
metadata:
labels:
app: langgraph-agent
spec:
containers:
- name: agent
image: langgraph-agent:latest
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: openai-secret
key: api-key
- name: OPENAI_MODEL
value: "gpt-4"
- name: USE_SQLITE
value: "true"
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "1Gi"
cpu: "1000m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: langgraph-agent
spec:
selector:
app: langgraph-agent
ports:
- protocol: TCP
port: 80
targetPort: 8000
type: LoadBalancer十、最佳实践与总结
10.1 架构设计原则
10.2 性能优化建议
1. 并发处理
# ❌ 错误：串行处理
for query in queries:
result = app.invoke(query)
# ✅ 正确：并发处理
import asyncio
results = await asyncio.gather(*[
app.ainvoke(query) for query in queries
])2. 缓存策略
from functools import lru_cache
@lru_cache(maxsize=1000)
def get_weather_cached(city: str) -> str:
"""带缓存的天气查询"""
return get_weather(city)3. 连接池
from langchain_openai import ChatOpenAI
# 复用模型实例
_model_instance = None
def get_model():
global _model_instance
if _model_instance is None:
_model_instance = ChatOpenAI(...)
return _model_instance10.3 安全性检查清单
API 密钥管理：使用环境变量，不硬编码
输入验证：验证所有用户输入
输出过滤：防止敏感信息泄露
速率限制：防止滥用
工具安全：限制工具权限（如 calculate 只允许数学运算）
日志脱敏：不记录敏感信息
HTTPS：生产环境必须使用 HTTPS
认证授权：API 需要认证
10.4 常见问题与解决方案
问题 1：状态过大导致性能下降
# 解决方案：定期清理历史消息
def trim_messages(state: AgentState) -> dict:
"""保留最近 10 条消息"""
messages = state["messages"]
if len(messages) > 10:
# 保留系统消息 + 最近 9 条
system_msgs = [m for m in messages if m.type == "system"]
recent_msgs = messages[-9:]
return {"messages": system_msgs + recent_msgs}
return {}问题 2：工具调用超时
# 解决方案：添加超时控制
import asyncio
async def call_tool_with_timeout(tool, args, timeout=10):
"""带超时的工具调用"""
try:
return await asyncio.wait_for(
tool.ainvoke(args),
timeout=timeout
)
except asyncio.TimeoutError:
return "工具调用超时"问题 3：循环调用失控
# 解决方案：添加最大步数限制
class AgentState(TypedDict):
messages: List[BaseMessage]
step_count: int # 新增步数计数
def should_continue(state: AgentState) -> str:
"""检查是否继续"""
if state.get("step_count", 0) >= 10:
return END # 超过 10 步强制结束
return tools_condition(state)10.5 生产环境检查清单
部署前
所有测试通过（单元测试、集成测试、API 测试）
性能测试达标（QPS、延迟、并发）
安全审计通过
日志配置正确
监控告警配置完成
文档完善（API 文档、运维文档）
部署后
健康检查正常
监控指标正常
日志输出正常
告警测试通过
灰度发布验证
回滚方案准备
十一、完整示例代码
11.1 简化版（单文件）
以下是一个生产级简化版，适合快速上手：
#!/usr/bin/env python3
"""
LangGraph 工程化 Demo（简化版）
- 多工具 Agent
- 配置管理
- 错误处理
- 流式输出
"""
import os
import logging
from datetime import datetime
from typing import Annotated, List, Dict, Any
from typing_extensions import TypedDict
from dotenv import load_dotenv
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph.message import add_messages
# 配置
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 工具定义
@tool
def get_weather(city: str) -> str:
"""获取城市天气"""
weather_db = {
"北京": "晴朗，25°C",
"上海": "多云，28°C",
"深圳": "雷阵雨，30°C"
}
return weather_db.get(city, f"无法获取{city}的天气")
@tool
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression, {"__builtins__": }, {})
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {e}"
@tool
def get_current_time() -> str:
"""获取当前时间"""
return f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
TOOLS = [get_weather, calculate, get_current_time]
# 状态定义
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
error_count: int
# 节点函数
def agent_node(state: AgentState) -> Dict[str, Any]:
"""Agent 节点"""
logger.info("--- Agent 节点 ---")
model = ChatOpenAI(model="gpt-4", temperature=0).bind_tools(TOOLS)
try:
response = model.invoke(state["messages"])
return {"messages": [response]}
except Exception as e:
logger.error(f"模型调用失败: {e}")
return {
"messages": [AIMessage(content=f"错误：{e}")],
"error_count": state.get("error_count", 0) + 1
}
# 构建图
def build_graph():
"""构建图"""
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(TOOLS))
graph.set_entry_point("agent")
graph.add_conditional_edges(
"agent",
tools_condition,
{"tools": "tools", END: END}
)
graph.add_edge("tools", "agent")
return graph.compile()
# 运行
def run_agent(user_input: str, stream: bool = False):
"""运行 Agent"""
app = build_graph()
initial_state = {
"messages": [HumanMessage(content=user_input)],
"error_count": 0
}
if stream:
print("🤖 流式响应:")
for event in app.stream(initial_state):
if "messages" in event:
msg = event["messages"][-1]
if hasattr(msg, "content") and msg.content:
print(msg.content, end="", flush=True)
print("\n")
else:
result = app.invoke(initial_state)
print(f"🤖 回复: {result['messages'][-1].content}\n")
# 测试
if __name__ == "__main__":
test_queries = [
"北京今天天气怎么样？",
"帮我计算 (12 + 35) * 2",
"现在几点了？"
]
for query in test_queries:
print(f"👤 用户: {query}")
run_agent(query)
print("-" * 50)11.2 运行与验证
1. 安装依赖
pip install langgraph langchain-openai python-dotenv2. 配置环境变量
# .env
OPENAI_API_KEY=sk-xxxxx3. 运行
python langgraph_demo.py预期输出
👤 用户: 北京今天天气怎么样？
INFO:__main__:--- Agent 节点 ---
🤖 回复: 北京今天的天气是晴朗，25°C。
--------------------------------------------------
👤 用户: 帮我计算 (12 + 35) * 2
INFO:__main__:--- Agent 节点 ---
🤖 回复: 计算结果: 94
--------------------------------------------------
👤 用户: 现在几点了？
INFO:__main__:--- Agent 节点 ---
🤖 回复: 当前时间: 2025-05-15 14:30:25
--------------------------------------------------十二、总结
本文从零开始，构建了一个生产级 LangGraph Agent 系统，涵盖了：
核心要点回顾
状态管理：使用 TypedDict 定义清晰的状态结构
图结构编排：节点 + 边 + 条件路由实现复杂流程
模块化设计：工具、节点、状态分离，易于维护
持久化：SQLite/Memory 实现多轮对话
人工干预：interrupt 节点实现 Human-in-the-loop
API 服务化：FastAPI 提供 REST 接口
监控可观测：LangSmith + Prometheus 全链路追踪
测试完善：单元测试 + 集成测试 + API 测试
部署方案：Docker + K8s 生产级部署
技术栈总结
最佳实践
进阶方向
多 Agent 协作：使用 LangGraph 的 subgraph 实现多 Agent 系统
RAG 集成：结合向量数据库实现知识增强
流式优化：实现 token 级流式输出
分布式部署：使用 Redis 作为共享存储
A/B 测试：不同 Prompt 策略的效果对比
参考资源
结语
LangGraph 为构建生产级 Agent 提供了强大的基础设施。通过本文的实践，你应该能够：
✅ 理解 LangGraph 的核心概念
✅ 构建模块化、可维护的 Agent 系统
✅ 实现持久化、人工干预等高级特性
✅ 部署到生产环境并进行监控
希望这篇文章能帮助你快速上手 LangGraph，构建出色的 AI Agent 应用！
如有问题，欢迎在评论区留言讨论。
本文示例代码已开源，欢迎 Star 和 Fork！
@tool
def get_weather(city: str) -> str:
"""
获取指定城市的天气（模拟）。
Args:
city: 城市名称，例如"北京"
Returns:
天气信息字符串
"""
logger.info(f"调用工具 get_weather，参数: city={city}")
# 模拟API调用，实际可替换为真实请求
if not city or not city.strip():
return "错误：城市名不能为空"
# 简单模拟
weather_db = {
"北京": "晴朗，25°C",
"上海": "多云，28°C",
"深圳": "雷阵雨，30°C"
}
result = weather_db.get(city.strip(), f"暂时无法获取{city}的天气，请稍后重试")
logger.debug(f"get_weather 返回: {result}")
return result
@tool
def calculate(expression: str) -> str:
"""
计算数学表达式的值（安全模式）。
Args:
expression: 数学表达式字符串，如"2+34"
Returns:
计算结果字符串
"""
logger.info(f"调用工具 calculate，参数: expression={expression}")
# 安全的表达式求值（只允许数字、运算符）
allowed_chars = set("0123456789+-/(). ")
if not all(c in allowed_chars for c in expression):
return "错误：表达式包含非法字符"
try:
# 注意：eval在受限环境下使用，生产建议用ast.literal_eval或math库
result = eval(expression, {"builtins": {}}, {})
return f"计算结果: {result}"
except Exception as e:
error_msg = f"计算错误: {str(e)}"
logger.error(error_msg)
return error_msg
@tool
def get_current_time(format: str = "%Y-%m-%d %H:%M:%S") -> str:
"""
获取当前日期时间。
Args:
format: 时间格式字符串，默认为'%Y-%m-%d %H:%M:%S'
Returns:
格式化后的当前时间字符串
"""
logger.info(f"调用工具 get_current_time，格式: {format}")
try:
now = datetime.now().strftime(format)
return f"当前时间: {now}"
except Exception as e:
error_msg = f"时间格式化错误: {str(e)}"
logger.error(error_msg)
return error_msg
工具列表
TOOLS = [get_weather, calculate, get_current_time]
---------------------------- 3. 状态定义 ----------------------------
class AgentState(TypedDict):
"""Agent状态结构"""
messages: Annotated[List[BaseMessage], add_messages]
# 可扩展其他字段，例如中间结果、错误计数等
error_count: int
step: str
---------------------------- 4. 模型与绑定 ----------------------------
def get_model():
"""工厂函数：创建并绑定工具的模型实例"""
llm = ChatOpenAI(
model=OPENAI_MODEL,
api_key=OPENAI_API_KEY,
temperature=0,
timeout=30,
max_retries=2
)
# 绑定工具
model_with_tools = llm.bind_tools(TOOLS)
return model_with_tools
---------------------------- 5. 节点函数 ----------------------------
def agent_node(state: AgentState) -> Dict[str, Any]:
"""
Agent节点：调用模型决定下一步动作（生成回复或调用工具）
"""
logger.info("--- Agent 节点开始 ---")
messages = state["messages"]
model = get_model()
try:
response = model.invoke(messages)
logger.debug(f"模型响应: {response}")
return {"messages": [response], "step": "agent"}
except Exception as e:
error_msg = f"模型调用失败: {str(e)}"
logger.error(error_msg)
# 返回错误消息给用户
return {
"messages": [AIMessage(content=f"抱歉，我遇到技术问题：{error_msg}")],
"error_count": state.get("error_count", 0) + 1,
"step": "error"
}
条件路由函数（可自定义，这里复用内置的tools_condition）
def should_continue(state: AgentState) -> str:
"""判断下一步：调用工具还是结束"""
# tools_condition 返回 "tools" 或 END
return tools_condition(state)
---------------------------- 6. 构建图 ----------------------------
def build_agent_graph():
"""构建LangGraph工作流"""
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(TOOLS))
# 设置入口
graph.set_entry_point("agent")
# 条件边
graph.add_conditional_edges(
"agent",
should_continue,
{
"tools": "tools",
END: END
}
)
# 工具执行后回到agent
graph.add_edge("tools", "agent")
# 编译
app = graph.compile()
logger.info("Agent图编译完成")
return app
---------------------------- 7. 运行与流式输出 ----------------------------
def run_agent(user_input: str, stream: bool = False):
"""
运行Agent并返回结果
Args:
user_input: 用户输入文本
stream: 是否启用流式输出（逐token）
"""
app = build_agent_graph()
initial_state = {
"messages": [HumanMessage(content=user_input)],
"error_count": 0,
"step": "start"
}
if stream:
# 流式输出（按事件块）
print("🤖 Agent 流式响应:")
for event in app.stream(initial_state, stream_mode="values"):
last_message = event["messages"][-1]
if isinstance(last_message, AIMessage) and last_message.content:
print(last_message.content, end="", flush=True)
print("\n")
else:
# 普通同步调用
final_state = app.invoke(initial_state)
final_message = final_state["messages"][-1].content
print(f"🤖 Agent 最终回复:\n{final_message}\n")
---------------------------- 8. 示例执行 ----------------------------
if name == "main":
# 测试用例
test_queries = [
"北京今天天气怎么样？",
"帮我计算 (12 + 35) * 2 - 8 的结果",
"现在几点了？请给出格式 '年-月-日 时:分:秒'",
"查询一下东京的天气", # 不在模拟数据库中，触发错误处理
"计算 10/0", # 触发计算错误
]for q in test_queries:
print(f"\n👤 用户: {q}")
run_agent(q, stream=False) # 非流式模式
# 若想体验流式，改为 run_agent(q, stream=True)
print("-" * 50)
# 额外演示流式输出
print("\n🌟 流式模式演示:")
run_agent("请告诉我上海的天气，并计算 100 - 45", stream=True)
---
## 🧪 运行与验证
### 1. 安装依赖
```bash
pip install langgraph langchain-openai python-dotenv2. 配置 .env 文件
OPENAI_API_KEY=sk-xxxxx
OPENAI_MODEL=gpt-3.5-turbo
LOG_LEVEL=INFO3. 执行脚本
python langgraph_agent.py预期输出片段：
👤 用户: 北京今天天气怎么样？
--- Agent 节点开始 ---
调用工具 get_weather，参数: city=北京
🤖 Agent 最终回复:
北京今天的天气是晴朗，25°C。📌 工程化要点说明
🔧 进阶扩展建议
持久化：使用
langgraph.checkpoint添加MemorySaver或SqliteSaver实现多轮对话记忆。人工干预：插入
interrupt节点实现人机协同（Human-in-the-loop）。监控集成：通过
langsmith追踪完整调用链。异步支持：将
invoke替换为ainvoke，配合asyncio提高并发能力。
这个Demo可直接用于生产环境的Agent原型开发，并能够轻松扩展更多工具和复杂逻辑。
评论区