构建react Ai代理：从零开始使用deepseek处理内存与工具的5个实用技巧

Rifx.Online
Large Language Models , AI Applications , Best Practices
23 Feb, 2025

AI代理在自动化中的应用

AI代理在自动化复杂任务和解决需要推理和外部数据检索的多步骤问题中发挥着至关重要的作用。它们正在改变涉及实时访问信息、多源数据分析和智能决策的过程，使自己成为现代AI驱动的解决方案的核心。

创建者：ReAct代理图

上面的图展示了我们将构建的ReAct（推理+行动）代理的架构。这个逐步指南提供了从从零开始创建代理所需的一切，涵盖了提示工程、工具集成和内存管理——使用Python和DeepSeek，而不依赖于LangChain或大型语言模型的功能调用能力。

为了展示结果，我们将把代理与FastAPI集成，以处理请求，并构建一个交互式Streamlit用户界面。这个界面将允许我们可视化代理的响应，包括它的思考、行动和观察。

以下是最终用户界面的预览：

创建者：ReAct代理的Streamlit用户界面

该实现的完整代码库可以在GitHub上找到。在本指南中，我将带您了解代码中最重要的部分，解释关键组件及其背后的推理。我还将包括参考文献和链接以证明这些选择，并在必要时提供额外的背景信息。

尽管该实现专注于DeepSeek，但我也会强调在使用其他大型语言模型时，特别是在实现令牌管理方面，可能会有所不同。

在本文结束时，您将对ReAct代理的操作和如何从零开始构建一个代理有深刻理解。下面是一个简短的介绍，随后是一个目录，以指导您完成实现过程。

什么是 ReAct AI 代理？

在其核心，传统AI系统旨在处理输入、分析数据，并根据预定义逻辑生成响应或执行操作。这些系统通常遵循线性决策过程，其中推理和执行在不同阶段中发生。虽然对于简单任务有效，但这种方法在处理需要持续调整和外部交互的复杂、多步骤问题时显得力不从心。

ReAct AI代理通过结合逻辑推理与动态行动执行来解决这个问题。该方法在2022年的论文 “ReAct: Synergizing Reasoning and Acting in Language Models” 中提出，增强了代理适应和智能响应其环境的能力。与线性执行操作的传统代理不同，ReAct代理在连续循环的思考 → 行动 → 观察中运行。这个迭代过程使它们能够分析输入、选择并使用工具、逐步执行任务，并根据实时反馈来改进其方法。

框架的抽象

像LangChain和LlamaIndex这样的框架提供了高级工具，简化了开发AI代理的关键方面。它们使得原型设计和部署代理变得更容易，而无需处理所有底层细节。然而，这些便利是以抽象的代价为代价的——隐藏了驱动代理决策和执行过程的基本机制。通过从零开始构建AI代理，我们获得了对其逻辑和交互的完全控制，使我们能够深刻理解它是如何推理、采取行动以及适应现实世界任务的。

DeepSeek的崛起

在最近几周，DeepSeek因其提供的LLMs而受到广泛关注，以其出色的性能、经济实惠的价格、开源特性和创新的训练方法而受到认可。

DeepSeek平台目前提供两个模型选项：

DeepSeek-Chat，使用DeepSeek-V3模型
DeepSeek-Reasoner，使用DeepSeek-R1模型

这两个模型之间的关键区别在于它们的目的和专业化。DeepSeek-Chat是一个多用途通用模型，旨在处理广泛的任务，使其可与GPT-4o等模型相媲美。相反，DeepSeek-Reasoner专门优化用于复杂推理，在高级问题解决、编程和数学证明方面表现出色，使其在高水平推理和分析能力上成为OpenAI o1的竞争对手。

LLM的知识截止日期

虽然DeepSeek的在线平台允许用户启用**“搜索”功能以检索最新信息（如ChatGPT所做），但其模型仍然存在知识截止**，这意味着它们对某个时间点之后的事件缺乏意识。

例如，如果我们问DeepSeek（未启用平台提供的搜索功能）：“谁赢得了2024年一级方程式世界锦标赛？”它将无法回答：

来自DeepSeek平台的截图：在未启用搜索功能的情况下询问DeepSeek有关最近事件

这一限制突显了根据我们希望构建的AI代理类型，集成外部工具进行信息检索的重要性。

为什么不使用从零开始构建的 ReAct 代理的 DeepSeek-R1？

由于我们正在从零开始构建一个 ReAct 代理并自行处理 提示、工具 和内存，我们将不使用 DeepSeek-R1。这已经是一个 推理模型，通过 系统提示 — 一组定义人工智能在所有交互中应如何行为的指令 — 来修改其逻辑并不推荐。正如官方文档所述：“避免添加系统提示；所有指令应包含在用户提示中。”

此外，DeepSeek-R1 在撰写时不支持 函数调用，这使得通过结构化请求直接与外部工具集成变得困难。对于那些不熟悉该术语的人来说，函数调用是一个内部 API 功能，允许模型通过生成结构化函数请求与外部工具进行交互。

使用DeepSeek-V3

相反，我们将应用提示工程来构建DeepSeek-V3的ReAct代理，完全从零开始使用工具和内存管理。

由于我们不使用函数调用能力，我们将手动管理所有外部工具交互。这将通过基于提示的控制来完成。

