上下文工程是指在给人工智能分配任务之前，为其创建合适的设置。该设置包括：

• 关于人工智能应如何行动的说明，例如成为一名乐于助人的经济型旅行指南
• 从数据库、文档或实时来源获取有用信息。
• 记住过往对话，避免重复或遗忘。
• 人工智能可以使用的工具，如计算器或搜索功能。
• 关于你的重要细节，比如你的偏好或位置。

上下文工程侧重于为人工智能提供正确的背景信息和工具，使其回答更智能、更有用。

在这篇文章中，我们将探讨LangChain和LangGraph这两个用于构建人工智能智能体、RAG应用程序和大语言模型（LLM）应用程序的强大工具，如何被用于有效地实施上下文工程以改进我们的人工智能智能体。

所有代码均可在这个GitHub代码库中获取：

https://github.com/FareedKhan-dev/contextual-engineering-guide?source=post_page-----0914d84601f3---------------------------------------

什么是上下文工程？

大语言模型（LLMs）的工作方式类似于一种新型操作系统。大语言模型就像中央处理器（CPU），其上下文窗口就像随机存取存储器（RAM），充当其短期记忆。但是，和随机存取存储器一样，上下文窗口对于不同信息的容纳空间有限。

就像操作系统决定哪些内容进入随机存取存储器（RAM）一样，“上下文工程”关乎选择大语言模型（LLM）应在其上下文中保留哪些内容。

在构建大语言模型（LLM）应用程序时，我们需要管理不同类型的上下文。上下文工程涵盖以下主要类型：

• 说明：提示词、示例、记忆内容和工具描述
• 知识：事实、存储的信息和记忆
• 工具：工具调用的反馈和结果

今年，更多人对智能体感兴趣，因为大语言模型在思考和使用工具方面表现更为出色。智能体通过结合使用大语言模型和工具来处理长期任务，并根据工具的反馈选择下一步行动。

但是，长时间的任务以及从工具中收集过多反馈会消耗大量的令牌。这可能会引发问题：上下文窗口可能溢出，成本和延迟可能增加，而且智能体的表现可能变差。

Anthropic在其研究中强调了对此的需求：

智能体之间的对话往往会有数百个轮次，因此谨慎管理上下文至关重要。

那么，如今人们是如何解决这个问题的呢？智能体上下文工程的常见策略可分为四大类：

• 编写：创建清晰有用的上下文
• 选择：只选取最相关的信息
• 压缩：缩短上下文以节省空间
• 隔离：将不同类型的上下文分开

LangGraph的构建旨在支持所有这些策略。我们将在LangGraph中逐一探讨这些组件，看看它们如何助力我们的智能体更好地工作。

使用LangGraph的草稿本

就像人类会做笔记，以便在后续任务中记住某些信息一样，智能体也可以使用暂存区做同样的事情。它将信息存储在上下文窗口之外，这样智能体在需要时就可以访问这些信息。

一个很好的例子是Anthropic多智能体研究员：

首席研究员规划其方法并将其保存到内存中，因为如果上下文窗口超过200,000个标记，它就会被截断，所以保存计划可确保不会丢失。

暂存区可以通过不同的方式实现：

• 作为一种工具调用，它写入文件。
• 作为运行时状态对象中的一个字段，在会话期间持续存在。

简而言之，暂存区可帮助智能体在会话期间记录重要信息，从而高效完成任务。

就语言图谱（LangGraph）而言，它同时支持短期（线程范围）和长期记忆。

• 短期记忆在会话期间使用检查点来保存智能体状态。它就像一个便签本，让你在智能体运行时存储信息，并在之后检索这些信息。

状态对象是在图节点之间传递的主要结构。你可以定义其格式（通常是Python字典）。它就像一个共享的便签本，每个节点都可以读取和更新特定的字段。

我们只会在需要的时候导入模块，这样我们就可以清晰地逐步学习。

为了获得更美观、更简洁的输出，我们将使用Python的pprint模块进行格式化打印，并使用rich库中的Console模块。首先让我们导入并初始化它们：

# 导入必要的库
from typing import TypedDict  # 用于通过类型提示定义状态模式
from rich.console import Console  # 用于美观的输出打印
from rich.pretty import pprint  # 用于Python对象的美观打印

# 初始化一个控制台，用于在笔记本中进行丰富的格式化输出
console = Console()

接下来，我们将为状态对象创建一个TypedDict。

# 使用TypedDict定义图的状态结构
class State(TypedDict):
    """
    定义我们的笑话生成工作流的状态结构。
    属性：
        topic：将为其生成笑话的输入主题。
        joke：将存储生成的笑话的输出字段。
    """
    topic: str
    joke: str

这个状态对象将存储主题，以及我们要求智能体根据给定主题生成的笑话。

创建状态图

一旦我们定义了一个状态对象，就可以使用状态图向其写入上下文。

状态图（StateGraph）是LangGraph用于构建有状态的智能体或工作流的主要工具。可以将其视为一个有向图：

• 节点是工作流程中的步骤。每个节点将当前状态作为输入，对其进行更新，并返回所做的更改。
• 边连接节点，定义执行如何流动，这可以是线性的、有条件的，甚至是循环的。

接下来，我们将：

1. 从Anthropic模型中选择，创建一个聊天模型。

# 导入环境管理、显示和LangGraph所需的必要库
import getpass
import os
from IPython.display import Image, display
from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, StateGraph

# --- 环境和模型设置 ---
# 设置Anthropic API密钥以进行身份验证
from dotenv import load_dotenv
api_key = os.getenv("ANTHROPIC_API_KEY")
ifnot api_key:
    raise ValueError("环境中缺少ANTHROPIC_API_KEY")

# 初始化将在工作流中使用的聊天模型
# 我们使用特定的Claude模型，设置temperature=0以获得确定性输出
llm = init_chat_model("anthropic:claude-sonnet-4-20250514", temperature=0)

我们已经初始化了我们的Sonnet模型。LangChain通过API支持许多开源和闭源模型，所以你可以使用其中任何一个。

现在，我们需要创建一个函数，使用这个十四行诗模型生成一个回复。

# --- 定义工作流节点 ---
def generate_joke(state: State) -> dict[str, str]:
    """
    基于当前状态中的主题生成笑话的节点函数。
    该函数从状态中读取“topic”，使用LLM生成笑话，并返回一个字典以更新状态中的“joke”字段。
    参数：
        state：图的当前状态，必须包含“topic”。
    返回：
        一个带有“joke”键的字典，用于更新状态。
    """
    # 从状态中读取主题
    topic = state["topic"]
    print(f"正在生成关于以下主题的笑话：{topic}")
    
    # 调用语言模型生成笑话
    msg = llm.invoke(f"写一个关于{topic}的短笑话")
    
    # 返回生成的笑话，以便写回状态
    return {"joke": msg.content}

