如何打造基于LangGraph的智能体（Agentic）AI应用
来源：     阅读:3
易浩激活码
发布于 2025-11-11 19:22
查看主页

在AI技术飞速迭代的当下，开发者们常常面临一个挑战：刚掌握的框架或技术，转眼就可能被新工具取代。而智能体（Agentic）系统作为AI应用的前沿形态，正凭借“自主决策、替代执行”的能力成为焦点——它能像专业助手一样，调用工具完成复杂任务。不过，构建实用的智能体应用并非易事，直接让大语言模型（LLM）自由决策往往会导致结果失控。

本文将以“澳大利亚丛林火灾生存计划生成器”为例，详细拆解如何用LangGraph框架打造受约束、高实用的智能体AI应用，从需求分析到架构设计，再到核心工作流落地，为你提供可复现的开发思路。

一、为什么选择LangGraph？智能体开发的痛点与解决方案

在讨论具体开发步骤前，我们需要先明确：为什么要使用LangGraph？直接用LangChain或纯LLM API不行吗？

1. 传统智能体开发的核心痛点

早期智能体开发常采用“LLM全权决策”模式：给LLM一个任务（如生成火灾计划），让它自主判断“需要什么信息→调用什么工具→下一步做什么”。但这种无约束设计很容易出问题：

LLM可能基于训练数据假设用户情况（如默认用户住在郊区），跳过关键风险评估步骤；

生成的计划可能忽略官方推荐的“风险评估→能力评估→计划生成”流程，实用性大打折扣；

当需要用户补充信息（如“你家距离灌木丛有多远？”）时，纯LLM难以暂停流程、等待输入并恢复上下文。

2. LangGraph的核心优势：用“图结构”约束智能体

LangGraph是LangChain生态下的框架，专为解决上述痛点设计。它的核心是“有向图工作流”，能将智能体的决策过程拆解为可控节点，同时保留LLM的灵活性。其三大关键特性完美适配实用型智能体开发：

用“节点”（任务单元）和“边”（流转规则）定义流程，支持循环（如“信息不足时重复提问”）、分支（如“风险低则询问是否继续，风险高则直接进入评估”）；

所有节点共享一个“全局状态”，存储用户输入、LLM结果等信息，避免上下文丢失，支持长流程断点恢复；

随时中断流程等待用户输入（如补充风险信息），再基于新信息继续执行，解决LLM“闭门造车”问题。

简单来说，LangGraph的价值在于：既不让LLM完全失控，也不把它变成“只能按线性步骤执行的工具”，而是在“约束”与“灵活”之间找到平衡——这正是实用智能体的核心需求。

二、开发前准备：明确需求、技术栈与前提条件

在动手写代码前，我们需要先理清“做什么”和“用什么做”，避免后期返工。

1. 需求定义：以“丛林火灾生存计划生成器”为例

我们的目标是打造一个能协助用户生成个性化火灾生存计划的智能体，核心需求基于澳大利亚新南威尔士州消防局的官方提议，拆解为三大核心步骤：

判断用户住所受丛林火灾影响的风险（低/高/不明），风险不明时向用户提问（如“住所距离最近的灌木丛有多远？”“周边是否有易燃物？”）；

判断用户是否具备“留守并保护住所”的条件（如是否有灭火设备、是否有火灾应对经验），同样支持补充提问；

根据用户选择（“撤离”或“留守”），生成结构化计划，包含具体行动步骤（如“撤离前需携带的物品”“留守时的灭火流程”）。

2. 技术栈选型

模块

工具/框架

作用说明

核心框架

LangGraph（Python版）

定义智能体工作流，管理节点、状态与流转规则

LLM服务

OpenAI ChatGPT（API调用）

处理自然语言理解（如分析风险）、生成内容（如提问、写计划），需付费API密钥

基础开发环境

Python 3.9+、pip

脚本开发与依赖管理

后续扩展（可选）

FastAPI、Streamlit

将命令行版本转为API服务，或添加Web UI界面

3. 前提条件

熟悉Python函数、字典（用于状态管理）、第三方库调用；

需充值并获取API密钥（测试阶段费用极低，单条请求成本一般低于0.01美元）；

提前通过pip安装所需库：pip install langgraph openai python-dotenv

三、核心开发步骤：用LangGraph构建智能体工作流

LangGraph开发的核心是“设计图结构→实现节点逻辑→定义流转规则→测试与优化”。我们以“火灾生存计划生成器”为例，逐步拆解每一步。

1. 第一步：设计工作流图（Graph）

第一要把“风险评估→能力评估→计划生成”的流程，转化为LangGraph可识别的“节点”和“边”。

每个“节点”对应一个Python函数，负责完成单一任务（如风险评估、提问）；“边”则基于“全局状态”中的值（如“风险等级”“用户选择”）决定下一步走向。

2. 第二步：定义全局状态（State）