虽然使用像LangChain这样的框架或利用函数调用能力将简化流程并使工具集成更容易，但本文的目标是深入理解AI代理的核心机制。

实施的目录：

1. 代理结构和执行流程概述

要理解我们的AI代理如何运作，首先让我们拆解其核心组件和执行流程。从根本上讲，代理遵循一个结构化的循环，包括输入处理、推理、行动执行和观察处理。

让我们用一个复杂的查询来说明这个过程，该查询需要推理和外部知识检索：

❓ “2024年金球奖得主出生的城市天气如何？”

乍一看，这个问题需要多个步骤来回答，而一个知识截止于2024年前的标准LLM将难以提供完整的响应。以下是代理如何逻辑性地分解问题的步骤：

确定2024年金球奖得主 → 由于LLM缺乏最新知识，它首先需要进行一次网页搜索以获取得主信息。
找到得主的出生地 → 确定了球员的名字后，代理可以使用维基百科、网页搜索或其自身的知识来确定他出生的地方。
获取当前天气 → 最后，使用天气API，代理可以查找该城市的实时天气数据并返回最终答案。

这种结构化的多步骤推理过程是ReAct代理的基础，使其能够分解复杂问题并确定何时需要外部工具。

创建者：ReAct代理

上面的第一个图表概述了ReAct AI代理的核心组件和执行流程。它在一个循环中运行，代理处理用户输入，确定是否需要外部信息，与工具互动，并根据观察来细化其响应。关键组件包括：

用户输入 → 用户向代理提供查询或请求。

DeepSeek LLM → 处理请求的核心推理组件，确定最佳行动方案并生成响应。

工具 → 代理可以选择在需要时调用外部工具，如虚线箭头所示。如果采取了行动，则调用工具与环境互动，检索必要的信息。一旦工具提供结果，观察结果将返回给LLM。

环境 → 这包括代理可以查询的外部数据源，我们将使用：

网页搜索（Tavily API） → 用于检索超出LLM知识截止的实时信息。
维基百科 → 用于访问维基百科的内容。
天气（OpenWeather API） → 用于获取实时天气数据。

尽管没有明确说明，代理在这个第一个图表中存储所有消息、推理步骤和工具操作在内存中。这使得代理能够在不同用户输入之间保持上下文，更像一个聊天机器人。然而，尽管代理保留了过去的交互，但它并没有有效地管理内存——没有 令牌使用管理、摘要 _或删除旧消息。

为了解决这个限制，实现的第7节将引入一个内存管理 组件，确保更有效地处理长对话。这个补充将使代理能够总结过去的消息并根据令牌使用情况删除旧消息，防止过度的内存积累，同时保留重要的上下文。为此，我们将整合消息摘要，利用LLM并利用Transformers库进行令牌计数。（如果您使用GPT作为您的LLM，我还将提供一个关于如何使用tiktoken库的链接）。

以下图表展示了扩展的架构，包含内存管理。

创建者：ReAct代理与内存管理组件 — 在第7节实施

⚠️ 注意： 此实现不包括持久内存。这意味着一旦会话结束，所有存储的消息将丢失。如果我们希望在不同会话之间保持内存，则应实施持久存储方法（例如使用数据库存储消息）。

2. 环境与工具：开始代码

要开始，我们需要定义代理将要操作的环境。为了创建一个可扩展的环境，我们将构建一个名为 BaseTool 的抽象类，并定义一个名为 use 的抽象方法。

任何扩展此类的工具都将定义两个关键属性：name 和 description，这有助于代理理解工具的目的和作用。这些工具还将实现 use 方法，以定义每个工具操作的具体执行逻辑。

以下是 BaseTool 类的实现：

from abc import ABC, abstractmethod

class BaseTool(ABC):
    """所有工具的抽象基类"""

    def __init__(self, name, description):
        self._name = name.lower()
        self._description = description

    @property
    def name(self):
        """工具名称的获取器。"""
        return self._name

    @property
    def description(self):
        """工具描述的获取器。"""
        return self._description

    @abstractmethod
    def use(self, query):
        """每个工具必须实现自己的 `use` 方法"""
        pass

如前所述，在此代理实现中，我们将有三个工具：维基百科API、网页搜索和天气API。每个工具的完整实现可以在 Agent/AgentTools 文件夹中找到。为了展示一个示例，以下是 Weather 类的实现。该类利用 OpenWeather API 并实现 use 方法以获取实时天气数据。

class Weather(BaseTool):
    def __init__(self):
        load_dotenv()
        super().__init__(
            name="weather",
            description="获取给定城市的天气信息。输入仅为城市名称，例如 'London'。"
        )
        self.api_key = os.getenv("OPENWEATHER_API_KEY")
        self.base_url = "http://api.openweathermap.org/data/2.5/weather"

    def use(self, query):
        """获取给定城市的天气数据"""
        if not query:
            raise ValueError("城市不能为空。")

        url = f"{self.base_url}?q={query}&appid={self.api_key}&units=metric"
        response = requests.get(url)

        if response.status_code == 200:
            data = response.json()
            temperature = data['main']['temp']
            description = data['weather'][0]['description']
            humidity = data['main']['humidity']
            wind_speed = data['wind']['speed']

            return (
                f"{query} 的温度为 {temperature}°C。"
                f"天气为 {description}。"
                f"湿度为 {humidity}%。"
                f"风速为 {wind_speed} m/s。"
            )

        return f"获取 {query} 的天气数据失败。"

