/ AI  LangGraph  Agent  LLM  Python  工作流  大模型  人工智能 

LangGraph 实战：从零构建生产级 AI Agent 工作流编排系统


封面

什么是 AI Agent 工作流编排？

随着大语言模型（LLM）能力的不断提升，单纯的"一问一答"模式已经无法满足复杂任务的需求。AI Agent 工作流编排应运而生——它将 LLM 的推理能力与外部工具调用、多步骤执行、状态管理有机结合，使 AI 能够自主完成原本需要人工介入的复杂任务。

LangGraph 是目前最受欢迎的 AI Agent 编排框架之一，由 LangChain 团队推出。它基于有向无环图（DAG）的思想，将 Agent 的执行过程抽象为节点（Node）和边（Edge），支持条件跳转、循环控制和并行执行，让开发者能够精细控制 Agent 的每一步决策。

  • 节点（Node）：代表一个具体的处理步骤，如调用 LLM、执行工具、处理数据等

  • 边（Edge）：定义节点之间的流转关系，支持条件分支

  • 状态（State）：贯穿整个图的共享数据结构，各节点均可读写

  • 图（Graph）：由节点和边组成的完整执行拓扑

环境搭建与基本概念

在开始实战之前，我们需要先搭建好开发环境。LangGraph 支持 Python 3.9+，推荐使用虚拟环境管理依赖。

# 创建虚拟环境
python3 -m venv agent-env
source agent-env/bin/activate

# 安装依赖
pip install langgraph langchain-openai langchain-core python-dotenv

# 验证安装
python3 -c "import langgraph; print(langgraph.__version__)"

创建 .env 文件配置 API 密钥：

OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1

LangGraph 的核心数据结构是 StateGraph，我们需要先定义状态类型：

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_action: str
    tool_result: str
    iteration_count: int

构建多工具 Agent 核心逻辑

一个实用的 AI Agent 通常需要多种工具的支持。下面我们定义几个常用工具：搜索、计算器和代码执行器。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

@tool
def search_web(query: str) -> str:
    """搜索互联网获取实时信息"""
    # 实际场景可接入 Tavily、SerpAPI 等
    return f"搜索结果：关于'{query}'的相关信息..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

tools = [search_web, calculate, get_current_time]

# 初始化 LLM 并绑定工具
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

接下来定义 Agent 的核心节点函数：

from langchain_core.messages import SystemMessage

def agent_node(state: AgentState) -> AgentState:
    """Agent 推理节点：决定下一步动作"""
    system_prompt = SystemMessage(content="""你是一个智能助手，能够使用工具解决复杂问题。
    请根据用户需求，合理规划步骤并调用适当的工具。""")
    
    messages = [system_prompt] + list(state["messages"])
    response = llm_with_tools.invoke(messages)
    
    return {
        "messages": [response],
        "iteration_count": state.get("iteration_count", 0) + 1
    }

def should_continue(state: AgentState) -> str:
    """条件边：判断是否继续执行"""
    last_message = state["messages"][-1]
    
    # 超过最大迭代次数，强制结束
    if state.get("iteration_count", 0) >= 10:
        return "end"
    
    # 有工具调用则继续，否则结束
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "continue"
    return "end"

组装图结构与条件分支

有了节点和工具之后，我们将它们组装成完整的 StateGraph：

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

def build_agent_graph():
    """构建 Agent 执行图"""
    graph = StateGraph(AgentState)
    
    # 添加节点
    graph.add_node("agent", agent_node)
    graph.add_node("tools", ToolNode(tools))
    
    # 设置入口节点
    graph.set_entry_point("agent")
    
    # 添加条件边：agent 节点根据 should_continue 决定走向
    graph.add_conditional_edges(
        "agent",
        should_continue,
        {
            "continue": "tools",  # 有工具调用 → 执行工具
            "end": END            # 无工具调用 → 结束
        }
    )
    
    # tools 执行完成后回到 agent 继续推理
    graph.add_edge("tools", "agent")
    
    return graph.compile()

# 编译图
app = build_agent_graph()
print("Agent 图构建成功！")

运行 Agent 的方式非常简单：

from langchain_core.messages import HumanMessage

# 初始化状态
initial_state = {
    "messages": [HumanMessage(content="现在几点了？然后帮我计算 (123 * 456) + 789 的结果")],
    "iteration_count": 0
}

# 流式执行，实时查看每个节点的输出
for event in app.stream(initial_state, stream_mode="values"):
    last_msg = event["messages"][-1]
    print(f"[节点输出] {type(last_msg).__name__}: {last_msg.content[:100] if last_msg.content else '(工具调用中...)'}")

生产环境最佳实践

将 AI Agent 部署到生产环境时，需要考虑以下关键问题：

  • 检查点持久化（Checkpointing）：使用 SqliteSaver 或 PostgresSaver 保存执行状态，支持断点续传和历史回溯

  • 人工介入（Human-in-the-loop）：在关键节点添加 interrupt_before，让人工审核后再继续执行

  • 并行执行：对于独立的子任务，使用 Send API 实现真正的并行处理，提升效率

  • 错误处理：为工具节点添加重试机制和降级策略，避免单点故障

  • 可观测性：接入 LangSmith 进行追踪和调试，快速定位问题

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用 SQLite 持久化检查点
with SqliteSaver.from_conn_string("agent_state.db") as memory:
    app_with_memory = build_agent_graph()
    
    # 使用 thread_id 区分不同会话
    config = {"configurable": {"thread_id": "user_session_001"}}
    
    result = app_with_memory.invoke(initial_state, config=config)
    
    # 之后可以继续同一会话
    follow_up = {
        "messages": [HumanMessage(content="基于刚才的计算，再乘以 2 是多少？")]
    }
    result2 = app_with_memory.invoke(follow_up, config=config)
    print("续对话结果：", result2["messages"][-1].content)

LangGraph 代表了 AI Agent 开发的新范式——从简单的链式调用到复杂的有状态图执行。掌握它将帮助你构建真正可靠、可扩展的生产级 AI 应用。随着 Agent 技术的快速演进，工作流编排能力正成为每位 AI 工程师的必备技能。

发布评论

热门评论区：