基于LangGraph的Agent

LangGraph 专为希望构建强大、适应性的 AI 智能体的开发者而设计。开发者选择 LangGraph 的原因是：

• 可靠性和可控性：通过将模型和人工干预相结合，LangGraph 可为长时间运行的工作流注入上下文，使您的智能体保持正常运行。
• 低层级和可扩展性：使用完全描述性的底层级语言构建自定义智能体，不受限制自定义的偏执约束。设计可扩展的多智能体系统，其中每个智能体都为您的用例量身定制特定角色。
• 一流的流式传输支持：通过多模态流式传输和中间步骤流式传输，LangGraph 让用户实时清晰地了解智能体的推理和行动过程。

LangGraph 支持两种对于构建对话代理至关重要的内存类型：

• 短期内存：通过在会话中维护消息历史来跟踪正在进行的对话。
• 长期内存：在不同会话之间存储用户特定或应用程序级别的数据。

常见问题：
使用 LangGraph 需要使用 LangChain 吗？它们有什么区别？

不。LangGraph 是一个用于复杂代理系统的编排框架，比 LangChain 代理更底层且更可控。 LangChain
提供了与模型及其他组件交互的标准接口，适用于直接的链式操作和检索流程。

---

LangGraph 与其他代理框架有何不同？

其他代理框架可以处理简单、通用的任务，但对于公司定制的复杂任务则力不从心。LangGraph
提供了一个更具表达力的框架来处理公司独特的任务，而不会将用户限制在单一的黑盒认知架构中。

---

LangGraph 会影响我的应用性能吗？

LangGraph 不会给您的代码增加任何开销，并且是专门为流式工作流设计的。

---

LangGraph 是开源的吗？它是免费的吗？

是的。LangGraph 是一个采用 MIT 许可的开源库，可免费使用。

---

LangGraph 和 LangGraph 平台有何不同？

LangGraph 是一个有状态的编排框架，为代理工作流带来了额外的控制。LangGraph 平台是一项用于部署和扩展 LangGraph
应用的服务，它提供了一个有主见的 API 来构建代理用户体验，并集成了一个开发者工作室。

---

LangGraph 平台是开源的吗？

不。LangGraph 平台是专有软件。

LangGraph 平台有一个免费的自托管版本，可访问基本功能。云 SaaS
部署选项在测试版期间免费，但最终将成为付费服务。我们将在收费前始终提供充分的通知，并以优惠价格回馈我们的早期用户。自托管部署选项是付费服务。联系我们的销售团队了解更多。

欲了解更多信息，请参阅我们的 LangGraph 平台定价页面。

---

LangGraph 是否支持不带工具调用的 LLM？

是的！您可以将 LangGraph 与任何 LLM 一起使用。我们使用支持工具调用的 LLM 的主要原因是，这通常是让 LLM
决定做什么的最便捷方式。如果您的 LLM 不支持工具调用，您仍然可以使用它——您只需要编写一些逻辑来将原始 LLM
字符串响应转换为要执行的操作的决策。

---

LangGraph 是否支持开源 LLM？

是的！LangGraph 对于底层使用的 LLM 完全不挑剔。我们在大多数教程中使用闭源 LLM 的主要原因是它们无缝支持工具调用，而开源
LLM 通常不支持。但是工具调用并非必需（请参阅本节），因此您可以完全使用 LangGraph 与开源 LLM。

---

我可以在不登录 LangSmith 的情况下使用 LangGraph Studio 吗？

是的！您可以使用 LangGraph Server 的开发版本在本地运行后端。这会连接到作为 LangSmith 一部分托管的 Studio
前端。如果您将环境变量 LANGSMITH_TRACING 设置为 false，则不会向 LangSmith 发送任何跟踪。

什么是 Agent？

人类在复杂的模式识别任务中表现卓越，但通常需要借助工具（如书籍、搜索引擎或计算器）来补充先验知识以得出结论。同理，生成式 AI 模型可通过训练使用工具获取实时信息或建议的实际动作。例如：

• 模型可利用数据库检索工具获取客户购买历史以生成个性化购物推荐
• 基于用户查询，模型可通过 API 调用发送邮件或完成金融交易

为实现此能力，模型需具备：

1. 外部工具集访问权限
2. 自主规划与执行任务的推理能力

这种结合推理逻辑与外部信息访问的系统，即构成智能体（Agent）。

智能体的认知架构中有三个基本组件：模型（Model）、工具（Tools）以及一个提供指令的提示。

LLM 在一个循环中运行。在每次迭代中，它会选择一个要调用的工具，提供输入，接收结果（一个观察），并利用该观察来指导下一个动作。循环会一直持续，直到满足停止条件——通常是 Agent 已经收集到足够的信息来响应用户时。