创建新工具非常简单。只需定义一个新的类，扩展自 BaseTool，指定 name 和 description 属性，并在 use 方法中实现工具的具体行为。

现在环境已经设置完成，我们可以实现将利用这些工具的代理。

3. 定义代理类

现在我们将定义构成代理类的方法。其中一些方法作为辅助方法，基本上是自解释的，例如register_tool和add_message。

标记为第7节注释的行将在我们引入内存管理 组件时用于消息摘要。

此外，我们定义代理将使用的模型：deepseek-ai/DeepSeek-V3，通过使用API密钥访问OpenAI API，该密钥由DeepInfra提供。

我们在本文中的主要重点是详细解释驱动代理执行流程的关键组件：think、decide和act。这些方法将在下一节中进行详细解释。

class Agent:
    def __init__(self):
        load_dotenv()
        self.client = OpenAI(
            api_key=os.getenv("DEEPSEEK_API_KEY"),
            base_url="https://api.deepinfra.com/v1/openai"
        )
        self.model_name = "deepseek-ai/DeepSeek-V3"
        self.tools = {}
        self.messages = []
        self.max_iterations = 5
        self.current_iteration = 0
        self.system_prompt = self.load_prompt("Prompts/system_prompt.txt")
        self.old_chats_summary = ""        
        self.messages_to_summarize = 5    
        self.max_messages_tokens = 10000    
        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name, cache_dir="./tokenizer_cache") 
        self.summary_prompt = self.load_prompt("Prompts/summary_prompt.txt")

    def register_tool(self, tool):

    def get_tools(self):

    def add_message(self, role, content):

    def get_chat_history(self):

    def load_prompt(self, path):

    def summarize_old_chats(self, lines):

    def extract_first_queries(self, chat_history):

    def num_tokens_from_messages(self, messages):

    def think(self):

    def decide(self, response):

    def act(self, tool_name, query):

    def memory_management(self, chat_history):

    def call_DeepSeek(self, prompt):

    def execute(self, query):

在代理的代码中，有一个用于存储消息的数组。这些消息是定义在utils.py文件中的Message类的实例。每条消息有两个关键属性：role和content。

class Message:
    def __init__(self, role, content):
        self.role = role
        self.content = content

角色指示消息的来源，而内容存储消息文本。大型语言模型使用角色来维护上下文并管理结构化交互：**用户**发送查询，**助手**回复，**系统**为LLM建立行为和指导方针。

此外，我们创建的所有工具都注册在self.tools字典中。

4. 实现思考-行动-观察循环

在本节中，我们将分解构成ReAct代理核心的关键功能。这些方法结合我们将在第5节中设计的系统提示，集成大型语言模型以构建一个处理输入、做出决策并使用环境采取行动的推理代理。

有关代理类的完整代码，请参见 agent.py 文件。

执行

execute 方法是每个新用户查询的入口点。该方法启动过程，将用户的消息添加到聊天记录中，并调用 think 方法，后者启动 思考 → 行动 → 观察 循环。然后，它等待循环完成后再返回为该请求生成的所有消息。

def execute(self, query):
    """Execute a user query and return the full Agent response."""
    self.current_iteration = 0
    self.add_message("user", query)
    self.think()

    result_messages = []
    for message in self.messages[::-1]:
        if message.role == "user":
            break
        elif message.role != "user":
            result_messages.append(message)

    return result_messages[::-1]

思考与决定

在这个过程的这个阶段，我们处于循环内部，因此定义和监控最大迭代限制以防止无限循环是至关重要的。

think 方法使用系统提示调用 LLM 和 call_DeepSeek（这两者在第 5 节中解释）来引导代理的推理。一旦收到响应，它会作为助手消息添加到聊天记录中。然后调用 decide 方法来分析响应。

def think(self):
    """Think and decide based on the response from DeepSeek."""
    self.current_iteration += 1

    if self.current_iteration > self.max_iterations:
        print("Reached maximum iterations. Stopping.")
        self.add_message("assistant", "I'm sorry, but I couldn't find a satisfactory answer within the allowed number of iterations.")
        return

    current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    prompt = self.system_prompt.format(
        tools=self.get_tools(),
        date=current_date
    )

    response = self.call_DeepSeek(prompt)
    self.add_message("assistant", response)
    self.decide(response)

decide 方法用于解析 LLM 的响应并确定下一步。它检查响应中是否包含最终答案或与环境中的工具交互的行动请求。

如果有行动请求（例如 Action: weather: Paris），该方法提取工具名称和查询，并调用 act 方法来执行请求的操作。

def decide(self, response):
    """Decide on the next action based on the response."""
    final_answer_match = re.search(r"Final Answer:", response)
    if final_answer_match:
        return

    action_match = re.search(r"Action:\s*(\w+):\s*(.*)", response)
    if action_match:
        tool_name = action_match.group(1).strip().lower()
        query = action_match.group(2).strip()
        self.act(tool_name, query)

    if not final_answer_match and not action_match:
        print("No action or final answer found in the response.")

行动与观察

act 方法在 LLM 请求时执行工具。如第 2 节所述，工具从抽象类扩展，并定义一个执行实际操作的 use 方法。

在此方法中，我们检查请求的工具是否存在于 self.tools 中。如果存在，我们使用 LLM 提供的查询执行工具的 use 方法。结果被格式化为观察，并且以角色 tool 的消息被添加到聊天记录中。之后，循环通过再次调用 think 继续。