全局状态是LangGraph的“数据中枢”，存储所有流程中需要共享的信息。我们用TypedDict（Python类型提示工具）定义状态结构，确保数据类型清晰：

from typing import TypedDict, Optional, List

class FirePlanState(TypedDict):
    # 用户基本信息（如住所位置、家庭人数）
    user_info: dict
    # 风险评估结果（low/high/unclear）
    risk_level: Optional[str]
    # 能力评估结果（capable/incapable/unclear）
    capability: Optional[str]
    # LLM生成的问题（用于向用户补充信息）
    questions: List[str]
    # 用户对问题的回答
    user_answers: List[str]
    # 最终计划内容（Markdown格式）
    final_plan: Optional[str]
    # 用户选择（evacuate/stay）
    user_choice: Optional[str]

所有节点函数的输入是FirePlanState，输出是“更新后的FirePlanState”——列如“风险评估节点”会修改risk_level和questions，“用户输入节点”会修改user_answers。

3. 第三步：实现节点逻辑（Nodes）

节点是工作流的“执行单元”，每个节点对应一个Python函数，完成单一任务。以下是核心节点的实现思路：

（1）风险评估节点（assess_risk）

功能：基于已有user_info和user_answers，让LLM判断风险等级，若信息不足则生成问题。

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

# 加载OpenAI API密钥（提议用.env文件存储，避免硬编码）
load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", api_key=os.getenv("OPENAI_API_KEY"))

def assess_risk(state: FirePlanState) -> FirePlanState:
    # 1. 整理上下文信息
    context = f"用户信息：{state['user_info']}