Anthropic 将 Agent 系统划分为两类：

1. 第一类是 workflow。遵循预定义的工作流，编排 LLM 和工具，固定代码路径。

2. Agent：此类 Agent 被定义为完全自主的系统，这些系统在较长时间内独立运行，可以动态地指导自身流程和工具使用的系统。通过自身的推理、规划能力，自主控制，完成任务。

1. LangGraph 本地服务

LangGraph CLI 是一个多平台命令行工具，用于在本地构建和运行 LangGraph API 服务器。生成的服务器包含您的图的所有运行、线程、助手等的 API 端点，以及运行您的代理所需的其他服务，包括用于检查点和存储的托管数据库。

LangGraph CLI 提供以下核心功能

命令	描述
langgraph build	构建一个可直接部署的 LangGraph API 服务器的 Docker 镜像。
langgraph dev	启动一个轻量级开发服务器，无需 Docker 安装。此服务器非常适合快速开发和测试。此功能在 0.1.55 及更高版本中可用。
langgraph dockerfile	生成一个 Dockerfile，可用于为 LangGraph API 服务器构建镜像并部署实例。如果您想进一步定制 Dockerfile 或更自定义的方式部署，此处能很有用。
langgraph up	在本地 Docker 容器中启动一个 LangGraph API 服务器实例。这要求 Docker 服务器在本地运行。本地开发需要 LangSmith API 密钥，生产使用需要许可证密钥。

---

安装和使用步骤

一、创建 Python 虚拟环境

虚拟环境的安装步骤

1. 安装好 Python 解释器：
   Python >= 3.11 is required.

2. 安装虚拟环境库，在 cmd 中输入：

pip install virtualenv

3. 创建虚拟环境，在 cmd 中切换到需要创建虚拟环境的目录下，执行：

。。。>D:\python-3.9\Scripts\virtualenv fastapi_env

virtualenv env_name

4. 激活虚拟环境，在 cmd 中进入到第三步创建的 env_name/Scripts 目录下，执行：

activate

执行成功后，在 cmd 中，当前输入行前面会有 (env_name) 的前缀
在当前状态下，使用 pip 就是在虚拟环境中安装第三方库了

5. 退出虚拟环境，cmd 中输入：

deactivate

二、安装 LangGraph CLI

#Python >= 3.11 is required.
pip install --upgrade "langgraph-cli[inmem]"

三、创建 LangGraph 应用

从 new-langgraph-project-python 模板或 new-langgraph-project-js 模板创建一个新应用。此模板演示了一个基于 Python 的代码生成器，可以根据自己的逻辑进行扩展。

注意: 如果您使用 langgraph new 命令时未指定模板，将显示一个交互式菜单，允许您从可用模板列表中进行选择。

langgraph new path/to/your/app --template new-langgraph-project-python

四、安装项目依赖

在您的 LangGraph 应用的根目录下，以编辑模式安装依赖项，以便服务器使用您的本地更改。

在 LangGraph 中，pyproject.toml 代替传统的 setup.py 和 requirements.txt，可能包含以下扩展配置：

• 依赖分组: 如 [project.optional-dependencies] 定义 dev（开发工具）和 test（测试框架）依赖。
• 动态版本控制: 通过 requires-python = ">=3.9" 指定 Python 版本兼容性。
• CI/CD 集成: 通过 [tool.*] 配置与 GitHub Actions 或 GitLab CI 的交互。
  

cd path/to/your/app
# pyproject.toml 对应依赖安装
pip install -e . 

五、修改 graph.py 的代码

# 本地私有化部署的大模型
llm = ChatOpenAI(
    model='qwen3-8b',
    temperature=0.8,
    api_key='xx',
    base_url="http://localhost:6006/v1",
    extra_body={'chat_template_kwargs': {'enable_thinking': False}},
)

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

graph = create_react_agent(
    llm,
    tools=[get_weather],
    prompt="You are a helpful assistant"
)

六、启动 LangGraph 服务器

langgraph dev 命令以内存模式启动 LangGraph 服务器。此模式适用于开发和测试目的。对于生产用途，请部署 LangGraph 服务器并使其能够访问持久存储后端。

命令选项：