def act(self, tool_name, query):
    """Act on the response by calling the appropriate tool."""
    tool = self.tools.get(tool_name)

    if tool:
        result = tool.use(query)
        observation = f"Observation from {tool_name}: {result}"
        self.add_message("tool", observation)
        self.think()
    else:
        print(f"No tool registered for choice: {tool_name}")
        self.add_message("system", f"Error: Tool {tool_name} not found")

⚠️ 关于观察角色的重要说明：

当从工具返回观察时，消息中使用的角色会显著影响 LLM 如何处理响应。如果您使用的是 GPT-4o 或 Gemini Pro 1.5，这里使用 system 角色是安全的，代理将正常工作。然而，对于 DeepSeek 和 Llama 3.3，重用 system 角色会导致问题，因为聊天记录中的多个系统消息可能会触发意外行为，例如助手重新生成观察消息。

另一种选择是使用 assistant 角色，但在我的测试中，这通常会导致幻觉，例如在观察中引入不正确的数据，因为模型在响应同一角色的先前消息时会感到困惑。

为了使其在 DeepSeek 中正常工作，最佳选择是使用 tool 或 user 角色。通常，tool 角色用于函数调用（请参见 DeepSeek API 函数调用指南），并需要一个 tool_call_id 来识别工具调用。由于我们是从零开始构建代理，而不是使用内置的函数调用功能，因此我们生成一个占位符 tool_call_id 来绕过此要求，以便在向 LLM 发送消息时使用。

以下是它在 get_chat_history 方法中的处理方式，该方法为 LLM 提供消息。如果角色是 tool，它会生成一个随机的 tool_call_id 以匹配 API 预期的结构：

def get_chat_history(self):
    """Return the chat history as a list of message dictionaries."""
    return [
        {
            "role": message.role,
            "content": message.content,
            **({"tool_call_id": random.randint(1, 1000)} if message.role == "tool" else {})
        }
        for message in self.messages
    ]

5. 使用少量示例定义系统提示

现在我们已经实现了代理的推理循环，是时候创建一个系统提示，指示大型语言模型如何构建其响应并管理工作流程。提示需要明确代理应如何响应，以便我们可以稍后解析其回复并确定是否需要行动。此外，我们还将包括特定规则，使代理在某些情况下更像一个聊天机器人。

我们将使用少量提示，其中包含示例以指导代理的响应。另一种方法是零样本提示，不提供任何示例。如果您对零样本 ReAct 代理示例感兴趣，请查看 Arun Shankar 的这篇 Medium 文章。

使用少量提示的系统提示：

您运行一个思考、行动、暂停、观察的循环。
在循环结束时，您输出一个答案。
使用思考来描述您对所问问题的想法。
使用行动来执行您可用的其中一个行动 - 然后返回暂停。
观察将是运行这些行动的结果。

您可用的行动是：

{tools}

规则：

如果输入是问候或告别，请直接以友好的方式回应，而不使用思考-行动循环。
否则，遵循思考-行动循环以找到最佳答案。
如果您已经对部分或整个问题有答案，请使用您的知识，而不依赖外部行动。
如果您需要执行多个行动，请在单独的调用中进行。
最后，提供最终答案。

供您参考，今天的日期是 {date}。

一些示例：

问题：Franco Colapinto 是谁？
思考：我应该在维基百科上查找 Franco Colapinto。
行动：wikipedia: Franco Colapinto
暂停

您将再次收到此内容：

维基百科的观察：Franco Colapinto 是一名阿根廷赛车手。

然后您输出：
最终答案：Franco Colapinto 是一名阿根廷赛车手。

问题：法国的首都是什么？
思考：我知道法国的首都为巴黎。

然后您输出：
最终答案：法国的首都为巴黎。

问题：里约热内卢的天气如何？
思考：我应该查找里约热内卢的天气。
行动：weather: Rio de Janeiro
暂停

您将再次收到此内容：

天气的观察：里约热内卢的温度为 26°C。天气多云。湿度为 88%。风速为 6.26 m/s。

然后您输出：
最终答案：里约热内卢的温度为 26°C。天气多云。

系统提示中的 {tools} 和 {date} 占位符在 think 方法中动态填充（第 4 节）。get_tools 方法生成工具内容，返回一个格式化的字符串，包含工具名称和描述，以向大型语言模型展示如何使用它们。

def get_tools(self):
    """返回可用工具的格式化字符串列表。"""
    return "\\n".join([f"{tool.name}: {tool.description}" for tool in self.tools.values()])

我们尚未实现 call_DeepSeek 方法，该方法对于处理与 DeepSeek 大型语言模型的交互至关重要，并且也在 think 步骤中使用。此方法将系统提示与完整的聊天记录一起发送到大型语言模型，以生成响应。通过包含聊天记录中的所有现有消息，大型语言模型获得完整上下文，确保它理解循环的当前阶段和对话——无论是处理新用户的查询还是来自先前行动的观察。

以下是该方法的代码：

def call_DeepSeek(self, prompt):
    """调用 DeepSeek API 以获取响应。"""
    chat_history = self.get_chat_history()

    messages = [{"role": "system", "content": prompt}] + chat_history

    response = self.client.chat.completions.create(
        model=self.model_name,
        messages=messages,
        max_tokens=500,
        temperature=0.5,
    )

    return response.choices[0].message.content.strip() if response.choices else "没有来自 DeepSeek 的响应"