这个函数只是返回一个包含生成的回复（即笑话）的字典。

现在，使用状态图（StateGraph），我们可以轻松构建并编译该图。接下来我们就这么做。

# --- 构建并编译图 ---
# 使用预定义的State模式初始化一个新的StateGraph
workflow = StateGraph(State)

# 将“generate_joke”函数添加为图中的一个节点
workflow.add_node("generate_joke", generate_joke)

# 定义工作流的执行路径：
# 图从START入口点开始，流向我们的“generate_joke”节点
workflow.add_edge(START, "generate_joke")

# “generate_joke”完成后，图的执行结束
workflow.add_edge("generate_joke", END)

# 将工作流编译为可执行的链
chain = workflow.compile()

# --- 可视化图 ---
# 显示编译后的工作流图的可视化表示
display(Image(chain.get_graph().draw_mermaid_png()))

现在我们可以执行这个工作流程了。

# --- 执行工作流 ---
# 使用包含主题的初始状态调用编译后的图
# `invoke`方法从START节点运行图到END节点
joke_generator_state = chain.invoke({"topic": "cats"})

# --- 显示最终状态 ---
# 打印执行后图的最终状态
# 这将显示输入的“topic”和写入状态的输出“joke”
console.print("\n[bold blue]笑话生成器状态：[/bold blue]")
pprint(joke_generator_state)

#### 输出 ####
{
'topic': 'cats',
'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'
}

它返回的字典基本上是我们智能体的笑话生成状态。这个简单的例子展示了我们如何将上下文写入状态。

你可以进一步了解用于保存和恢复图表状态的检查点，以及用于暂停工作流程以在继续之前获取人工输入的人工介入。

LangGraph中的记忆写入

暂存区有助于智能体在单个会话中开展工作，但有时智能体需要在多个会话间记住某些信息。

• 反思（Reflexion）提出了智能体在每一轮行动后进行反思并复用自身生成提示的概念。
• 生成式智能体通过总结过去的智能体反馈来创建长期记忆。

这些想法如今应用于诸如ChatGPT、Cursor和Windsurf等产品中，这些产品能根据用户交互自动创建长期记忆。

• 检查点在线程中的每一步保存图的状态。一个线程有唯一的ID，并且通常代表一次交互，比如ChatGPT中的一次单独聊天。
• 长期记忆使你能够在不同线程间保留特定上下文。你可以保存单个文件（例如，用户配置文件）或记忆集合。
• 它使用BaseStore接口，这是一种键值存储。你可以在内存中使用它（如这里所示），也可以与LangGraph平台部署一起使用。

现在，让我们创建一个InMemoryStore，以便在本笔记本的多个会话中使用。

from langgraph.store.memory import InMemoryStore

# --- 初始化长期记忆存储 ---
# 创建InMemoryStore的实例，它提供了一个简单的、非持久化的、
# 键值存储系统，用于当前会话内使用
store = InMemoryStore()

# --- 定义用于组织的命名空间 ---
# 命名空间用于在存储中对相关数据进行逻辑分组
# 这里，我们使用元组来表示分层命名空间，
# 它可以对应于用户ID和应用程序上下文
namespace = ("rlm", "joke_generator")

# --- 向记忆存储写入数据 ---
# 使用`put`方法将键值对保存到指定的命名空间中
# 此操作会持久化上一步生成的笑话，使其
# 可在不同的会话或线程中检索
store.put(
    namespace,  # 要写入的命名空间
    "last_joke",  # 数据条目的键
    {"joke": joke_generator_state["joke"]},  # 要存储的值
)

在接下来的部分，我们将讨论如何从命名空间中选择上下文。目前，我们可以使用search方法来查看命名空间中的项目，并确认我们已成功写入其中。

# 搜索命名空间以查看所有存储的项目
stored_items = list(store.search(namespace))

# 使用丰富的格式显示存储的项目
console.print("\n[bold green]记忆中存储的项目：[/bold green]")
pprint(stored_items)

#### 输出 ####
[
  Item(namespace=['rlm', 'joke_generator'], key='last_joke',
   value={'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'},
  created_at='2025-07-24T02:12:25.936238+00:00',
  updated_at='2025-07-24T02:12:25.936238+00:00', score=None)
]

现在，让我们将所做的一切嵌入到LangGraph工作流程中。

我们将使用两个参数来编译工作流程：

• checkpointer在线程中保存每一步的图状态。
• store可在不同线程间保留上下文。

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore

# 初始化存储组件
checkpointer = InMemorySaver()  # 用于线程级别的状态持久化
memory_store = InMemoryStore()  # 用于跨线程的记忆存储

def generate_joke(state: State, store: BaseStore) -> dict[str, str]:
    """具有记忆感知的笑话生成。
    
    这个增强版本在生成新笑话之前会检查记忆中是否有现有笑话。
    
    参数：
        state：包含主题的当前状态
        store：用于持久化上下文的记忆存储
        
    返回：
        包含生成的笑话的字典
    """
    # 检查记忆中是否有现有笑话
    existing_jokes = list(store.search(namespace))
    if existing_jokes:
        existing_joke = existing_jokes[0].value
        print(f"现有笑话：{existing_joke}")
    else:
        print("现有笑话：无现有笑话")
    
    # 基于主题生成新笑话
    msg = llm.invoke(f"写一个关于{state['topic']}的短笑话")
    
    # 将新笑话存储到长期记忆中
    store.put(namespace, "last_joke", {"joke": msg.content})
    
    # 返回要添加到状态中的笑话
    return {"joke": msg.content}

# 构建具有记忆功能的工作流
workflow = StateGraph(State)

# 添加具有记忆感知的笑话生成节点
workflow.add_node("generate_joke", generate_joke)

# 连接工作流组件
workflow.add_edge(START, "generate_joke")
workflow.add_edge("generate_joke", END)

# 使用检查点和记忆存储进行编译
chain = workflow.compile(checkpointer=checkpointer, store=memory_store)

太棒了！现在我们可以直接执行更新后的工作流程，并测试在启用记忆功能的情况下它是如何运行的。

# 使用基于线程的配置执行工作流
config = {"configurable": {"thread_id": "1"}}
joke_generator_state = chain.invoke({"topic": "cats"}, config)

# 使用丰富的格式显示工作流结果
console.print("\n[bold cyan]工作流结果（线程1）：[/bold cyan]")
pprint(joke_generator_state)

#### 输出 ####
现有笑话：无现有笑话
工作流结果（线程1）：
{
'topic': 'cats',
'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'
}

由于这是线程1，我们的人工智能智能体内存中没有存储现成的笑话，这正是我们对新线程的预期。

由于我们使用检查点工具编译了工作流程，现在我们可以查看图的最新状态。

# --- 检索并检查图状态 ---
# 使用`get_state`方法检索`config`中指定的线程（在本例中为线程“1”）的最新状态快照。
# 这之所以可能，是因为我们使用检查点工具编译了图。
latest_state = chain.get_state(config)