选项	默认	描述
-c, --config 文件	langgraph.json	声明依赖项、图和环境变量的配置文件的路径
–host TEXT	127.0.0.1	服务器绑定的主机
–port 整数	2024	服务器绑定的端口
–no-reload		禁用自动重载
–n-jobs-per-worker 整数		每个工作进程的作业数。默认为 10
–debug-port 整数		调试器监听的端口
–wait-for-client	FALSE	等待调试器客户端连接到调试端口后再启动服务器
–no-browser		服务器启动时跳过自动打开浏览器
–studio-url TEXT		要连接的 LangGraph Studio 实例的 URL。默认为 https://smith.langchain.com
–allow-blocking	FALSE	不为代码中的同步 I/O 阻塞操作引发错误（在 0.2.6 中添加）
–tunnel	FALSE	通过公共隧道（Cloudflare）暴露本地服务器，以便远程前端访问。这避免了 Safari 等浏览器或网络阻塞 localhost 连接的问题
–help		显示命令文档

---

七、测试和访问 Agent 的 API

1. LangGraph Studio

LangGraph Studio 是一个专用 UI，您可以将其连接到 LangGraph API 服务器，以在本地可视化、交互和调试您的应用。通过访问 langgraph dev 命令输出中提供的 URL，在 LangGraph Studio 中测试您的 Agent 和图。

2. Python SDK 测试

pip install langgraph-sdk

异步测试:

from langgraph_sdk import get_client
import asyncio

client = get_client(url="http://localhost:2024")

async def main():
    async for chunk in client.runs.stream(
        None,  # Threadless run
        "agent",  # Name of assistant. Defined in langgraph.json.
        input={
            "messages": [
                {
                    "role": "human",
                    "content": "What is LangGraph?",
                }
            ],
        },
    ):
        print(f"Receiving new event of type: {chunk.event}...")
        print(chunk.data)
        print("\n\n")

asyncio.run(main())

同步测试:

from langgraph_sdk import get_sync_client

client = get_sync_client(url="http://localhost:2024")

for chunk in client.runs.stream(
    None,  # Threadless run
    "agent",  # Name of assistant. Defined in langgraph.json.
    input={
        "messages": [
            {
                "role": "human",
                "content": "What is LangGraph?",
            }
        ],
    },
    stream_mode="messages-tuple",
):
    print(f"Receiving new event of type: {chunk.event}...")
    print(chunk.data)
    print("\n\n")

	# if isinstance(chunk.data, list) and 'type' in chunk.data[0] and chunk.data[0]['type'] == 'AIMessageChunk':
        # print(chunk.data[0]['content'], end='')

#if isinstance(chunk.data, list) and 'type' in chunk.data[0] and chunk.data[0]['type'] == 'AIMessageChunk':
    #print(chunk.data[0]['content'], end='\n')        

3. JavaScript SDK测试

安装 LangGraph JS SDK

npm install @langchain/langgraph-sdk

向 LangGraph 服务区发送消息：

const { Client } = await import("@langchain/langgraph-sdk");

// only set the apiUrl if you changed the default port when calling langgraph dev
const client = new Client({ apiUrl: "http://localhost:2024" });

const streamResponse = client.runs.stream(
    null, // Threadless run
    "agent", // Assistant ID
    {
        input: {
            messages: [
                { role: "user", content: "What is LangGraph?" }
            ]
        },
        streamMode: "messages-tuple",
    }
);

for await (const chunk of streamResponse) {
    console.log(chunk);
    console.log(JSON.stringify(chunk.data));
    console.log("\n\n");
}

4. REST API测试：

curl -s --request POST \
    --url "http://localhost:2024/runs/stream" \
    --header 'Content-Type: application/json' \
    --data "{
    \"assistant_id\": \"agent\",
    \"input\": {
        \"messages\": [
            {
                \"role\": \"human\",
                \"content\": \"What is LangGraph?\"
            }
        ]
    }
}"

2. Tool工具的定义

在构建Agent时，您需要为其提供一个它可以使用的工具列表。除了实际调用的函数之外，工具还包括几个组件：

属性	类型	描述
名称	str	在提供给LLM或代理的一组工具中必须是唯一的。
描述	str	描述工具的作用。被LLM或代理用作上下文。
args_schema	pydantic.BaseModel	可选但推荐，如果使用回调处理程序则为必需。它可用于为预期参数提供更多信息（例如，少量示例）或验证。
return_direct	boolean	仅与代理相关。当为True时，在调用给定工具后，代理将停止并将结果直接返回给用户。

注意： 如果工具具有精心选择的名称、描述和args_schema，模型将表现得更好。

LangChain 支持从以下对象创建工具：

1. 函数；
2. LangChain Runnables；
3. 通过从 BaseTool 子类化 – 这是最灵活的方法，它提供了最大的控制程度，但代价是需要付出更多的努力和编写更多的代码。

一、从函数创建工具