6. 测试代理

现在我们不仅定义了代理的推理逻辑以及如何解析其响应以与环境互动，还制定了系统提示以控制响应的生成方式，是时候测试代理了。我们将问它一系列问题，以观察其行为并确保它在不同情况下的响应符合预期。

为此，我们将执行来自 de>agent.py 文件的脚本。

让我们从非常简单的开始，以测试推理是否正确生成：

❓ 第一个问题：“利昂内尔·梅西出生的国家的首都天气如何？”

用户：利昂内尔·梅西出生的国家的首都天气如何？
助手：

思考：利昂内尔·梅西出生在阿根廷，阿根廷的首都是布宜诺斯艾利斯。我应该检查布宜诺斯艾利斯的天气。
行动：天气：布宜诺斯艾利斯
暂停
工具：
来自天气的观察：布宜诺斯艾利斯的温度为22.53°C。天气为阴云密布。湿度为89%。风速为6.26 m/s。
助手：最终答案：布宜诺斯艾利斯的温度为22.53°C。天气为阴云密布。

在这个例子中，LLM 识别出利昂内尔·梅西出生在阿根廷，并知道布宜诺斯艾利斯是首都。它并没有搜索这些信息，而是直接使用 天气工具 检查布宜诺斯艾利斯的天气。这表明 代理利用其内部知识 处理常识，只有在必要时才使用外部工具来优化其推理过程。

❓ 第二个问题：“目前马德里和布宜诺斯艾利斯哪个更冷？”

用户：目前马德里和布宜诺斯艾利斯哪个更冷？
助手：

思考：要确定哪个城市更冷，我需要检查马德里和布宜诺斯艾利斯的当前天气。
行动：天气：马德里
暂停
工具：来自天气的观察：马德里的温度为8.27°C。天气为雾。湿度为95%。风速为1.79 m/s。
助手：
思考：现在我需要检查布宜诺斯艾利斯的当前天气，以便与马德里进行比较。
行动：天气：布宜诺斯艾利斯
暂停
工具：来自天气的观察：布宜诺斯艾利斯的温度为22.3°C。天气为零星云。湿度为89%。风速为6.26 m/s。
助手：
思考：比较温度，马德里为8.27°C，而布宜诺斯艾利斯为22.3°C。因此，目前马德里比布宜诺斯艾利斯冷。
最终答案：目前马德里比布宜诺斯艾利斯冷。马德里的温度为8.27°C，而布宜诺斯艾利斯为22.3°C。

在这里，目标是观察代理如何执行 比较推理，它必须从多个来源检索和结合数据。代理成功收集了两个城市的天气数据，比较观察结果，并逻辑地得出哪个城市更冷的结论。这表明代理能够超越事实检索，在提供连贯的响应之前分析和结合数据。

❓ 第三个问题：“2024年金球奖得主出生的城市天气如何？”

用户：2024年金球奖得主出生的城市天气如何？
助手：

