LangGraph 提供了用于构建基于代理的应用程序的低级原语和高级预构建组件。重点介绍预构建的、即用型组件，旨在帮助您快速可靠地构建代理系统，而无需从头开始实现编排、内存或人工反馈处理。

Agent 核心架构

代理（Agent）由三大组件构成

组件	功能说明
大型语言模型（LLM）	驱动代理决策的核心推理引擎。
工具集（Tools 在线的 mcp 提供的能力）	代理可调用的函数/API（如搜索、计算、数据库操作等）。
提示（Prompt）	定义代理行为规则的指令模板（如任务目标、约束条件）。

代理运行流程（循环迭代）

选择工具 → 2. 执行工具 → 3. 观察结果 → 4. 决策下一步 → （循环至任务完成）

核心特性

特性	关键能力
内存集成	短期（会话内记忆） + 长期（跨会话持久化存储）。
人机交互控制	支持异步暂停/恢复，允许人工干预（审批、修正）。
实时流式支持	代理状态、工具输出、模型Token流式传输。
生产级部署工具	内置测试/调试/部署支持，兼容多种基础设施。
可视化IDE（Studio）	提供工作流检查与调试的可视化界面。

高级构建模块

LangGraph 提供预构建组件，避免从零开发：

编排逻辑：自动处理状态转移与任务调度。
内存管理：内置短期/长期记忆实现。
人工反馈集成：支持异步人工审核流程。

开发者价值：聚焦业务逻辑，无需重复实现底层基础设施。

功能包生态系统

包名	用途	安装命令
`langgraph-prebuilt`	基础代理组件（内置）	`pip install -U langgraph`
`langgraph-supervisor`	监督型代理工具	`pip install -U langgraph-supervisor`
`langgraph-swarm`	多智能体群体系统	`pip install -U langgraph-swarm`
`langchain-mcp-adapters`	工具/资源集成接口	`pip install -U langchain-mcp-adapters`
`langmem`	短期/长期记忆管理	`pip install -U langmem`
`agentevals`	代理性能评估工具	`pip install -U agentevals`

Agent智能体创建

创建ReAct Agent（create_react_agent）

使用 LangGraph 提供的 create_react_agent 快速创建一个具备工具调用能力的代理。

先了解下什么是 create_react_agent（https://langchain-ai.github.io/langgraph/reference/agents/）：

使用以下工具可视化由 create_react_agent 并查看相应代码的大纲。它允许您探索代理的基础结构，如以下内容的存在所定义：

tools：代理可用于执行任务的工具（函数、API 或其他可调用对象）的列表。
pre_model_hook：在调用模型之前调用的函数。它可用于压缩消息或执行其他预处理任务
response_format：用于约束最终输出类型的数据结构，例如 pydantic BaseModel

示例代码如下：

通过create_react_agent创建一个智能体，生成代理图（Mermaid 格式）

# @Time：2025/9/28 16:29
# @Author：jinglv
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

from app.agent.model.llms import qv_llm


@tool("加法计算", description="计算数值相加的工具")
def add(a: int, b: int) -> int:
    """
    计算两个数值相加的工具
    :param a: 数值a
    :param b: 数值b
    :return:
    """
    print("正在执行工具：add")
    return a + b


@tool("乘法计算", description="计算数值相加的工具")
def multiply(a: int, b: int) -> int:
    """
    计算两个数值相乘的工具
    :param a: 数值a
    :param b: 数值b
    :return:
    """
    print("正在执行工具：multiply")
    return a * b


# ===============创建一个agent==============
agent = create_react_agent(
    model=qv_llm.bind_tools([add, multiply]),
    tools=[add, multiply],
    # 使用字符串，直接传模版对象会有问题
    prompt="您是一位数学计算助手，请根据输入的指令，选择合适的工具计算出结果,",
)
# 流式执行
response = agent.stream({"messages": [{"role": "user", "content": "请计算1*2的结果"}]})
for item in response:
    print(item)

# ===============获取agent执行的流程图谱==============
res = agent.get_graph().draw_mermaid_png()
print(res)  # 输出二进制编码
with open("langgraph开发Agent.png", "wb") as f:
    f.write(res)

Agent详解

执行模式概览

LangGraph 代理支持同步与异步两种执行方式：

模式	方法	适用场景	代码示例
同步调用	`.invoke()` / `.stream()`	简单脚本/阻塞式任务	`response = agent.invoke(input)`
异步调用	`.ainvoke()` / `.astream()`	Web服务/高并发场景	`response = await agent.ainvoke(input)`

输入输出规范

输入格式

代理输入必须为包含 messages 键的字典，支持多种消息结构：

# 字符串快捷输入（自动转为 HumanMessage）
{"messages": "Hello"}  