这个 @tool 装饰器是定义自定义工具的最简单方法。默认情况下，装饰器使用函数名称作为工具名称，但可以通过将字符串作为第一个参数传递来覆盖。此外，装饰器将使用函数的文档字符串作为工具的描述 - 因此必须提供文档字符串。请注意，@tool 支持解析注释、嵌套模式和其他特性：

---

二、从可运行对象（Runnable）创建工具

接受字符串或 dict 输入的 LangChain Runnables 可以使用 as_tool 方法转换为工具，该方法允许为参数指定名称、描述和其他模式信息。

三、子类化 BaseTool

可以通过从 BaseTool 子类化来定义自定义工具。这提供了对工具定义的最大控制，但需要编写更多代码。

---

3. Agent的上下文和记忆

一、上下文

上下文包括消息列表之外的任何数据，这些数据可以影响代理行为或工具执行。这可以是：

• 运行时传入的信息，如 user_id 或 API 凭据。
• 多步推理过程中更新的内部状态。
• 来自先前交互的持久记忆或事实。

LangGraph 提供了三种提供上下文的主要方式：

类型	描述	可变？	生命周期
Configurable	在运行开始时传入的数据	❌	每次运行
AgentState	执行期间可更改的动态数据	✅	每次运行或对话
长期记忆（存储）	可在对话之间共享的数据	✅	跨对话

---

上下文用途

您可以使用上下文来：

• 调整模型看到的系统提示
• 为工具提供必要的输入
• 在正在进行的对话中跟踪事实

---

Configurable

配置适用于不可变数据，如用户元数据或 API 密钥。当您有在运行期间不会更改的值时使用。

使用保留用于此目的的键 “configurable” 指定配置。

agent.invoke(
    {"messages": [{"role": "user", "content": "hi!"}]},
    config={"configurable": {"user_id": "user_123"}}
)

案例：通过 Configurable 传入参数，来动态设置系统提示词

状态 AgentState（可变上下文）

状态在运行期间充当 Agent 的记忆，可以短期存储也可以长期存储。它保存在执行期间演变的动态数据，例如从工具或 LLM 输出派生的值。

二、记忆存储

这是一个强大的功能，允许您在多次调用中持久化代理的状态。否则，状态仅限于单次运行。

短期存储：线程级存储

短期存储使 Agent 能够跟踪多轮对话。要使用它，您必须：

1. 在创建代理时提供 checkpointer。checkpointer 可以实现代理状态的持久性。
2. 在运行代理时在配置中提供 thread_id。thread_id 是对话会话的唯一标识符。

# 内存：开发环境
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

# 在生产环境中，使用由数据库支持的检查点
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 使用 checkpointer 的代码块
    pass

# 必须安装：pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres

# 生产环境：Redis
from langgraph.checkpoint.redis import RedisSaver

# pip install -U langgraph langgraph-checkpoint-redis

DB_URI = "redis://localhost:6379"

with RedisSaver.from_conn_string(DB_URI) as checkpointer:
    # 使用 checkpointer 的代码块
    pass

长期存储：跨线程存储

使用长期内存来跨对话存储用户特定或应用程序特定的数据。这对于聊天机器人等应用程序非常有用，您可能希望记住用户偏好或其他信息。

要使用长期记忆，您需要：

1. 配置一个存储以在调用之间持久化数据。
2. 使用 get_store 函数从工具或提示中访问存储。

# 开发环境中：内存
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 在生产环境中，使用由数据库支持的存储
from langgraph.store.postgres import PostgreSQLStore
from langgraph.checkpoint.postgres import PostgreSQLSaver

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres"

with (
    PostgreSQLStore.from_conn_string(DB_URI) as store,
    PostgreSQLSaver.from_conn_string(DB_URI) as checkpointer,
):
    store.setup()
    checkpointer.setup()

# Redis 存储
from langgraph.store.redis import RedisStore
from langgraph.checkpoint.redis import RedisSaver

DB_URI = "redis://localhost:6379"
with (
    RedisStore.from_conn_string(DB_URI) as store,
    RedisSaver.from_conn_string(DB_URI) as checkpointer,
):
    store.setup()
    checkpointer.setup()

需要的安装命令：

# PostgreSQL
pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres langgraph-store-postgres

# Redis
pip install -U langgraph langgraph-checkpoint-redis langgraph-store-redis

注意：

• 首次使用 Postgres 存储时，您需要调用 store.setup()
• 首次使用 Redis 存储时，您需要调用 store.setup(), checkpointer.setup()

【代码仓库】https://github.com/monkeyhlj/LLM_development_learning
【参考】https://docs.langchain.com/
【参考】https://www.bilibili.com/video/BV1Uhz4BWEEw