# --- 显示状态快照 ---
# 将检索到的状态打印到控制台。StateSnapshot不仅包括
# 数据（'topic'、'joke'），还包括执行元数据。
console.print("\n[bold magenta]最新图状态（线程1）：[/bold magenta]")
pprint(latest_state)

看看输出结果：

### 我们最新状态的输出 ###
最新图状态：
StateSnapshot(
    values={
        'topic': 'cats',
        'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'
    },
    next=(),
    config={
        'configurable': {
            'thread_id': '1',
            'checkpoint_ns': '',
            'checkpoint_id': '1f06833a-53a7-65a8-8001-548e412001c4'
        }
    },
    metadata={'source': 'loop', 'step': 1, 'parents': {}},
    created_at='2025-07-24T02:12:27.317802+00:00',
    parent_config={
        'configurable': {
            'thread_id': '1',
            'checkpoint_ns': '',
            'checkpoint_id': '1f06833a-4a50-6108-8000-245cde0c2411'
        }
    },
    tasks=(),
    interrupts=())

你可以看到，我们的状态现在显示了我们在这个案例中与智能体的最后一次对话，当时我们让它讲一个关于猫的笑话。

让我们用不同的ID重新运行该工作流程。

# 使用不同的线程ID执行工作流
config = {"configurable": {"thread_id": "2"}}
joke_generator_state = chain.invoke({"topic": "cats"}, config)

# 显示跨线程的记忆持久性结果
console.print("\n[bold yellow]工作流结果（线程2）：[/bold yellow]")
pprint(joke_generator_state)

#### 输出 ####
现有笑话：{'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'}
工作流结果（线程2）：
{'topic': 'cats', 'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'}

我们可以看到，第一个线程中的笑话已成功保存到内存中。

暂存区选择方法

你如何从暂存区选择上下文取决于其实现方式：

• 如果它是一个工具，智能体可以通过调用工具直接读取它。
• 如果它是智能体运行时状态的一部分，你（开发者）需要决定在每一步向智能体共享哪些状态部分。这使您能够对所公开的上下文进行细粒度控制。

在上一步中，我们学习了如何写入LangGraph状态对象。现在，我们将学习如何从状态中选择上下文，并将其传递给下游节点中的大语言模型调用。

这种选择性方法使你能够精确控制大语言模型（LLM）在执行过程中看到的上下文。

def generate_joke(state: State) -> dict[str, str]:
    """基于主题生成初始笑话。
    
    参数：
        state：包含主题的当前状态
        
    返回：
        包含生成的笑话的字典
    """
    msg = llm.invoke(f"写一个关于{state['topic']}的短笑话")
    return {"joke": msg.content}

def improve_joke(state: State) -> dict[str, str]:
    """通过添加文字游戏来改进现有笑话。
    
    这展示了从状态中选择上下文——我们从状态中读取现有笑话，并使用它生成改进版本。
    
    参数：
        state：包含原始笑话的当前状态
        
    返回：
        包含改进后的笑话的字典
    """
    print(f"初始笑话：{state['joke']}")
    
    # 从状态中选择笑话以呈现给LLM
    msg = llm.invoke(f"通过添加文字游戏让这个笑话更有趣：{state['joke']}")
    return {"improved_joke": msg.content}

为了让事情变得更复杂一些，我们现在为智能体添加两个工作流程：

1. 像之前一样生成笑话。
2. 改进笑话：获取生成的笑话并对其进行优化。

此设置将帮助我们了解在LangGraph中暂存区选择是如何工作的。现在，让我们按照之前的方式编译此工作流程，并查看我们的图是什么样子。

# 构建具有两个顺序节点的工作流
workflow = StateGraph(State)

# 添加两个笑话生成节点
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)

# 按顺序连接节点
workflow.add_edge(START, "generate_joke")
workflow.add_edge("generate_joke", "improve_joke")
workflow.add_edge("improve_joke", END)

# 编译工作流
chain = workflow.compile()

# 显示工作流可视化
display(Image(chain.get_graph().draw_mermaid_png()))

当我们执行这个工作流程时，得到的就是这样的结果。

# 执行工作流以查看上下文选择的作用
joke_generator_state = chain.invoke({"topic": "cats"})

# 使用丰富的格式显示最终状态
console.print("\n[bold blue]最终工作流状态：[/bold blue]")
pprint(joke_generator_state)

#### 输出 ####
初始笑话：为什么猫加入了乐队？
因为它想成为“喵”克风手！
最终工作流状态：
{
'topic': 'cats',
'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'
}

既然我们已经执行了工作流程，就可以继续在记忆选择步骤中使用它。

记忆选择能力

如果智能体能够保存记忆，那么它们还需要为当前任务选择相关记忆。这在以下方面很有用：

• 情景记忆：展示期望行为的少样本示例。
• 程序性记忆：指导行为的指令。
• 语义记忆：提供与任务相关背景的事实或关系。

一些智能体使用范围有限的预定义文件来存储记忆：

• Claude Code使用CLAUDE.md。
• Cursor和Windsurf使用“规则”文件来获取指令或示例。

但当存储大量事实（语义记忆）的集合时，选择就变得更加困难。

• ChatGPT有时会检索到不相关的记忆，正如西蒙·威利森所展示的那样，ChatGPT错误地获取了他的位置，并将其注入到一张图片中，使得上下文感觉“不再属于他”。
• 为了改进筛选，嵌入或知识图谱被用于索引。

在上一节中，我们向图节点中的InMemoryStore进行写入操作。现在，我们可以使用get方法从其中选择上下文，以便将相关状态引入到我们的工作流程中。

from langgraph.store.memory import InMemoryStore

# 初始化记忆存储
store = InMemoryStore()

# 定义用于组织记忆的命名空间
namespace = ("rlm", "joke_generator")

# 将生成的笑话存储到记忆中
store.put(
    namespace,                             # 用于组织的命名空间
    "last_joke",                          # 键标识符
    {"joke": joke_generator_state["joke"]} # 要存储的值
)

# 从记忆中选择（检索）笑话
retrieved_joke = store.get(namespace, "last_joke").value

# 显示检索到的上下文
console.print("\n[bold green]从记忆中检索到的上下文：[/bold green]")
pprint(retrieved_joke)

#### 输出 ####
从记忆中检索到的上下文：
{'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'}

它成功地从记忆中检索出正确的笑话。

现在，我们需要编写一个合适的generate_joke函数，该函数能够：

1. 获取当前状态（用于暂存区上下文）。
2. 使用记忆（如果我们正在执行笑话改进任务，则获取过去的笑话）。

接下来我们把这个写成代码。

# 初始化存储组件
checkpointer = InMemorySaver()
memory_store = InMemoryStore()