思考：要找到2024年金球奖得主出生城市的天气，我首先需要确定2024年金球奖得主是谁及其出生地。我将搜索2024年金球奖得主。
行动：网页搜索：2024年金球奖得主
暂停
工具：来自网页搜索的观察：[{‘title’: “2024年金球奖得主：完整名单 - 体育画报”, ‘content’: “关注SI足球的2024年金球奖颁奖典礼的实时博客，包括新闻、更新、得主等。2024年金球奖提名人包括该运动最大的超级明星，包括皇家马德里的维尼修斯·儒尼奥尔、曼城的罗德里和巴萨的艾塔娜·博纳蒂，但只有少数人能带着奖项离开沙特剧院。罗德里（西班牙，曼城）是2024年男子金球奖得主。艾塔娜·博纳蒂（西班牙，巴萨）是2024年女子金球奖得主。卡洛·安切洛蒂（意大利，皇家马德里）是2024年男子约翰·克鲁伊夫奖得主。艾玛·海耶斯（英格兰，切尔西/美国国家队）是2024年女子约翰·克鲁伊夫奖得主。哈里·凯恩（英格兰，拜仁慕尼黑）和基利安·姆巴佩（法国，巴黎圣日耳曼/皇家马德里）是2024年格尔德·穆勒奖得主。”, ‘url’: ‘https:

代理正确地分解了步骤：首先，通过 网页搜索 确定了2024年金球奖得主罗德里。由于它已经知道罗德里出生在马德里，因此跳过了对 维基百科 或其他 网页搜索 的不必要调用。代理直接检查马德里的当前天气。

❓ 第四个问题：“2024年金球奖得主出生的城市天气如何？并告诉我布宜诺斯艾利斯的天气。”

用户：2024年金球奖得主出生的城市天气如何？并告诉我布宜诺斯艾利斯的天气。
助手：

思考：要回答这个问题，我需要找出2024年金球奖得主出生在哪里，然后检查该城市的天气。此外，我还需要检查布宜诺斯艾利斯的天气。让我们先找出2024年金球奖得主。
行动：网页搜索：2024年金球奖得主
暂停
工具：来自网页搜索的观察：[{‘title’: “2024年金球奖得主：完整名单 - 体育画报”, ‘content’: “关注SI足球的2024年金球奖颁奖典礼的实时博客，包括新闻、更新、得主等。2024年金球奖提名人包括该运动最大的超级明星，包括皇家马德里的维尼修斯·儒尼奥尔、曼城的罗德里和巴萨的艾塔娜·博纳蒂，但只有少数人能带着奖项离开沙特剧院。罗德里（西班牙，曼城）是2024年男子金球奖得主。艾塔娜·博纳蒂（西班牙，巴萨）是2024年女子金球奖得主。卡洛·安切洛蒂（意大利，皇家马德里）是2024年男子约翰·克鲁伊夫奖得主。艾玛·海耶斯（英格兰，切尔西/美国国家队）是2024年女子约翰·克鲁伊夫奖得主。哈里·凯恩（英格兰，拜仁慕尼黑）和基利安·姆巴佩（法国，巴黎圣日耳曼/皇家马德里）是2024年格尔德·穆勒奖得主。”, ‘url’: ‘https:

在这个最后的例子中，代理通过回答 多部分问题 展示了其潜力。它通过 网页搜索 正确识别了2024年金球奖得主，并检索了马德里的天气（罗德里出生的地方），然后轻松切换到检查布宜诺斯艾利斯的天气，而 没有偏离 原始查询。

这里突出的特点是代理如何管理 上下文，跟踪多个子任务并将结果结合成一个连贯的最终响应。这展示了它 成功处理复杂、分层查询 的能力，同时高效利用其外部工具。

7. 基于令牌使用的内存摘要

我们已经提到，当前的代理通过在每次迭代中将整个聊天记录发送给模型来维护过去输入的记忆。这种方法使代理能够跟进之前用户的查询，并给出相关的回应，确保多轮对话的连续性。

以下示例演示了这一点：

USER: 梅西是阿根廷人吗？
ASSISTANT:

思考：我知道利昂内尔·梅西来自阿根廷，所以我不需要使用行动来确认这一点。
最终答案：是的，利昂内尔·梅西来自阿根廷。

USER: 他踢足球吗？
ASSISTANT:

思考：我知道利昂内尔·梅西是一名职业足球运动员。
最终答案：是的，利昂内尔·梅西踢足球。

虽然通过发送整个聊天记录来维持上下文的方法在短时间交互中是有效的，但随着对话的增长，它变得越来越低效。

大型语言模型通过将输入拆分为令牌来处理输入，这些令牌随后用于生成响应。问题在于大型语言模型的有限上下文窗口，这指的是模型一次可以处理的最大令牌数。尽管较新的模型提供了更大的上下文窗口——例如，GPT-4o-mini和DeepSeek V3具有128k令牌的上下文窗口——但每增加一个令牌就会增加成本。尽管这些模型允许大量输入，但优化令牌使用对于提高效率和节省成本至关重要。

优化令牌使用的一些策略包括：

删除旧消息： 删除最旧的消息以释放上下文窗口中的空间。
总结整个对话： 通过创建摘要来减少对话的大小。
总结旧消息然后删除： 这种混合方法在保留关键上下文的同时节省令牌。

在这篇文章中，我们将实施第三种方法，这是最全面的。我们将监控聊天记录中的令牌计数，一旦它超过指定限制，就对旧消息进行摘要并将其删除。这将使代理能够保持重要的上下文，同时保持令牌使用的可控性。

理解令牌计数：大型语言模型如何对消息进行分词和计算长度

虽然在纯文本中计数令牌相对简单，但在处理基于消息的格式时，过程变得更加复杂。这些消息包括用于元数据的附加令牌，例如角色标签（“用户”，“助手”）和消息分隔符。总的令牌计数取决于模型使用的特定分词器，这意味着不同的LLM可能会根据其格式规则以不同方式对相同的消息进行分词。

在将内存管理组件集成到代理之前，我们将回顾一个令牌计数示例，以了解如何计算令牌使用情况以及原始内容和格式化消息之间的差异。

为了演示这一点，我在Agent/Utils中包含了一个名为test_token_count.py的文件，我将在下面进行解释。该文件展示了如何使用Transformers库对文本进行分词并计算令牌数量。

⚠️ 注意：

如果您使用的是另一个开源LLM，只需更改模型名称，如果分词器有聊天模板，它应该可以工作。
对于那些使用GPT模型的人，您可以查看OpenAI Cookbook中使用tiktoken库进行令牌计数的示例：如何使用tiktoken计数令牌。

比较内容与完整消息格式的令牌计数

我们将从定义分词器开始。在这个例子中，我们直接从Transformers库中导入它。您也可以从DeepSeek平台下载分词器并从那里导入（DeepSeek令牌使用文档）。

from transformers import AutoTokenizer

model_name = "deepseek-ai/DeepSeek-V3"
tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir="./tokenizer_cache")

⚠️ 注意： transformers库在最初加载分词器时可能需要几秒钟，特别是如果它第一次下载分词器文件。在这个例子中，我使用cache_dir参数缓存文件，以加快后续加载速度。用于GPT模型的tiktoken库通常更快。

接下来，让我们比较仅消息的_内容_与整个消息格式（包括如角色（system，user，assistant）的元数据）之间的令牌计数。

def num_tokens_from_content(messages):
    """返回消息列表中内容的总令牌数。"""
    return sum(len(tokenizer.encode(message["content"])) for message in messages if "content" in message)

def num_tokens_from_messages(messages):
    """返回消息列表使用的令牌数。"""
    value = tokenizer.apply_chat_template(messages, tokenize=False)
    print(f"使用模板的消息: {value}")
    encoded = tokenizer.encode(value)
    return len(encoded)

example_messages = [
    {"role": "system", "content": "你是一个有帮助的助手。"},
    {"role": "user", "content": "你好，你好吗？"},
    {"role": "assistant", "content": "我只是一个机器人，但我在这里帮助你！"},
    {"role": "user", "content": "你能做什么？"},
]

print(f"消息内容的令牌计数: {num_tokens_from_content(example_messages)}")
print(f"消息的令牌计数: {num_tokens_from_messages(example_messages)}")

在这个例子中，我们定义了两个函数：

num_tokens_from_content：该函数仅计算每条消息的内容中的令牌。
num_tokens_from_messages：该函数使用Transformers库中的apply_chat_template（聊天模板文档）将消息转换为模型所需格式的单个可分词字符串。

当我们执行代码时，我们可以看到令牌计数的差异：（我在这里也打印了消息如何使用模板格式化）

消息内容的令牌计数: 34
使用模板的消息: 
<|begin▁of▁sentence|\>你是一个有帮助的助手。<|User|\>你用的哪个版本的DeepSeek？<|Assistant|\>我只是一个机器人，但我在这里帮助你！<|end▁of▁sentence|\><|User|\>你能做什么？
消息的令牌计数: 36

通过API验证令牌计数

最后，当我们使用相同的消息调用模型并检查API响应中的prompt_tokens时，我们可以确认实际的令牌计数与我们使用聊天模板计算的结果一致。

chat_completion = openai.chat.completions.create(
    model="deepseek-ai/DeepSeek-V3",
    messages=example_messages
)
print(f"调用API时提示的令牌计数: {chat_completion.usage.prompt_tokens}")

调用API时提示的令牌计数: 36

这确认了在这些情况下，应用聊天模板进行令牌计数比仅计算消息内容更准确。

将内存摘要添加到代理

现在我们已经看到令牌计数是如何工作的，是时候将内存摘要集成到我们代理的工作流程中。如第1节的第二个图所示，我们将引入一个内存管理组件，以确保对话保持在指定的令牌限制内。一旦超过该限制，代理将总结旧消息并删除它们，以释放空间，同时保留关键上下文。

为了实现这一点，我们将修改Agent类，以在其构造函数中包含以下属性：

self.old_chats_summary = ""
self.max_messages_tokens = 10000 
self.messages_to_summarize = 5   
self.summary_prompt = self.load_system_prompt("Prompts/summary_prompt.txt")
self.tokenizer = AutoTokenizer.from_pretrained(self.model_name, cache_dir="./tokenizer_cache")

首先，我们将添加方法memory_management，该方法将监控对话历史的令牌计数。如果总数超过self.max_message_tokens所规定的令牌限制，它将总结并删除由self.messages_to_summarize指定的旧消息数量。

def memory_management(self, chat_history):
    """通过总结和删除旧聊天记录来管理内存"""
    try:
        user_messages = [msg for msg in chat_history if msg["role"] == "user"]
        if len(user_messages) > self.messages_to_summarize and self.num_tokens_from_messages(chat_history) > self.max_messages_tokens:
            indices = self.extract_first_queries(chat_history)
            if indices:
                start_index, end_index = indices
                lines = chat_history[start_index:end_index]
                print(f"对话中用于总结的令牌数: {self.num_tokens_from_messages(lines)}")
                new_summary = self.summarize_old_chats(lines)
                if new_summary != "No response from DeepSeek":
                    print(f"新摘要使用的令牌数: {self.num_tokens_from_text(new_summary)}")
                    self.old_chats_summary = f"{self.old_chats_summary} {new_summary}".strip()
                    del self.messages[start_index:end_index]
    except Exception as e:
        print(f"内存管理期间发生错误: {e}")

每次调用模型时——无论是新的查询还是在工具执行期间——call_DeepSeek现在都会请求memory_management来验证当前chat_history的令牌计数，使用num_tokens_from_messages函数。如果令牌计数超过限制，则代理使用summarize_old_chats方法总结旧的聊天交互。如果创建或更新了摘要，它将附加到系统提示中以提供上下文。

更新后的call_DeepSeek如下所示：

def call_DeepSeek(self, prompt):
    """调用DeepSeek API以获取响应。"""
    chat_history = self.get_chat_history()

    self.memory_management(chat_history)

    if self.old_chats_summary:
        prompt += f"\n\n旧消息摘要:\n{self.old_chats_summary}"

    messages = [{"role": "system", "content": prompt}] + chat_history

    response = self.client.chat.completions.create(
        model=self.model_name,
        messages=messages,
        max_tokens=500,
        temperature=0.5,
    )

    return response.choices[0].message.content.strip() if response.choices else "No response from DeepSeek"

用于检查令牌计数的方法与我们之前用于测试的方法相同：

def num_tokens_from_messages(self, messages): 
    """返回一系列消息使用的令牌数。"""
    value = self.tokenizer.apply_chat_template(messages, tokenize=False)
    encoded = self.tokenizer.encode(value)
    return len(encoded)

extract_first_queries方法查找第一个用户消息并提取由self.messages_to_summarize指定的连续用户消息数量，包括它们之间的所有消息。

def extract_first_queries(self, chat_history):
    """从给定聊天历史中提取指定数量的连续用户查询。"""
    user_indices = [i for i, msg in enumerate(chat_history) if msg["role"] == "user"]

    start_index = user_indices[0]
    end_index = user_indices[self.messages_to_summarize]

    return start_index, end_index

为了避免总结正在进行的消息，这些消息可能包括不完整的工具调用，我们仅提取完整的对话段落（从一个user消息到下一个user消息）。

summarize_old_chats方法调用LLM并使用摘要提示生成提取对话的简明摘要。

def summarize_old_chats(self, lines):
    """总结旧聊天记录并返回简明的摘要响应。"""
    prompt = self.summary_prompt.format(lines=lines)

    response = self.client.chat.completions.create(
        model=self.model_name,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=120,
    )

    return response.choices[0].message.content.strip() if response.choices else "No response from DeepSeek"

⚠️ 注意： 如果您使用的是GPT，您可以为此调用使用系统角色，DeepSeek至少需要一个用户消息。

以下是用于摘要的提示示例：

您是一个AI助手，负责总结用户与助手之间的对话，助手具备访问系统操作的能力，如维基百科搜索、网页搜索和天气搜索。
给定以下对话，创建一个简明的摘要，清晰概述用户的问题或请求，并提供助手响应的简要概述。突出关键点，包括用户的请求和助手的具体最终答案。
确保摘要清晰、简洁，并以一个简短段落书写。

聊天记录：
{lines}

需要注意的一个关键点是，摘要是累积的——每个新摘要都添加到之前的摘要中而不合并旧摘要到新摘要中。然而，摘要提示可以修改为将旧摘要与新提取的消息合并。这种其他方法就是Langchain的ConversationSummaryBufferMemory的工作原理。

如果您更喜欢这种行为，可以调整摘要提示，以包含现有摘要作为上下文，并指示模型将其与新内容合并。

为了查看令牌使用的差异，让我们尝试以下输入：

❓ “2024年F1世界冠军出生的城市天气如何？”

以下是代理处理此查询的方式：

[
    {'role': 'user', 'content': '2024年F1世界冠军出生的城市天气如何？'},
    {'role': 'assistant', 'content': '思考: 根据搜索结果，马克斯·维斯塔潘是2024年的F1世界冠军。我需要找到他出生的城市。\n\n行动: wikipedia: 马克斯·维斯塔潘\n暂停'},
    {'role': 'tool', 'content': '来自维基百科的观察: ', 'tool_call_id': 965},
    {'role': 'assistant', 'content': '思考: 马克斯·维斯塔潘出生在比利时哈瑟尔。我现在将查看哈瑟尔的天气。\n\n行动: weather: 哈瑟尔\n暂停'},
    {'role': 'tool', 'content': '来自天气的观察: 哈瑟尔的温度为-0.69°C。天气为阴云。湿度为84%。风速为1.34 m/s。', 'tool_call_id': 496},
    {'role': 'assistant', 'content': '最终答案: 2024年F1世界冠军马克斯·维斯塔潘出生的城市哈瑟尔的温度为-0.69°C。天气为阴云。'}
]

对于整个交互，这代表了回答单个用户查询的完整推理过程，令牌消耗为1397个令牌。

对话中用于总结的令牌数: 1397

如果将self.max_message_tokens设置为1000，并将self.messages_to_summarize设置为1，代理将自动总结消息：

用户询问2024年F1世界冠军出生的城市天气如何。
助手确定马克斯·维斯塔潘赢得了2024年F1冠军，识别出他的出生地为比利时哈瑟尔，并提供了那里的天气情况。
最终答案: 哈瑟尔的温度为-0.69°C，天气为阴云。

这个总结的对话现在消耗：78个令牌

新摘要使用的令牌数: 78

通过引入这个内存管理组件，代理在处理长对话时变得更加高效。它防止了上下文窗口被旧消息淹没，同时确保通过简洁的摘要保留必要的信息。

8. 集成 FastAPI 和 Streamlit

现在我们已经完成了代理并且其完全功能正常，下一步是创建一个简单的 FastAPI 聊天端点。这将允许我们通过 Streamlit用户界面 与代理进行交互。

FastAPI 是一个现代化的网络框架，用于快速高效地构建 API，在我们的案例中，它将作为后端来提供代理的响应。而 Streamlit 是一个开源的 Python 库，旨在为数据科学和机器学习项目创建 交互式网络应用。

Streamlit 的优势 在于其轻量级特性、易于设置和适合 快速原型开发。它允许我们用最少的代码构建一个交互式聊天界面。

FastAPI 代码可以在以下位置找到：

Agent/app.py

Streamlit 用户界面在以下位置定义：

chat_ui.py

Streamlit 界面包含一个 输入框 用于用户查询，一个 聊天显示 显示代理的最终答案，以及一个 侧边栏 显示代理的最后推理链（思考、行动和观察）。

下面是显示一些聊天输入和代理最后推理链的用户界面：

这完成了我们关于从零开始使用 DeepSeek 构建 ReAct 代理的逐步指南。我们涵盖了从提示工程和工具集成到内存管理的所有内容，赋予您 对代理逻辑和执行流程的完全控制。无论您是使用 DeepSeek 还是其他大型语言模型，过程都保持灵活并可适应您的项目。

正如我之前所澄清的，完整的实现可以在 GitHub 上找到。