LangGraph框架入门：构建复杂AI Agent的图结构编程

1. LangGraph与AI Agent开发入门

最近在AI应用开发领域，LangGraph这个新兴框架开始受到广泛关注。作为一个基于图结构的编程模型，它特别适合构建复杂的AI Agent系统。今天我就来分享一下如何用LangGraph快速搭建你的第一个AI Agent，并附上完整可运行的代码示例。

LangGraph本质上是一个用于编排AI工作流的框架，它通过节点(Node)和边(Edge)的概念，将复杂的AI任务分解为可管理的组件。与传统的线性编程不同，图结构能够更好地表达现实世界中决策的复杂性和条件分支。

提示：如果你熟悉LangChain，可以把LangGraph理解为它的"升级版"——专门为处理更复杂的多步骤、有条件的工作流而设计。

2. 环境准备与基础配置

2.1 安装必要依赖

首先需要创建一个干净的Python环境（3.8+版本），然后安装核心包：

bash复制pip install langgraph langchain-openai

这里我们选择使用OpenAI的模型作为AI核心，但你也可以替换为其他兼容的模型提供商。

2.2 配置API密钥

在项目根目录创建.env文件，添加你的OpenAI API密钥：

text复制OPENAI_API_KEY=你的实际密钥

然后在代码中通过python-dotenv加载配置：

python复制from dotenv import load_dotenv
load_dotenv()

3. 构建第一个AI Agent

3.1 定义基础节点

我们先创建一个简单的问答节点：

python复制from langgraph.graph import Graph
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 定义问答节点
def qa_node(state):
    human_input = state["messages"][-1].content
    response = llm.invoke([HumanMessage(content=human_input)])
    return {"messages": [AIMessage(content=response.content)]}

3.2 创建图结构

现在将这些节点组装成工作流：

python复制# 初始化图
workflow = Graph()

# 添加节点
workflow.add_node("qa", qa_node)

# 设置入口点
workflow.set_entry_point("qa")

# 设置结束点
workflow.set_finish_point("qa")

# 编译图
app = workflow.compile()

3.3 运行Agent

现在可以测试这个基础Agent了：

python复制# 初始状态
initial_state = {"messages": [HumanMessage(content="你好！")]}

# 执行图
result = app.invoke(initial_state)
print(result["messages"][-1].content)

4. 扩展为条件工作流

4.1 添加决策节点

让我们的Agent能根据输入内容决定不同的响应策略：

python复制def router_node(state):
    last_message = state["messages"][-1].content
    if "?" in last_message:
        return "qa"
    elif "计算" in last_message:
        return "calc"
    else:
        return "general"

# 添加计算节点
def calc_node(state):
    # 这里可以接入计算工具
    return {"messages": [AIMessage(content="计算结果为...")]}

4.2 完善图结构

更新工作流配置：

python复制workflow.add_node("router", router_node)
workflow.add_node("calc", calc_node)

workflow.add_edge("qa", "router")
workflow.add_edge("calc", "router")

workflow.add_conditional_edges(
    "router",
    router_node,
    {
        "qa": "qa",
        "calc": "calc",
        "general": "qa"
    }
)

5. 高级功能实现

5.1 记忆功能集成

让Agent能记住对话历史：

python复制from langgraph.checkpoint import MemorySaver

memory = MemorySaver()

# 更新图配置
workflow = Graph()
workflow.add_node("qa", qa_node)
workflow.add_edge("qa", "__end__")
app = workflow.compile(checkpointer=memory)

5.2 多Agent协作

创建两个专业Agent协同工作：

python复制# 专业问答Agent
expert_llm = ChatOpenAI(model="gpt-4")
def expert_node(state):
    response = expert_llm.invoke(state["messages"])
    return {"messages": [AIMessage(content=f"专家意见:{response.content}")]}

# 添加到工作流
workflow.add_node("expert", expert_node)
workflow.add_conditional_edge(
    "router",
    lambda x: "expert" if "专业" in x["messages"][-1].content else "qa"
)

6. 部署与优化

6.1 性能优化技巧

对于生产环境，可以考虑以下优化：

1. 节点级缓存：

python复制from langgraph.cache import InMemoryCache
workflow.cache = InMemoryCache()

2. 批量处理请求：

python复制app.batch([input1, input2])

3. 异步支持：

python复制await app.ainvoke(input)

6.2 错误处理机制

增强Agent的健壮性：

python复制from langgraph.graph import ERROR

def safe_node(state):
    try:
        # 业务逻辑
    except Exception as e:
        return {"messages": [AIMessage(content=f"错误:{str(e)}")], ERROR: True}

workflow.add_node("safe_node", safe_node)
workflow.add_edge(ERROR, "error_handler")

7. 完整代码示例

以下是整合所有功能的完整实现：

python复制from langgraph.graph import Graph
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

# 初始化组件
llm = ChatOpenAI(model="gpt-3.5-turbo")
expert_llm = ChatOpenAI(model="gpt-4")

# 定义节点
def qa_node(state):
    response = llm.invoke(state["messages"])
    return {"messages": [AIMessage(content=response.content)]}

def expert_node(state):
    response = expert_llm.invoke(state["messages"])
    return {"messages": [AIMessage(content=f"专家意见:{response.content}")]}

def router_node(state):
    last_msg = state["messages"][-1].content
    if "?" in last_msg:
        return "qa"
    elif "专家" in last_msg:
        return "expert"
    else:
        return "qa"

# 构建图
workflow = Graph()
workflow.add_node("qa", qa_node)
workflow.add_node("expert", expert_node)
workflow.add_node("router", router_node)

workflow.add_edge("qa", "router")
workflow.add_edge("expert", "router")

workflow.add_conditional_edges(
    "router",
    router_node,
    {
        "qa": "qa",
        "expert": "expert"
    }
)

workflow.set_entry_point("qa")
app = workflow.compile()

# 测试运行
result = app.invoke({
    "messages": [HumanMessage(content="专家您好，请问...")]
})
print(result["messages"][-1].content)

8. 实际应用中的经验分享

在真实项目中使用LangGraph时，有几个关键点需要注意：

1. 节点设计原则：

• 每个节点应该只做一件事
• 节点间通过清晰定义的接口通信
• 避免节点间产生隐式依赖

2. 调试技巧：

python复制# 查看图结构
print(workflow.get_graph().draw_ascii())

# 跟踪执行流程
app = workflow.compile(debug=True)

3. 性能监控：

python复制from langgraph.monitoring import WandbTracer

app = workflow.compile(tracers=[WandbTracer()])

4. 版本控制：

python复制# 导出图配置
config = workflow.to_json()

# 导入配置
new_workflow = Graph.from_json(config)

对于复杂业务场景，建议采用渐进式开发策略：先构建最小可行图，然后逐步添加节点和条件分支。同时要注意控制图的复杂度，当节点超过15个时，考虑拆分为子图。