def generate_joke(state: State, store: BaseStore) -> dict[str, str]:
    """具有记忆感知上下文选择的笑话生成。
    
    该函数展示了从记忆中选择上下文，然后生成新内容，确保一致性并避免重复。
    
    参数：
        state：包含主题的当前状态
        store：用于持久化上下文的记忆存储
        
    返回：
        包含生成的笑话的字典
    """
    # 如果存在，从记忆中选择先前的笑话
    prior_joke = store.get(namespace, "last_joke")
    if prior_joke:
        prior_joke_text = prior_joke.value["joke"]
        print(f"先前的笑话：{prior_joke_text}")
    else:
        print("先前的笑话：无！")
    
    # 生成一个与先前笑话不同的新笑话
    prompt = (
        f"写一个关于{state['topic']}的短笑话，"
        f"但要与你之前写过的任何笑话不同：{prior_joke_text if prior_joke else '无'}"
    )
    msg = llm.invoke(prompt)
    
    # 将新笑话存储到记忆中，用于未来的上下文选择
    store.put(namespace, "last_joke", {"joke": msg.content})
    return {"joke": msg.content}

现在，我们可以像之前一样简单地执行这个有记忆感知的工作流程。

# 构建具有记忆感知的工作流
workflow = StateGraph(State)
workflow.add_node("generate_joke", generate_joke)

# 连接工作流
workflow.add_edge(START, "generate_joke")
workflow.add_edge("generate_joke", END)

# 使用检查点和记忆存储进行编译
chain = workflow.compile(checkpointer=checkpointer, store=memory_store)

# 使用第一个线程执行工作流
config = {"configurable": {"thread_id": "1"}}
joke_generator_state = chain.invoke({"topic": "cats"}, config)

#### 输出 ####
先前的笑话：无！

未检测到之前的笑话，我们现在可以打印最新的状态结构。

# 获取图的最新状态
latest_state = chain.get_state(config)
console.print("\n[bold magenta]最新图状态：[/bold magenta]")
pprint(latest_state)

我们的输出：

#### 最新状态的输出 ####
最新图状态：
StateSnapshot(
    values={
        'topic': 'cats',
        'joke': "这是一个新的：\n\n为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！"
    },
    next=(),
    config={
        'configurable': {
            'thread_id': '1',
            'checkpoint_ns': '',
            'checkpoint_id': '1f068357-cc8d-68cb-8001-31f64daf7bb6'
        }
    },
    metadata={'source': 'loop', 'step': 1, 'parents': {}},
    created_at='2025-07-24T02:25:38.457825+00:00',
    parent_config={
        'configurable': {
            'thread_id': '1',
            'checkpoint_ns': '',
            'checkpoint_id': '1f068357-c459-6deb-8000-16ce383a5b6b'
        }
    },
    tasks=(),
    interrupts=())

我们从记忆中提取上一个笑话，并将其传递给大语言模型进行优化。

# 使用第二个线程执行工作流，以展示跨线程的记忆持久性
config = {"configurable": {"thread_id": "2"}}
joke_generator_state = chain.invoke({"topic": "cats"}, config)

#### 输出 ####
先前的笑话：这是一个新的：
为什么猫加入了乐队？
因为它想成为“喵”克风手！
工作流结果（线程2）：
{'topic': 'cats', 'joke': '为什么猫加入了乐队？\n\n因为它想成为“喵”克风手！'}

它已经成功地从记忆中提取出正确的笑话，并如预期那样对其进行了改进。

LangGraph大工具调用的优势

智能体使用工具，但给它们过多的工具可能会导致混淆，尤其是当工具描述重叠时。这使得模型更难选择正确的工具。

一种解决方案是对工具描述使用检索增强生成（RAG），以便根据语义相似性仅获取最相关的工具，德鲁·布罗伊尼格（Drew Breunig）将这种方法称为工具配置。

根据最近的研究，这将工具选择的准确率提高了多达3倍。

对于工具选择，LangGraph Bigtool库是理想之选。它通过对工具描述进行语义相似性搜索，为任务选择最相关的工具。它利用LangGraph的长期记忆存储，使智能体能够针对特定问题搜索并检索合适的工具。

让我们通过使用一个具备Python内置数学库所有功能的智能体来理解langgraph-bigtool。

import math
import types

# 从`math`内置库收集函数
all_tools = []
for function_name in dir(math):
    function = getattr(math, function_name)
    ifnot isinstance(
        function, types.BuiltinFunctionType
    ):
        continue
    # 这是`math`库的一个特性
    if tool := convert_positional_only_function_to_tool(
        function
    ):
        all_tools.append(tool)

我们首先将Python的math模块中的所有函数追加到一个列表中。接下来，我们需要将这些工具描述转换为向量嵌入，以便智能体可以执行语义相似性搜索。

为此，我们将使用一个嵌入模型，在我们的例子中，即OpenAI文本嵌入模型。

import uuid

# 创建工具注册表。这是一个将标识符映射到工具实例的字典
tool_registry = {
    str(uuid.uuid4()): tool
    for tool in all_tools
}

# 在LangGraph存储中索引工具名称和描述。这里我们使用一个简单的内存存储
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 1536,
        "fields": ["description"],
    }
)

for tool_id, tool in tool_registry.items():
    store.put(
        ("tools",),
        tool_id,
        {
            "description": f"{tool.name}: {tool.description}",
        },
    )

每个函数都被赋予一个唯一的ID，并且我们将这些函数组织成一种合适的标准化格式。这种结构化格式确保了函数能够轻松转换为嵌入向量，以便进行语义搜索。

现在，让我们将智能体可视化，看看嵌入所有数学函数并准备好进行语义搜索后它是什么样子！

# 初始化智能体
builder = create_agent(llm, tool_registry)
agent = builder.compile(store=store)
agent

现在，我们可以用一个简单的查询来调用智能体，并观察我们的工具调用智能体如何选择并使用最相关的数学函数来回答问题。

# 导入用于格式化和显示消息的实用函数
from utils import format_messages

# 定义智能体的查询
# 这个查询要求智能体使用其数学工具之一来求反余弦
query = "使用可用工具计算0.5的反余弦。"

# 用查询调用智能体。智能体将搜索其工具，
# 根据查询的语义选择'acos'工具，并执行它
result = agent.invoke({"messages": query})