# 单条消息字典
{"messages": {"role": "user", "content": "Hello"}}  

# 消息列表（支持多轮对话）
{"messages": [{"role": "user", "content": "Hello"}]}

输出格式

输出字典包含以下字段：

messages: 完整对话消息历史（含工具调用/响应）
structured_response: 结构化输出（如配置）
自定义状态字段（若定义 state_schema）

流式传输控制

流模式 (`stream_mode`)	输出内容
`"updates"`	状态增量更新（默认）
`"tokens"`	LLM 生成的原始 Token
`"tools"`	工具调用/响应事件

代码示例

# 同步流式调用
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "上海天气"}]},
    stream_mode="updates"  # 可选 "tokens" 或 "tools"
):
    print(chunk)  # 实时处理数据块

# 异步流式调用
async for chunk in agent.astream(...):
    process(chunk)

执行安全控制(设置最大递归限制)

防止无限循环，通过 recursion_limit 控制最大迭代次数

设置默认最大递归次数

agent = create_react_agent(
    # 绑定模型
    model=llm.bind_tools([add, multiply]),
    # 绑定工具
    tools=[add, multiply],
    # 使用字符串，直接传模版对象会有问题
    prompt=PromptTemplate(template="您是一位数学计算助手，请根据输入的指令，选择合适的工具计算出结果,请计算1*2的结果"),
).with_config({"recursion_limit": 30})

或者

agent = create_react_agent(
    # 绑定模型
    model=llm.bind_tools([add, multiply]),
    # 绑定工具
    tools=[add, multiply],
    # 使用字符串，直接传模版对象会有问题
    prompt=PromptTemplate(template="您是一位数学计算助手，请根据输入的指令，选择合适的工具计算出结果,请计算1*2的结果"),
)
# 设置默认最大递归次数
agent = agent.with_config({"recursion_limit": 30})

执行时设置最大递归次数

agent = create_react_agent(
    # 绑定模型
    model=llm.bind_tools([add, multiply]),
    # 绑定工具
    tools=[add, multiply],
    # 使用字符串，直接传模版对象会有问题
    prompt=PromptTemplate(template="您是一位数学计算助手，请根据输入的指令，选择合适的工具计算出结果,请计算1*2的结果"),
)

# 在执行的时候去设置最大递归的次数
response = agent.stream({"messages": [{"role": "user", "content": "请计算1*2的结果"}]},
                        {"recursion_limit": 30}
                        )
for item in response:
    print(item)

Agent开发案例

实现一个数据库操作智能体

# @Time：2025/9/29 09:20
# @Author：jinglv

import dotenv
import pymysql
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

from app.agent.model.llms import qv_llm

# 加载.env文件中的环境变量
dotenv.load_dotenv()


# 第一步定义工具函数
@tool('数据库操作', description='连接数据库,执行sql语句，并返回执行的结果')
def mysql_executor(sql: str):
    """
    连接数据库，执行sql语句
    需要先安装pymysql
    :param sql:
    :return:
    """
    try:
        connent = pymysql.connect(host='localhost',
                                  port=3306,
                                  user='root',
                                  password='12345678',
                                  cursorclass=pymysql.cursors.DictCursor,
                                  db='test',
                                  autocommit=True,
                                  )
        cursor = connent.cursor()
        cursor.execute(sql)
        connent.commit()
    except pymysql.err.OperationalError as e:
        return 0
    result = cursor.fetchall()
    cursor.close()
    connent.cursor()
    return result


class DBAgent:

    def __init__(self):
        self.llm = qv_llm

    def main(self, input):
        agent = create_react_agent(
            model=self.llm,
            tools=[mysql_executor],
            # # 这个是给智能体进行身份定位，和决策建议的提示词
            prompt="""
            你是一位资深的DBA，现在需要你根据用户的需求，编写对应的sql语句，调用数据库操作的工具，执行sql语句，并返回执行的结果,
            每一步执行完都需要去分析当前的执行进度，以及规划下一步的任务执行
            """
        )
        response = agent.stream({"messages": input})
        for item in response:
            print(item)


if __name__ == '__main__':
    DBAgent().main("创建一个用户表，需要用户id，用户名，用户密码，用户邮箱，用户手机号，用户地址，用户性别这些字段")
    # DBAgent().main("往用户表里面插入10条数据")
    # DBAgent().main(
    #     "创建一个学生表，需要学生id，学生名，学生性别，学生年龄，学生手机号，学生邮箱，学生地址，学生生日，学生班级，学生部门这些字段，并往学生表中插入20条数据")

注意：以上没有实现记忆功能，在进行已创建表的基础上进行数据表的增删改查操作是会报错的

Jean's Blog

LangGraph之Agent开发