用户已回答：{state['user_answers']}"
    # 2. 构造LLM提示词（明确任务和输出格式）
    prompt = f"""
    基于以下信息，评估用户住所受丛林火灾的风险等级（仅输出low/high/unclear）：
    {context}
    若风险为unclear，请生成1-2个问题向用户补充信息（问题用中文，以列表形式呈现）；若风险明确，问题列表为空。
    输出格式要求：
    风险等级：[low/high/unclear]
    问题：[若有，列出来；若无，写“无”]
    """
    # 3. 调用LLM并解析结果
    response = llm.invoke(prompt).content
    # 4. 提取风险等级和问题，更新状态
    risk_level = response.split("风险等级：")[1].split("
")[0].strip()
    questions = response.split("问题：")[1].strip()
    questions_list = questions.split("
") if questions != "无" else []

    return {
        **state,  # 保留原有状态
        "risk_level": risk_level,
        "questions": questions_list
    }

（2）提问节点（ask_user）

功能：向用户展示questions中的问题，获取输入并更新user_answers。

def ask_user(state: FirePlanState) -> FirePlanState:
    if not state["questions"]:
        return state  # 无问题则直接返回

    # 1. 打印问题并获取用户输入
    print("需要您补充以下信息：")
    for i, q in enumerate(state["questions"], 1):
        print(f"{i}. {q}")
    user_input = input("请回答（用逗号分隔多个问题的答案）：")
    answers = [ans.strip() for ans in user_input.split(",")]

    # 2. 更新用户回答（追加到已有答案中，避免覆盖）
    updated_answers = state["user_answers"] + answers

    return {
        **state,
        "user_answers": updated_answers,
        "questions": []  # 清空问题，避免重复提问
    }

（3）其他核心节点

若风险等级为low，询问用户“是否仍需生成计划”，用户选择“否”则结束流程；

逻辑与assess_risk类似，判断用户是否具备留守防御能力；

基于全局状态中的风险、能力信息，让LLM生成结构化计划（Markdown格式）；

打印final_plan，结束流程。

4. 第四步：组装图结构（Graph）

用LangGraph的Graph类将节点和流转规则组合起来，核心是定义“边”的逻辑——即“从A节点出发，下一步该到哪个节点”。

from langgraph.graph import Graph

# 1. 初始化图，指定状态类型
graph = Graph(FirePlanState)

# 2. 添加节点（节点名→节点函数）
graph.add_node("assess_risk", assess_risk)
graph.add_node("ask_user", ask_user)
graph.add_node("confirm_continue", confirm_continue)
graph.add_node("assess_capability", assess_capability)
graph.add_node("generate_evac_plan", generate_evac_plan)
graph.add_node("generate_stay_plan", generate_stay_plan)
graph.add_node("show_plan", show_plan)

# 3. 定义边（流转规则）
# 规则1：从assess_risk出发，若风险不明→ask_user；若风险明确→confirm_continue
def risk_to_next(state: FirePlanState) -> str:
    if state["risk_level"] == "unclear":
        return "ask_user"
    return "confirm_continue"
graph.add_edge("assess_risk", risk_to_next)

# 规则2：从ask_user出发，返回assess_risk重新评估
graph.add_edge("ask_user", "assess_risk")

# 规则3：从confirm_continue出发，用户选“否”→结束；选“是”→assess_capability
def continue_to_next(state: FirePlanState) -> str:
    if state["user_choice"] == "no":
        return "__end__"  # LangGraph内置的“结束”标识
    return "assess_capability"
graph.add_edge("confirm_continue", continue_to_next)

# 规则4：能力评估后，根据用户选择进入对应计划生成节点（类似风险评估逻辑）
graph.add_edge("assess_capability", capability_to_next)  # 需自定义capability_to_next函数
graph.add_edge("generate_evac_plan", "show_plan")
graph.add_edge("generate_stay_plan", "show_plan")
graph.add_edge("show_plan", "__end__")

# 4. 设置起点节点
graph.set_entry_point("assess_risk")

# 5. 编译图（生成可运行的应用）
app = graph.compile()

5. 第五步：测试与优化

运行编译后的app，传入初始状态（如空的user_info和user_answers），模拟用户交互流程：

# 初始状态（空信息）
initial_state = {
    "user_info": {},
    "risk_level": None,
    "capability": None,
    "questions": [],
    "user_answers": [],
    "final_plan": None,
    "user_choice": None
}

# 运行应用
app.invoke(initial_state)

测试中需重点关注：

列如风险不明时是否会提问，用户回答后是否重新评估；

计划是否包含关键步骤，是否有“幻觉信息”（如虚构的消防提议）；

用户输入是否被保存，后续节点是否能读取到之前的信息。

若发现问题（如LLM生成的问题不相关），可通过优化提示词（如明确“问题需围绕住所与火灾的距离、周边环境”）或调整节点逻辑（如增加“问题有效性校验”）解决。

四、扩展与迭代：从命令行到生产级应用

完成命令行版本后，可根据需求进行扩展，让应用更贴近实际使用场景：

1. 转为API服务

用FastAPI将LangGraph应用封装为API，支持前端（如Web、移动端）调用：

from fastapi import FastAPI
from pydantic import BaseModel

app_fastapi = FastAPI()

# 定义请求体模型
class StateRequest(BaseModel):
    state: FirePlanState

@app_fastapi.post("/run-fire-plan")
def run_fire_plan(request: StateRequest):
    result = app.invoke(request.state)
    return result

2. 添加Web UI

用Streamlit快速搭建可视化界面，让非技术用户也能使用：

import streamlit as st

st.title("丛林火灾生存计划生成器")

# 存储状态（用session_state保存上下文）
if "state" not in st.session_state:
    st.session_state.state = initial_state

# 运行按钮
if st.button("开始评估"):
    # 调用LangGraph应用
    updated_state = app.invoke(st.session_state.state)
    st.session_state.state = updated_state

    # 展示结果
    if st.session_state.state["final_plan"]:
        st.markdown(st.session_state.state["final_plan"])
    elif st.session_state.state["questions"]:
        # 展示问题并获取输入
        answers = st.text_input("请补充信息：")
        if answers:
            st.session_state.state["user_answers"].extend(answers.split(","))

3. 切换LLM服务

若不想用OpenAI，可替换为开源LLM（如Llama 3、Qwen），只需修改llm的初始化逻辑：

# 示例：用LangChain调用本地Llama 3（需提前部署API服务）
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="llama-3-70b-chat",
    base_url="http://localhost:8000/v1",  # 本地LLM的API地址
    api_key="none"  # 开源LLM一般无需API密钥
)

五、智能体开发的核心原则

通过“火灾生存计划生成器”的案例，我们可以提炼出基于LangGraph开发实用智能体的核心原则：

流程先行，约束为王

先明确业务流程（如“风险→能力→计划”），再用LangGraph的图结构约束LLM的决策范围，避免无意义的自由发挥；

状态是核心

确保所有节点共享全局状态，避免上下文丢失——这是长流程智能体稳定运行的关键；

人类在环不可少

LLM 无法掌握所有用户的个性化信息（如 “住所是否有消防水源”“家中是否有老人需要协助撤离”），必须通过 “提问 - 回答” 的交互补充关键信息，这也是智能体区别于 “一次性生成工具” 的核心优势

迭代优化，贴近实际

初期无需追求 “完美的智能体”，可先搭建基础流程，再通过测试发现问题（如 LLM 提问偏离重点、流程分支遗漏），逐步优化节点逻辑和提示词 —— 本文案例中 “从无约束智能体改为 LangGraph 约束流程” 的调整，正是迭代优化的典型体现。

免责声明：本文为用户发表，不代表网站立场，仅供参考，不构成引导等用途。 系统环境
相关推荐
SQL存储过程（MySQL）详情及使用方法
EndNote 20 安装教程（附安装包下载）
Java中HashMap底层数据结构
strong、weak和unowned的区别
为什么go中的receiver name不推荐使用this或者者self
首页
搜索
订单
购物车
我的