# 格式化并显示智能体执行后的最终消息
format_messages(result['messages'])
┌────────────── 人类   ───────────────┐
│ 使用可用工具计算0.5的反余弦。        │
└──────────────────────────────────────┘
┌────────────── 📝 AI ─────────────────┐
│ 我将搜索一个工具来计算0.5的反余弦。   │
│                                      │
│ 🔧 工具调用：retrieve_tools         │
│ 参数：{                              │
│   "query": "反余弦 arccos 反余弦 三角"│
│ }                                    │
└──────────────────────────────────────┘
┌────────────── 🔧 工具输出 ────────┐
│ 可用工具：['acos', 'acosh']         │
└──────────────────────────────────────┘
┌────────────── 📝 AI ─────────────────┐
│ 太好了！我找到了`acos`函数，它可以计算反余弦。│
│ 现在我将用它来计算0.5的反余弦。     │
│                                      │
│ 🔧 工具调用：acos                   │
│ 参数：{ "x": 0.5 }                   │
└──────────────────────────────────────┘
┌────────────── 🔧 工具输出 ────────┐
│ 1.0471975511965976                   │
└──────────────────────────────────────┘
┌────────────── 📝 AI ─────────────────┐
│ 0.5的反余弦约为**1.047**弧度。       │
│                                      │
│ ✔ 检查：cos(π/3)=0.5，π/3≈1.047弧度（60°）。│
└──────────────────────────────────────┘

带上下文工程的检索增强生成

RAG（检索增强生成）是一个广泛的主题，代码智能体是生产中智能体RAG的一些最佳示例。

在实践中，检索增强生成（RAG）往往是上下文工程的核心挑战。正如Windsurf的瓦伦（Varun）所指出的：

索引编制≠上下文检索。基于抽象语法树（AST）分块的嵌入搜索是可行的，但随着代码库的增长会失效。我们需要混合检索：即通过grep命令/文件搜索、知识图谱链接以及基于相关性的重排序。

我们将首先使用WebBaseLoader工具提取页面内容。

# 导入WebBaseLoader工具以从URL获取文档
from langchain_community.document_loaders import WebBaseLoader

# 定义莉莲·翁博客文章的URL列表
urls = [
    "https://lilianweng.github.io/posts/2025-05-01-thinking/",
    "https://lilianweng.github.io/posts/2024-11-28-reward-hacking/",
    "https://lilianweng.github.io/posts/2024-07-07-hallucination/",
    "https://lilianweng.github.io/posts/2024-04-12-diffusion-video/",
]

# 使用列表推导式从每个URL加载文档。WebBaseLoader为每个URL返回一个文档列表，
# 因此我们有一个列表的列表。这个推导式将它们组合成一个单一的列表
docs = [WebBaseLoader(url).load() for url in urls]

对于检索增强生成（RAG），有不同的数据分块方法，而恰当的分块对于有效检索至关重要。

在这里，我们将在把获取到的文档索引到向量数据库之前，将其分割成较小的块。我们将采用一种简单直接的方法，比如使用带有重叠片段的递归分块，这样既能在各块之间保留上下文信息，又能使这些块在进行嵌入和检索时易于处理。

# 导入用于分块文档的文本分割器
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 展平文档列表。WebBaseLoader为每个URL返回一个文档列表，
# 所以我们有一个列表的列表。这个推导式将它们组合成一个单一的列表
docs_list = [item for sublist in docs for item in sublist]

# 初始化文本分割器。这将把文档分割成指定大小的较小块，
# 块之间有一些重叠以保持上下文
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=2000, chunk_overlap=50
)

# 将文档分割成块
doc_splits = text_splitter.split_documents(docs_list)

既然我们已经有了拆分后的文档，就可以将它们索引到向量数据库中，以便用于语义搜索。

# 导入用于创建内存向量存储的必要类
from langchain_core.vectorstores import InMemoryVectorStore

# 从文档块创建内存向量存储
# 这使用上一个单元格中创建的'doc_splits'和之前初始化的'embeddings'模型
# 来创建文本块的向量表示
vectorstore = InMemoryVectorStore.from_documents(
    documents=doc_splits, embedding=embeddings
)

# 从向量存储创建检索器
# 检索器提供了一个接口，用于基于查询搜索相关文档
retriever = vectorstore.as_retriever()

我们必须创建一个检索器工具，以便在智能体中使用。

# 导入用于创建检索器工具的函数
from langchain.tools.retriever import create_retriever_tool

# 从向量存储检索器创建检索器工具
# 这个工具允许智能体基于查询从博客文章中搜索和检索相关文档
retriever_tool = create_retriever_tool(
    retriever,
    "retrieve_blog_posts",
    "搜索并返回关于莉莲·翁博客文章的信息。",
)

# 以下行是直接调用工具的示例
# 它被注释掉，因为它不需要用于智能体执行流程，但可能对测试有用
# retriever_tool.invoke({"query": "types of reward hacking"})

现在，我们可以实现一个能够从工具中选择上下文的智能体。

# 用工具增强LLM
tools = [retriever_tool]
tools_by_name = {tool.name: tool for tool in tools}
llm_with_tools = llm.bind_tools(tools)

对于基于检索增强生成（RAG）的解决方案，我们需要创建一个清晰的系统提示，以引导智能体的行为。这个提示相当于其核心指令集。

from langgraph.graph import MessagesState
from langchain_core.messages import SystemMessage, ToolMessage
from typing_extensions import Literal

rag_prompt = """你是一个 helpful 的助手，负责从莉莲·翁的一系列技术博客文章中检索信息。在使用检索工具收集上下文之前，先与用户明确研究范围。反思你获取的任何上下文，并继续下去，直到你有足够的上下文来回答用户的研究请求。"""

接下来，我们定义图的节点。我们需要两个主要节点：

1. llm_call：这是我们智能体的核心。它接收当前的对话历史（用户查询 + 之前的工具输出）。然后决定下一步，调用工具或生成最终答案。
2. tool_node：这是我们智能体的行动部分。它执行llm_call请求的工具调用。它将工具的结果返回给智能体。

# --- 定义智能体节点 ---
def llm_call(state: MessagesState):
    """LLM决定是调用工具还是生成最终答案。"""
    # 将系统提示添加到当前消息状态
    messages_with_prompt = [SystemMessage(content=rag_prompt)] + state["messages"]
    
    # 用增强的消息列表调用LLM
    response = llm_with_tools.invoke(messages_with_prompt)
    
    # 返回LLM的响应以添加到状态
    return {"messages": [response]}

def tool_node(state: dict):
    """执行工具调用并返回观察结果。"""
    # 获取最后一条消息，它应该包含工具调用
    last_message = state["messages"][-1]
    
    # 执行每个工具调用并收集结果
    result = []
    for tool_call in last_message.tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=str(observation), tool_call_id=tool_call["id"]))
    
    # 将工具的输出作为消息返回
    return {"messages": result}

我们需要一种方法来控制智能体的流程，决定它应该调用工具还是已经完成。

为了处理这个问题，我们将创建一个名为should_continue的条件边函数。

• 该函数检查LLM的最后一条消息是否包含工具调用。
• 如果包含，图将路由到tool_node。
• 如果不包含，执行结束。

# --- 定义条件边 ---
def should_continue(state: MessagesState) -> Literal["Action", END]:
    """基于LLM是否进行了工具调用来决定下一步。"""
    last_message = state["messages"][-1]
    
    # 如果LLM进行了工具调用，路由到tool_node
    if last_message.tool_calls:
        return "Action"
    # 否则，结束工作流
    return END

现在，我们只需构建工作流程并编译图表。

# 构建工作流
agent_builder = StateGraph(MessagesState)

# 添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("environment", tool_node)

# 添加边以连接节点
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    {
        # should_continue返回的名称：要访问的下一个节点的名称
        "Action": "environment",
        END: END,
    },
)
agent_builder.add_edge("environment", "llm_call")

# 编译智能体
agent = agent_builder.compile()

# 显示智能体工作流
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

该图表展示了一个清晰的循环：

1. 智能体启动，调用大语言模型。
2. 基于大语言模型的决策，它要么执行一个动作（调用我们的检索工具）并循环返回，要么完成并给出答案。

让我们测试一下我们的检索增强生成（RAG）智能体。我们会问它一个关于**“奖励破解”**的具体问题，这个问题只能通过从我们编入索引的博客文章中检索信息来回答。

# 定义用户的查询
query = "博客中讨论了哪些类型的奖励破解？"

# 用查询调用智能体
result = agent.invoke({"messages": [("user", query)]})

# --- 显示最终消息 ---
# 格式化并打印对话流程
format_messages(result['messages'])
┌────────────── 人类  ───────────────┐
│ 明确范围：我想要莉莲·翁关于强化学习博客中│
│ 的奖励破解类型。                    │
└──────────────────────────────────────┘
┌────────────── 📝 AI ─────────────────┐
│ 从她的文章中获取上下文...           │
└──────────────────────────────────────┘
┌────────────── 🔧 工具输出 ────────┐
│ 她在强化学习中列出了3种主要的奖励破解类型：│
└──────────────────────────────────────┘
┌────────────── 📝 AI ─────────────────┐
│ 1.** 规则博弈 **– 利用奖励漏洞，而非真正目标。│
│                                      │
│ 2.** 奖励篡改 **– 更改或破解奖励信号。     │
│                                      │
│ 3.** 导线头 **– 自我刺激奖励而非完成任务。  │
└──────────────────────────────────────┘
┌────────────── 📝 AI ─────────────────┐
│ 这些可能会导致强化学习智能体产生有害的、非预期的行为。│
└──────────────────────────────────────┘

如你所见，智能体正确识别出需要使用检索工具。随后，它成功从博客文章中检索到相关上下文，并利用这些信息给出了详细且准确的答案。

这是一个很好的例子，展示了通过检索增强生成（RAG）进行上下文工程，如何创建强大且知识渊博的智能体。

知识型智能体的压缩策略

智能体交互可能会持续数百轮，并涉及大量令牌的工具调用。摘要提取是处理这种情况的常用方法。

例如：

• 当上下文窗口超过95%时，Claude Code会使用“自动压缩”功能，总结整个用户-智能体交互历史。
• 总结可以使用诸如递归或分层总结等策略来压缩智能体轨迹。

你也可以在特定节点添加摘要：

• 在调用大量令牌的工具（例如搜索工具）之后此处示例。
• 在智能体与智能体之间进行知识转移的边界处，认知功能在Devin中使用微调模型来实现这一点。

LangGraph是一个底层编排框架，让你能够完全掌控：

• 将你的智能体设计为一组节点。
• 在每个节点内显式定义逻辑。
• 在节点之间传递共享状态对象。

这使得以不同方式压缩上下文变得很容易。例如，你可以：

• 使用消息列表作为智能体状态。
• 使用内置工具对其进行总结。

我们将使用之前编写的基于RAG的相同工具调用智能体，并添加其对话历史的摘要。

首先，我们需要扩展图的状态，以包含最终总结的字段。

# 定义带有摘要字段的扩展状态
class State(MessagesState):
    """包含用于上下文压缩的摘要字段的扩展状态。"""
    summary: str

接下来，我们将定义一个专门用于总结的提示，并保留之前的检索增强生成（RAG）提示。

# 定义总结提示
summarization_prompt = """总结完整的聊天历史和所有工具反馈，概述用户所问的内容以及智能体所做的事情。"""

现在，我们将创建一个summary_node。

• 此节点将在智能体工作结束时触发，以生成整个交互过程的简要总结。
• llm_call和tool_node保持不变。

def summary_node(state: MessagesState) -> dict:
    """
    生成对话和工具交互的摘要。
    参数：
        state：图的当前状态，包含消息历史。
    返回：
        一个带有“summary”键和生成的摘要字符串作为值的字典，
        用于更新状态。
    """
    # 将总结系统提示添加到消息历史的前面
    messages = [SystemMessage(content=summarization_prompt)] + state["messages"]
    
    # 调用语言模型生成摘要
    result = llm.invoke(messages)
    
    # 返回要存储在状态的'summary'字段中的摘要
    return {"summary": result.content}

我们的条件边should_continue现在需要决定是调用一个工具还是前进到新的summary_node。

def should_continue(state: MessagesState) -> Literal["Action", "summary_node"]:
    """基于LLM是否进行了工具调用来确定下一步。"""
    last_message = state["messages"][-1]
    
    # 如果LLM进行了工具调用，执行它们
    if last_message.tool_calls:
        return "Action"
    # 否则，继续进行总结
    return "summary_node"

让我们在最后加上这个新的总结步骤来构建图表。

# 构建RAG智能体工作流
agent_builder = StateGraph(State)

# 向工作流添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("Action", tool_node)
agent_builder.add_node("summary_node", summary_node)

# 定义工作流边
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    {
        "Action": "Action",
        "summary_node": "summary_node",
    },
)
agent_builder.add_edge("Action", "llm_call")
agent_builder.add_edge("summary_node", END)

# 编译智能体
agent = agent_builder.compile()

# 显示智能体工作流
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

现在，让我们用一个需要获取大量上下文的查询来运行它。

from rich.markdown import Markdown

query = "根据博客，为什么强化学习能改进大语言模型的推理？"
result = agent.invoke({"messages": [("user", query)]})

# 向用户打印最终消息
format_message(result['messages'][-1])

# 打印生成的摘要
Markdown(result["summary"])

#### 输出 ####
用户询问了为什么强化学习（RL）能改进大语言模型的推理……

不错，但它使用了11.5万个令牌！你可以在此处查看完整的追踪信息。对于那些有大量令牌工具调用的智能体来说，这是一个常见的挑战。

一种更有效的方法是在上下文进入智能体的主暂存区_之前_对其进行压缩。让我们更新检索增强生成（RAG）智能体，以便实时总结工具调用输出。

首先，针对此特定任务的新提示：

tool_summarization_prompt = """将为你提供来自RAG系统的文档。总结这些文档，确保保留所有相关/必要的信息。你的目标只是将文档的大小（令牌）减小到更易于管理的大小。"""

接下来，我们将修改工具节点，以包含这个总结步骤。

def tool_node_with_summarization(state: dict):
    """执行工具调用，然后总结输出。"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        
        # 总结文档
        summary_msg = llm.invoke([
            SystemMessage(content=tool_summarization_prompt),
            ("user", str(observation))
        ])
        
        result.append(ToolMessage(content=summary_msg.content, tool_call_id=tool_call["id"]))
    return {"messages": result}

现在，我们的should_continue边可以简化，因为我们不再需要最后的summary_node了。

def should_continue(state: MessagesState) -> Literal["Action", END]:
    """决定我们是否应该继续循环或停止。"""
    if state["messages"][-1].tool_calls:
        return "Action"
    return END

让我们构建并编译这个效率更高的智能体。

# 构建工作流
agent_builder = StateGraph(MessagesState)

# 添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("Action", tool_node_with_summarization)

# 添加边以连接节点
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    {
        "Action": "Action",
        END: END,
    },
)
agent_builder.add_edge("Action", "llm_call")

# 编译智能体
agent = agent_builder.compile()

# 显示智能体
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

让我们运行相同的查询，看看有什么不同。

query = "根据博客，为什么强化学习能改进大语言模型的推理？"
result = agent.invoke({"messages": [("user", query)]})
format_messages(result['messages'])
┌────────────── 用户 ───────────────┐
│ 为什么强化学习能改进大语言模型的推理？根据博客？│
└───────────────────────────────────┘
┌────────────── 📝 AI ──────────────┐
│ 搜索莉莲·翁的博客，了解强化学习如何改进大语言模型的推理…… │
│                                   │
│ 🔧 工具调用：retrieve_blog_posts │
│ 参数：{                           │
│ "query": "强化学习用于大语言模型推理"  │
│ }                                │
└───────────────────────────────────┘
┌────────────── 🔧 工具输出 ─────┐
│ 莉莲·翁解释说，强化学习通过对每个推理步骤的奖励进行训练（基于过程的奖励模型）来帮助大语言模型推理。这引导模型逐步思考，提高连贯性和逻辑性。    │
└───────────────────────────────────┘
┌────────────── 📝 AI ──────────────┐
│ 强化学习通过奖励逐步推理来改进大语言模型的推理 │

使用子智能体架构隔离上下文

隔离上下文的一种常见方法是将其分散到各个子智能体中。OpenAI Swarm库就是为这种“关注点分离”而设计的，其中每个智能体通过自身的工具、指令和上下文窗口来管理特定的子任务。

Anthropic的多智能体研究工具表明，具有隔离上下文的多个智能体的表现比单个智能体高出90.2%，这是因为每个子智能体都专注于更具体的子任务。

子智能体在各自的上下文窗口中并行运行，同时探索问题的不同方面。

然而，多智能体系统存在一些挑战：

• 令牌使用量高得多（有时比单智能体聊天多15倍）。
• 需要精心设计提示词来规划子智能体的工作。
• 协调子智能体可能比较复杂。

LangGraph支持多智能体设置。一种常见的方法是监督式架构，Anthropic的多智能体研究工具也采用了这种架构。监督者将任务委派给子智能体，每个子智能体在自己的上下文窗口中运行。

让我们构建一个简单的监督者来管理两个智能体：

• math_expert（数学专家）处理数学计算。
• research_expert（研究专家）搜索并提供研究所得信息。

监督者将根据查询内容决定调用哪个专家，并在LangGraph工作流中协调他们的响应。

from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor

# --- 为每个智能体定义工具 ---
def add(a: float, b: float) -> float:
    """将两个数字相加。"""
    return a + b

def multiply(a: float, b: float) -> float:
    """将两个数字相乘。"""
    return a * b

def web_search(query: str) -> str:
    """模拟网络搜索功能，返回FAANG公司的员工数量。"""
    return (
        "以下是2024年每家FAANG公司的员工数量：\n"
        "1. **Facebook（Meta）**：67,317名员工。\n"
        "2. **Apple（苹果）**：164,000名员工。\n"
        "3. **Amazon（亚马逊）**：1,551,000名员工。\n"
        "4. **Netflix（奈飞）**：14,000名员工。\n"
        "5. **Google（Alphabet）**：181,269名员工。"
    )

现在我们可以创建专门的智能体和管理它们的监督者了。

# --- 创建具有隔离上下文的专门智能体 ---
math_agent = create_react_agent(
    model=llm,
    tools=[add, multiply],
    name="math_expert",
    prompt="你是一名数学专家。每次只使用一种工具。"
)

research_agent = create_react_agent(
    model=llm,
    tools=[web_search],
    name="research_expert",
    prompt="你是一名世界级的研究员，可以使用网络搜索。不进行任何数学运算。"
)

# --- 创建用于协调智能体的监督者工作流 ---
workflow = create_supervisor(
    [research_agent, math_agent],
    model=llm,
    prompt=(
        "你是管理一名研究专家和一名数学专家的团队监督者。"
        "将任务委派给合适的智能体以回答用户的查询。"
        "对于时事或事实性内容，使用research_agent（研究专家）。"
        "对于数学问题，使用math_agent（数学专家）。"
    )
)

# 编译多智能体应用程序
app = workflow.compile()

让我们执行这个工作流，看看监督者是如何委派任务的。

# --- 执行多智能体工作流 ---
result = app.invoke({
    "messages": [
        {
            "role": "user",
            "content": "2024年FAANG公司的员工总数是多少？"
        }
    ]
})

# 格式化并显示结果
format_messages(result['messages'])
┌────────────── 用户 ───────────────┐
│ 了解更多关于LangGraph Swarm以及多智能体系统的信息。          │
└───────────────────────────────────┘

┌────────────── 📝 AI ──────────────┐
│ 正在获取关于LangGraph Swarm及相关资源的详细信息……    │
└───────────────────────────────────┘

┌────────────── 🔧 工具输出 ─────┐
│ **LangGraph Swarm**               │
│ 代码库：                             │
│ https://github.com/langchain-ai/  │
│ langgraph-swarm-py                │
│                                   │
│ • 用于具有动态协作功能的多智能体AI的Python库。  │
│ • 智能体根据专业分工移交控制权，同时保留对话上下文。  │
│ • 支持自定义移交、流式传输、记忆功能和人工参与。  │
│ • 安装方法：                        │
│   `pip install langgraph-swarm`   │
└───────────────────────────────────┘

┌────────────── 🔧 工具输出 ─────┐
│ **关于多智能体系统的视频** │
│ 1. https://youtu.be/4nZl32FwU-o   │
│ 2. https://youtu.be/JeyDrn1dSUQ   │
│ 3. https://youtu.be/B_0TNuYi56w   │
└───────────────────────────────────┘

┌────────────── 📝 AI ──────────────┐
│ LangGraph Swarm使构建具有上下文感知的多智能体系统变得容易。查看视频以深入了解多智能体行为。  │
└───────────────────────────────────┘

在这里，监督者正确地为每个任务隔离了上下文，将研究查询发送给研究员，将数学问题发送给数学家，展示了有效的上下文隔离。

使用沙盒环境进行隔离

HuggingFace的深度研究工具展示了一种很棒的上下文隔离方法。大多数智能体使用工具调用API，这些API返回JSON参数来运行像搜索API这样的工具并获取结果。

HuggingFace使用CodeAgent（代码智能体）来编写调用工具的代码。这些代码在安全的沙盒中运行，运行代码得到的结果会发送回大语言模型。

这将大量数据（如图像或音频）保持在大语言模型的令牌限制之外。HuggingFace解释道：

[代码智能体允许]更好地处理状态……需要为以后存储这个图像/音频/其他内容吗？只需将其作为变量保存在你的状态中，以便以后使用。

在LangGraph中使用沙盒很简单。LangChain沙盒使用Pyodide（编译为WebAssembly的Python）安全地运行不可信的Python代码。你可以将其作为工具添加到任何LangGraph智能体中。

from langchain_sandbox import PyodideSandboxTool
from langgraph.prebuilt import create_react_agent

# 创建具有网络访问权限的沙盒工具，用于安装包
tool = PyodideSandboxTool(allow_net=True)

# 使用沙盒工具创建ReAct智能体
agent = create_react_agent(llm, tools=[tool])

# 使用沙盒执行数学查询
result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "5加7等于多少？"}]},
)

# 格式化并显示结果
format_messages(result['messages'])

LangGraph中的状态隔离

智能体的运行时状态对象是另一种很好的上下文隔离方式，类似于沙盒。你可以通过模式（如Pydantic模型）来设计这种状态，该模式具有不同的字段用于存储上下文。

例如，一个字段（如messages）在每一轮都展示给大语言模型，而其他字段则将信息隔离起来，直到需要时才使用。

LangGraph围绕状态对象构建，允许你创建自定义的状态模式，并在智能体的整个工作流中访问其字段。

例如，你可以将工具调用结果存储在特定的字段中，使它们在必要之前对大语言模型保持隐藏。你在这些笔记本中已经看到了许多这样的例子。

总结全文

让我们总结一下到目前为止所做的工作：

我们使用LangGraph StateGraph创建了用于短期记忆的“草稿本”和用于长期记忆的InMemoryStore，使我们的智能体能够存储和回忆信息。

我们展示了如何有选择地从智能体的状态和长期记忆中提取相关信息。这包括使用检索增强生成（RAG）来查找特定知识，以及使用langgraph-bigtool从众多选项中选择合适的工具。

为了管理长时间的对话和令牌密集型的工具输出，我们实施了总结功能。

我们展示了如何实时压缩RAG结果，使智能体更高效，并减少令牌使用量。

我们通过构建一个多智能体系统（其中监督者将任务委派给专门的子智能体）以及使用沙盒环境来运行代码，探索了如何将上下文分开以避免混淆。

所有这些技术都属于“上下文工程”的范畴，这是一种通过精心管理AI智能体的工作记忆（上下文）来改进它们的策略，使它们更高效、更准确，并且能够处理复杂的、长期运行的任务。

普通人如何抓住AI大模型的风口？

领取方式在文末

为什么要学习大模型？

目前AI大模型的技术岗位与能力培养随着人工智能技术的迅速发展和应用 ， 大模型作为其中的重要组成部分 ， 正逐渐成为推动人工智能发展的重要引擎 。大模型以其强大的数据处理和模式识别能力， 广泛应用于自然语言处理 、计算机视觉 、 智能推荐等领域 ，为各行各业带来了革命性的改变和机遇 。

目前，开源人工智能大模型已应用于医疗、政务、法律、汽车、娱乐、金融、互联网、教育、制造业、企业服务等多个场景，其中，应用于金融、企业服务、制造业和法律领域的大模型在本次调研中占比超过 30%。

随着AI大模型技术的迅速发展，相关岗位的需求也日益增加。大模型产业链催生了一批高薪新职业：

人工智能大潮已来，不加入就可能被淘汰。如果你是技术人，尤其是互联网从业者，现在就开始学习AI大模型技术，真的是给你的人生一个重要建议！

最后

只要你真心想学习AI大模型技术，这份精心整理的学习资料我愿意无偿分享给你，但是想学技术去乱搞的人别来找我！

在当前这个人工智能高速发展的时代，AI大模型正在深刻改变各行各业。我国对高水平AI人才的需求也日益增长，真正懂技术、能落地的人才依旧紧缺。我也希望通过这份资料，能够帮助更多有志于AI领域的朋友入门并深入学习。

真诚无偿分享！！！
vx扫描下方二维码即可
加上后会一个个给大家发
【附赠一节免费的直播讲座，技术大佬带你学习大模型的相关知识、学习思路、就业前景以及怎么结合当前的工作发展方向等，欢迎大家~】

大模型全套学习资料展示

自我们与MoPaaS魔泊云合作以来，我们不断打磨课程体系与技术内容，在细节上精益求精，同时在技术层面也新增了许多前沿且实用的内容，力求为大家带来更系统、更实战、更落地的大模型学习体验。

希望这份系统、实用的大模型学习路径，能够帮助你从零入门，进阶到实战，真正掌握AI时代的核心技能！

01 教学内容

• 从零到精通完整闭环：【基础理论 →RAG开发 → Agent设计 → 模型微调与私有化部署调→热门技术】5大模块，内容比传统教材更贴近企业实战！

• 大量真实项目案例： 带你亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

02适学人群

应届毕业生‌： 无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌： 非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能突破瓶颈： 传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

vx扫描下方二维码即可
【附赠一节免费的直播讲座，技术大佬带你学习大模型的相关知识、学习思路、就业前景以及怎么结合当前的工作发展方向等，欢迎大家~】

本教程比较珍贵，仅限大家自行学习，不要传播！更严禁商用！

03 入门到进阶学习路线图

大模型学习路线图，整体分为5个大的阶段：

04 视频和书籍PDF合集

从0到掌握主流大模型技术视频教程（涵盖模型训练、微调、RAG、LangChain、Agent开发等实战方向）

新手必备的大模型学习PDF书单来了！全是硬核知识，帮你少走弯路（不吹牛，真有用）

05 行业报告+白皮书合集

收集70+报告与白皮书，了解行业最新动态！

06 90+份面试题/经验

AI大模型岗位面试经验总结（谁学技术不是为了赚$呢，找个好的岗位很重要）

07 deepseek部署包+技巧大全

由于篇幅有限

只展示部分资料

并且还在持续更新中…

真诚无偿分享！！！
vx扫描下方二维码即可
加上后会一个个给大家发
【附赠一节免费的直播讲座，技术大佬带你学习大模型的相关知识、学习思路、就业前景以及怎么结合当前的工作发展方向等，欢迎大家~】