什么是 ‘Stateful Tool Calls’？如何让 Agent 在多次交互间记住 Tool 返回的中间结果？

Stateful Tool Calls：构建智能 Agent 的记忆与连贯性

在人工智能领域，Agent 的崛起正在改变我们与计算机交互的方式。一个智能 Agent 能够理解复杂指令，自主规划，并利用各种工具（Tools）来完成任务。然而，当任务变得多步骤、需要跨越多次交互时，我们常常会遇到一个核心挑战：Agent 如何记住它在之前步骤中获得的中间结果？这就是我们今天要深入探讨的“Stateful Tool Calls”——有状态的工具调用。

我们将从 Agent 和工具调用的基本概念开始，逐步剖析无状态调用的局限性，然后深入理解有状态工具调用的核心原理、实现策略、最佳实践及未来展望。

1. 理解 Agent 与工具调用的基石

在深入有状态工具调用之前，我们首先要明确一些基础概念。

什么是 AI Agent？

一个 AI Agent 可以被视为一个能够感知环境、进行思考、规划行动并执行任务的实体。它通常包含以下几个核心组件：

1. 感知器 (Perceptors)：接收来自环境的信息（例如，用户输入、API 响应）。
2. 规划器 (Planner)：基于感知到的信息和预设的目标，生成一系列行动步骤。这通常涉及到大型语言模型（LLM）的推理能力，用于分解复杂任务和选择合适的工具。
3. 记忆 (Memory)：存储过去的交互、学习到的知识或任务相关的数据。这是我们今天讨论的重点之一。
4. 执行器 (Actuator)：执行规划器生成的行动，这通常是通过调用外部工具或 API 来完成。

Agent 的核心优势在于其自主决策和规划能力，它能够动态地适应环境并利用各种资源来解决问题。

什么是工具调用 (Tool Calls)？

大型语言模型（LLM）虽然强大，但它们的知识是静态的，且不具备执行外部操作的能力。为了克服这些局限性，Agent 被赋予了调用外部“工具”的能力。一个工具可以是：

• 一个查询数据库的函数。
• 一个调用外部 API（如天气服务、股票行情、电子邮件发送）的接口。
• 一个执行特定计算（如代码解释器、数学计算器）的模块。
• 一个操作本地文件系统的函数。

当 LLM 识别到它需要外部信息或执行外部操作时，它会生成一个工具调用指令，指定要调用的工具名称和参数。Agent 框架会捕获这个指令，执行相应的工具函数，并将工具的返回结果反馈给 LLM，供其继续推理。

无状态工具调用的局限性

大多数简单的工具调用默认是“无状态”的。这意味着每一次工具调用都是独立的，它不依赖于之前的任何调用，也不会保留任何关于之前调用的信息。

例如，一个查询天气的工具，每次调用都只关心当前的城市和日期，不会记住上次查询的城市。这对于原子性的、独立的任务非常有效。

import requests
import json

class WeatherTool:
    def get_current_weather(self, location: str) -> str:
        """
        获取指定地点的当前天气信息。
        :param location: 城市名称，例如 "北京", "上海"。
        :return: JSON格式的天气信息。
        """
        print(f"Calling Weather API for location: {location}...")
        try:
            # 模拟API调用
            # response = requests.get(f"https://api.weather.com/current?q={location}&appid=YOUR_API_KEY")
            # data = response.json()

            # 模拟数据
            if location == "北京":
                data = {"location": "北京", "temperature": "25°C", "condition": "晴朗", "humidity": "60%"}
            elif location == "上海":
                data = {"location": "上海", "temperature": "28°C", "condition": "多云", "humidity": "75%"}
            else:
                return json.dumps({"error": f"无法获取 {location} 的天气信息。"})

            return json.dumps(data, ensure_ascii=False)
        except Exception as e:
            return json.dumps({"error": str(e)})

# 模拟Agent调用
weather_tool = WeatherTool()
print(weather_tool.get_current_weather("北京"))
# 再次调用，与上次调用无关
print(weather_tool.get_current_weather("上海"))

在这个例子中，get_current_weather 函数每次被调用时，它都独立地完成任务。它不关心之前是否查询过北京的天气，也不会记住北京的天气信息。这种无状态的特性对于许多简单任务是足够的。

2. 无状态工具调用的局限性：为什么我们需要进化

尽管无状态工具调用简单高效，但它们在处理复杂、多步骤任务时会暴露出严重的局限性。

考虑一个典型的电商购物场景，用户想要完成以下操作：

1. 搜索商品：用户说“我想买一件T恤”。
2. 筛选商品：用户说“要红色的，尺码L”。
3. 加入购物车：用户说“把这件加入购物车”。
4. 查看购物车：用户说“我的购物车里有什么？”
5. 修改购物车：用户说“把T恤的数量改为2”。
6. 结算：用户说“我要付款”。

如果我们的工具都是无状态的，Agent 会面临以下问题：

• 信息丢失：当用户说“要红色的，尺码L”时，Agent 需要知道用户之前搜索的是“T恤”。如果“搜索商品”工具是无状态的，它不会将搜索结果传递给“筛选商品”工具。Agent 必须在 LLM 的上下文中重新提及“T恤”，或者更糟的是，筛选工具无法知道它正在筛选什么。
• 重复信息传递与效率低下：为了让后续工具知道上下文，Agent 必须在每次工具调用时，将所有相关的历史信息（例如，之前搜索到的商品列表、当前购物车内容）都作为参数传递。这会导致：
  • LLM 上下文爆炸：LLM 的 prompt 会变得非常长，包含大量重复或冗余的信息，这会消耗更多的 token，增加推理成本，并可能超出模型的上下文窗口限制。
  • 推理效率降低：LLM 需要处理更多的文本，才能提取出关键信息，这会降低其推理速度和准确性。
  • 编程复杂性：Agent 的规划逻辑会变得复杂，它需要主动管理并传递这些状态信息。
• 用户体验差：Agent 显得“健忘”。用户可能会觉得 Agent 无法理解上下文，每次对话都像从头开始。例如，用户问“我的购物车里有什么？”，如果购物车工具是无状态的，它每次都返回一个空购物车，或者需要用户再次提供所有商品信息。
• 难以支持复杂流程：许多业务流程本质上就是有状态的，例如订单创建、表单填写、多步骤配置等。无状态工具强制 Agent 在每次交互中重建或重新获取所有状态，这几乎是不可行的。

特性	无状态工具调用	有状态工具调用
独立性	每次调用独立，不依赖前次调用结果。	每次调用可利用或修改前次调用创建的状态。
上下文	每次调用需完整上下文。	仅需传递变化或引用，状态在工具或外部管理。
记忆	无内建记忆，Agent 需在 LLM 上下文维护。	工具或外部系统持有记忆。
适用场景	简单、原子性任务（如天气查询、一次性计算）。	复杂、多步骤、会话式任务（如购物、数据分析）。
复杂性	工具本身简单，Agent 维护上下文复杂。	工具或状态管理层复杂，Agent 交互相对简化。
成本	LLM 上下文长，推理成本高。	LLM 上下文短，推理成本相对低。
用户体验	易显“健忘”，连贯性差。	连贯性好，更符合人类对话习惯。

为了克服这些挑战，我们需要一种机制，让 Agent 能够“记住”工具返回的中间结果，并在后续的工具调用中利用这些结果。这就是“Stateful Tool Calls”的核心价值所在。

3. 什么是 ‘Stateful Tool Calls’？核心概念解析

‘Stateful Tool Calls’，顾名思义，是指工具调用在多次交互中能够保持并利用先前的上下文或结果。它的核心思想是将任务执行过程中产生的“状态”结构化并持久化，而不是仅仅依赖于 LLM 的瞬时上下文。

核心思想：状态的结构化与持久化

当 Agent 启动一个复杂任务（例如，一个购物会话），它会创建一个与该任务相关的“状态”。这个状态可以是一个数据结构，包含任务的所有关键信息，比如：

• 当前用户是谁。
• 用户正在进行什么操作（例如，搜索商品、查看购物车）。
• 已搜索到的商品列表。
• 购物车中的商品及其数量。
• 订单的当前状态。
• 任何其他在任务执行过程中产生的中间数据。

这个状态不是作为原始文本混杂在 LLM 的 prompt 中，而是以结构化的方式存储在某个地方——可能是工具内部、Agent 框架层、或者外部的数据库/缓存中。当 Agent 需要调用工具时，它会传递一个引用（例如，一个会话 ID），工具可以通过这个引用来访问和更新这个状态。

优势总结：

1. 持久化中间结果：关键信息不会在每次工具调用后丢失，为后续步骤提供上下文。
2. 提升 Agent 的“记忆力”和连贯性：Agent 能够理解并执行多步骤任务，提供更流畅的用户体验。
3. 简化 Agent 的决策逻辑：LLM 的 prompt 可以更简洁，因为它不需要承载大量的历史状态信息。LLM 只需要关注当前的决策点，而具体的历史数据由状态管理机制处理。
4. 支持复杂、多阶段的业务流程：例如，一个复杂的表单填写过程，可以在多个工具调用中逐步收集和验证用户输入。
5. 降低成本：减少 LLM 上下文长度，从而降低 token 使用量和 API 调用成本。

现在，我们来看几种实现 Stateful Tool Calls 的具体策略。

4. 实现 ‘Stateful Tool Calls’ 的策略与模式

实现有状态工具调用有多种方法，每种方法都有其适用场景、优缺点和复杂性。我们将探讨以下四种主要策略：

策略一：在 Agent 的上下文（LLM Prompt）中维护状态

这是最直接也最容易实现的方法，尤其适用于原型开发和状态信息不多的场景。

原理：
将工具返回的中间结果或关键信息，直接作为纯文本或结构化文本（如 JSON 字符串），注入到后续的 LLM prompt 中。LLM 在每次推理时，都能看到这些历史状态信息。

优点：

• 实现简单：无需额外的状态管理层或数据库。
• 完全依赖 LLM：LLM 有机会“看到”所有的历史上下文，理论上可以做出更全面的决策。

缺点：

• 上下文窗口限制：随着任务的进行和状态的积累，prompt 会变得越来越长，很容易超出 LLM 的上下文窗口限制。
• 成本增加：更长的 prompt 意味着更多的 token 使用，从而增加 API 调用成本。
• 噪音干扰：大量的状态信息可能会稀释 LLM 的注意力，甚至干扰其推理能力，使其难以识别当前任务的真正重点。
• 结构化困难：纯文本形式的状态管理难以被 LLM 准确解析和利用，容易产生误解。

代码示例：模拟一个简单的购物 Agent，将购物车内容作为状态传递。

import json

class ShoppingCartAgent:
    def __init__(self, llm_mock_function):
        self.llm_mock = llm_mock_function
        self.conversation_history = []
        self.current_cart = {} # 初始购物车状态

    def _call_llm(self, prompt: str) -> str:
        """模拟LLM调用，实际会调用 OpenAI/Anthropic 等API"""
        print(f"n--- LLM Prompt ---")
        print(prompt)
        print(f"--- End LLM Prompt ---")
        return self.llm_mock(prompt)

    def _add_to_cart_tool(self, item_name: str, quantity: int) -> str:
        """工具：将商品加入购物车"""
        self.current_cart[item_name] = self.current_cart.get(item_name, 0) + quantity
        return f"已将 {quantity} 件 {item_name} 加入购物车。"

    def _view_cart_tool(self) -> str:
        """工具：查看购物车内容"""
        if not self.current_cart:
            return "购物车是空的。"
        cart_items = [f"{qty} 件 {item}" for item, qty in self.current_cart.items()]
        return "您的购物车包含：" + ", ".join(cart_items) + "。"

    def _process_user_input(self, user_input: str) -> str:
        # 将历史对话和当前购物车状态作为上下文传递给LLM
        context = "n".join(self.conversation_history)
        current_cart_json = json.dumps(self.current_cart, ensure_ascii=False)

        prompt = f"""
        你是一个购物助手。你需要理解用户的意图，并使用提供的工具来完成任务。
        当前购物车状态: {current_cart_json}

        可用工具:
        - add_to_cart(item_name: str, quantity: int): 将指定商品和数量加入购物车。
        - view_cart(): 查看购物车当前内容。

        当前对话历史:
        {context}

        用户: {user_input}
        助手:
        """

        llm_response = self._call_llm(prompt)
        self.conversation_history.append(f"用户: {user_input}")
        self.conversation_history.append(f"助手: {llm_response}")

        # 模拟LLM响应是工具调用指令
        if "CALL_TOOL:add_to_cart" in llm_response:
            try:
                parts = llm_response.split("CALL_TOOL:add_to_cart(")[1].split(")", 1)[0]
                item_name = parts.split(',')[0].strip().strip("'"")
                quantity = int(parts.split(',')[1].strip())
                tool_output = self._add_to_cart_tool(item_name, quantity)
                return tool_output
            except Exception as e:
                return f"工具调用失败: {e}"
        elif "CALL_TOOL:view_cart" in llm_response:
            tool_output = self._view_cart_tool()
            return tool_output
        else:
            return llm_response # LLM 直接回复

# 模拟 LLM 行为的函数
def mock_llm_for_prompt_state(prompt: str) -> str:
    if "将 2 件 衬衫 加入购物车" in prompt:
        return "CALL_TOOL:add_to_cart('衬衫', 2)"
    elif "将 1 件 裤子 加入购物车" in prompt:
        return "CALL_TOOL:add_to_cart('裤子', 1)"
    elif "查看购物车" in prompt:
        return "CALL_TOOL:view_cart()"
    elif "你的购物车包含" in prompt: # LLM 看到购物车内容后，可能会直接回复
        return "好的，我知道了。还有其他需要帮助的吗？"
    else:
        return "抱歉，我不明白您的意思。"

# 运行 Agent
agent_with_prompt_state = ShoppingCartAgent(mock_llm_for_prompt_state)

print("--- 场景1: 将商品加入购物车并查看 ---")
print(agent_with_prompt_state._process_user_input("我想买两件衬衫"))
print(agent_with_prompt_state._process_user_input("再加一条裤子"))
print(agent_with_prompt_state._process_user_input("我的购物车里有什么？"))

# 注意观察每次prompt中 "当前购物车状态" 的变化

在这个例子中，current_cart 是 Agent 内部维护的状态，但每次调用 _process_user_input 时，它都会被 JSON 序列化后，作为 当前购物车状态 字符串的一部分，注入到 LLM 的 prompt 中。这样，LLM 就能“看到”当前的购物车内容，并据此决定下一步操作。

策略二：工具内部维护状态 (Tool-Managed State)

这种策略将状态管理责任下放给工具本身。

原理：
工具被设计为有状态的。这意味着工具实例内部会有一个持久化的数据结构（例如，一个字典、一个列表或者一个数据库连接），用于存储其相关的状态。每次调用工具时，工具会根据其内部状态和新的输入进行操作。

优点：

• 状态与工具逻辑紧密耦合：封装性好，状态的生命周期和管理逻辑与工具的功能高度一致。
• 减轻 LLM 上下文负担：LLM 的 prompt 可以更简洁，因为它不需要知道工具的内部状态。它只需要知道调用哪个工具以及传递什么参数。
• 适用于特定业务流程的工具：例如，一个专门处理订单的工具，可以内部维护订单的 ID 和状态；一个文件操作工具可以维护当前打开的文件句柄。

缺点：

• 工具设计复杂度增加：工具的实现者需要考虑状态的初始化、更新、持久化和清理。
• 状态在工具之间难以共享：如果一个任务需要多个工具协作，并且它们需要共享某个状态，这种模式会比较困难。每个工具都只管理自己的状态。
• 需要明确的生命周期管理：工具实例何时创建、何时销毁、何时重置其状态，需要 Agent 框架或上层逻辑来协调。

代码示例：一个会话级别的购物车服务工具，模拟一个购物车服务。

import json
import uuid

# 假设这是一个在Agent外部运行的购物车服务（或一个有状态的类实例）
class StatefulShoppingCartService:
    def __init__(self):
        self.carts = {} # 存储所有会话的购物车 {session_id: {item_name: quantity}}
        print("StatefulShoppingCartService initialized.")

    def add_item(self, session_id: str, item_name: str, quantity: int) -> str:
        """
        向指定会话的购物车添加商品。
        :param session_id: 唯一的会话ID。
        :param item_name: 商品名称。
        :param quantity: 数量。
        :return: 操作结果描述。
        """
        if session_id not in self.carts:
            self.carts[session_id] = {}
        self.carts[session_id][item_name] = self.carts[session_id].get(item_name, 0) + quantity
        print(f"Service: Cart for {session_id} updated: {self.carts[session_id]}")
        return f"已将 {quantity} 件 {item_name} 加入您的购物车 (会话ID: {session_id})。"

    def view_cart(self, session_id: str) -> str:
        """
        查看指定会话的购物车内容。
        :param session_id: 唯一的会话ID。
        :return: 购物车内容描述。
        """
        cart = self.carts.get(session_id)
        if not cart:
            return f"您的购物车是空的 (会话ID: {session_id})。"

        cart_items = [f"{qty} 件 {item}" for item, qty in cart.items()]
        return f"您的购物车包含：{', '.join(cart_items)} (会话ID: {session_id})。"

    def clear_cart(self, session_id: str) -> str:
        """
        清空指定会话的购物车。
        :param session_id: 唯一的会话ID。
        :return: 操作结果描述。
        """
        if session_id in self.carts:
            del self.carts[session_id]
            print(f"Service: Cart for {session_id} cleared.")
            return f"您的购物车已清空 (会话ID: {session_id})。"
        return f"您的购物车已经是空的 (会话ID: {session_id})。"

class AgentWithToolManagedState:
    def __init__(self, llm_mock_function, shopping_service: StatefulShoppingCartService):
        self.llm_mock = llm_mock_function
        self.shopping_service = shopping_service
        self.session_id = str(uuid.uuid4()) # 为每个Agent实例生成一个会话ID
        self.conversation_history = []
        print(f"Agent initialized with session_id: {self.session_id}")

    def _call_llm(self, prompt: str) -> str:
        print(f"n--- LLM Prompt (Agent Session ID: {self.session_id}) ---")
        print(prompt)
        print(f"--- End LLM Prompt ---")
        return self.llm_mock(prompt, self.session_id) # LLM mock function now takes session_id

    def _process_user_input(self, user_input: str) -> str:
        context = "n".join(self.conversation_history)

        prompt = f"""
        你是一个购物助手。你需要理解用户的意图，并使用提供的工具来完成任务。
        当前会话ID: {self.session_id}

        可用工具:
        - add_item_to_cart(session_id: str, item_name: str, quantity: int): 向指定会话的购物车添加商品。
        - view_session_cart(session_id: str): 查看指定会话的购物车内容。
        - clear_session_cart(session_id: str): 清空指定会话的购物车。

        当前对话历史:
        {context}

        用户: {user_input}
        助手:
        """

        llm_response = self._call_llm(prompt)
        self.conversation_history.append(f"用户: {user_input}")
        self.conversation_history.append(f"助手: {llm_response}")

        # 模拟LLM响应是工具调用指令
        if "CALL_TOOL:add_item_to_cart" in llm_response:
            try:
                parts = llm_response.split("CALL_TOOL:add_item_to_cart(")[1].split(")", 1)[0]
                # 假设LLM正确传递了session_id
                _session_id = parts.split(',')[0].strip().strip("'"")
                item_name = parts.split(',')[1].strip().strip("'"")
                quantity = int(parts.split(',')[2].strip())
                tool_output = self.shopping_service.add_item(_session_id, item_name, quantity)
                return tool_output
            except Exception as e:
                return f"工具调用失败: {e}"
        elif "CALL_TOOL:view_session_cart" in llm_response:
            try:
                _session_id = llm_response.split("CALL_TOOL:view_session_cart(")[1].split(")", 1)[0].strip().strip("'"")
                tool_output = self.shopping_service.view_cart(_session_id)
                return tool_output
            except Exception as e:
                return f"工具调用失败: {e}"
        elif "CALL_TOOL:clear_session_cart" in llm_response:
            try:
                _session_id = llm_response.split("CALL_TOOL:clear_session_cart(")[1].split(")", 1)[0].strip().strip("'"")
                tool_output = self.shopping_service.clear_cart(_session_id)
                return tool_output
            except Exception as e:
                return f"工具调用失败: {e}"
        else:
            return llm_response

# 模拟 LLM 行为的函数 (现在需要知道session_id才能生成正确的工具调用)
def mock_llm_for_tool_state(prompt: str, session_id: str) -> str:
    if "两件衬衫" in prompt:
        return f"CALL_TOOL:add_item_to_cart('{session_id}', '衬衫', 2)"
    elif "一条裤子" in prompt:
        return f"CALL_TOOL:add_item_to_cart('{session_id}', '裤子', 1)"
    elif "查看购物车" in prompt:
        return f"CALL_TOOL:view_session_cart('{session_id}')"
    elif "清空购物车" in prompt:
        return f"CALL_TOOL:clear_session_cart('{session_id}')"
    else:
        return "抱歉，我不明白您的意思。"

# 运行 Agent
shopping_service = StatefulShoppingCartService()
agent_with_tool_state = AgentWithToolManagedState(mock_llm_for_tool_state, shopping_service)

print("n--- 场景2: 工具内部维护状态 ---")
print(agent_with_tool_state._process_user_input("我想买两件衬衫"))
print(agent_with_tool_state._process_user_input("再加一条裤子"))
print(agent_with_tool_state._process_user_input("我的购物车里有什么？"))
print(agent_with_tool_state._process_user_input("清空购物车"))
print(agent_with_tool_state._process_user_input("我的购物车里有什么？")) # 应该显示为空

# 模拟另一个会话，验证状态隔离
print("n--- 场景2.1: 另一个用户/会话 ---")
another_agent = AgentWithToolManagedState(mock_llm_for_tool_state, shopping_service)
print(another_agent._process_user_input("给我买一件夹克"))
print(another_agent._process_user_input("查看购物车"))
print(agent_with_tool_state._process_user_input("我的购物车里有什么？")) # 验证第一个Agent的购物车未受影响

在这个例子中，StatefulShoppingCartService 实例持有所有会话的购物车状态。每个 AgentWithToolManagedState 实例都有一个唯一的 session_id，并在调用工具时将其传递给 shopping_service。shopping_service 根据 session_id 存取相应的购物车状态。LLM 的 prompt 变得更短，因为它不需要知道 current_cart 的具体内容，只需要知道当前会话的 session_id。

策略三：Agent 框架层或外部存储维护状态 (Agent/External State Management)

这种策略将状态管理从工具内部进一步抽象出来，交给 Agent 框架层或一个独立的外部存储系统。

原理：
Agent 框架（或专门的状态管理器）负责维护会话或任务相关的状态。当 Agent 决定调用工具时，它会提供一个“会话 ID”或“上下文 ID”。工具在执行时，可以利用这个 ID 从 Agent 框架或外部存储（如 Redis、数据库、内存缓存）中获取或更新状态。

优点：

• 灵活且可扩展：状态可以独立于工具的生命周期进行管理，支持更复杂的持久化需求。
• 状态可以跨工具、跨会话共享：如果设计得当，不同工具甚至不同 Agent 实例可以访问和修改同一个会话状态。
• 减轻单个工具和 LLM 的负担：工具可以专注于其核心业务逻辑，而无需关心状态的持久化细节。LLM 的 prompt 依旧保持简洁。
• 支持复杂的事务和并发控制：外部存储系统通常提供这些高级功能。

缺点：

• 实现复杂度最高：需要额外的基础设施（数据库、缓存）和设计清晰的状态 Schema 及访问接口。
• 需要明确的状态管理策略：包括状态的创建、存取、更新、删除、过期策略等。

代码示例：使用简单的 Python 字典模拟一个外部状态存储，Agent 通过会话 ID 管理购物车。

import json
import uuid

# 模拟一个外部状态存储 (可以是 Redis, 数据库等)
# 实际应用中，这会是一个单独的服务或模块
class ExternalStateStore:
    def __init__(self):
        self.data = {} # {session_id: {state_key: value}}
        print("ExternalStateStore initialized.")

    def get_state(self, session_id: str, key: str = None):
        if key:
            return self.data.get(session_id, {}).get(key)
        return self.data.get(session_id, {})

    def set_state(self, session_id: str, key: str, value):
        if session_id not in self.data:
            self.data[session_id] = {}
        self.data[session_id][key] = value
        print(f"Store: State for {session_id}.{key} set to {value}")

    def update_state(self, session_id: str, updates: dict):
        if session_id not in self.data:
            self.data[session_id] = {}
        self.data[session_id].update(updates)
        print(f"Store: State for {session_id} updated with {updates}")

    def delete_session_state(self, session_id: str):
        if session_id in self.data:
            del self.data[session_id]
            print(f"Store: Session state for {session_id} deleted.")

# 工具不再直接管理状态，而是通过 ExternalStateStore 访问
class CartTool:
    def __init__(self, state_store: ExternalStateStore):
        self.state_store = state_store

    def add_item_to_cart(self, session_id: str, item_name: str, quantity: int) -> str:
        current_cart = self.state_store.get_state(session_id, 'cart') or {}
        current_cart[item_name] = current_cart.get(item_name, 0) + quantity
        self.state_store.set_state(session_id, 'cart', current_cart)
        return f"已将 {quantity} 件 {item_name} 加入您的购物车 (会话ID: {session_id})。"

    def view_cart_content(self, session_id: str) -> str:
        cart = self.state_store.get_state(session_id, 'cart')
        if not cart:
            return f"您的购物车是空的 (会话ID: {session_id})。"
        cart_items = [f"{qty} 件 {item}" for item, qty in cart.items()]
        return f"您的购物车包含：{', '.join(cart_items)} (会话ID: {session_id})。"

    def clear_session_cart(self, session_id: str) -> str:
        self.state_store.set_state(session_id, 'cart', {}) # 清空购物车
        return f"您的购物车已清空 (会话ID: {session_id})。"

class AgentWithExternalState:
    def __init__(self, llm_mock_function, state_store: ExternalStateStore):
        self.llm_mock = llm_mock_function
        self.state_store = state_store
        self.session_id = str(uuid.uuid4())
        self.conversation_history = []
        self.cart_tool = CartTool(state_store) # Agent持有工具实例

        print(f"Agent initialized with session_id: {self.session_id}")

    def _call_llm(self, prompt: str) -> str:
        print(f"n--- LLM Prompt (Agent Session ID: {self.session_id}) ---")
        print(prompt)
        print(f"--- End LLM Prompt ---")
        return self.llm_mock(prompt, self.session_id)

    def _process_user_input(self, user_input: str) -> str:
        context = "n".join(self.conversation_history)

        prompt = f"""
        你是一个购物助手。你需要理解用户的意图，并使用提供的工具来完成任务。
        当前会话ID: {self.session_id}

        可用工具:
        - add_item_to_cart(session_id: str, item_name: str, quantity: int): 向指定会话的购物车添加商品。
        - view_cart_content(session_id: str): 查看指定会话的购物车内容。
        - clear_session_cart(session_id: str): 清空指定会话的购物车。

        当前对话历史:
        {context}

        用户: {user_input}
        助手:
        """

        llm_response = self._call_llm(prompt)
        self.conversation_history.append(f"用户: {user_input}")
        self.conversation_history.append(f"助手: {llm_response}")

        # 模拟LLM响应是工具调用指令
        if "CALL_TOOL:add_item_to_cart" in llm_response:
            try:
                parts = llm_response.split("CALL_TOOL:add_item_to_cart(")[1].split(")", 1)[0]
                _session_id = parts.split(',')[0].strip().strip("'"")
                item_name = parts.split(',')[1].strip().strip("'"")
                quantity = int(parts.split(',')[2].strip())
                tool_output = self.cart_tool.add_item_to_cart(_session_id, item_name, quantity)
                return tool_output
            except Exception as e:
                return f"工具调用失败: {e}"
        elif "CALL_TOOL:view_cart_content" in llm_response:
            try:
                _session_id = llm_response.split("CALL_TOOL:view_cart_content(")[1].split(")", 1)[0].strip().strip("'"")
                tool_output = self.cart_tool.view_cart_content(_session_id)
                return tool_output
            except Exception as e:
                return f"工具调用失败: {e}"
        elif "CALL_TOOL:clear_session_cart" in llm_response:
            try:
                _session_id = llm_response.split("CALL_TOOL:clear_session_cart(")[1].split(")", 1)[0].strip().strip("'"")
                tool_output = self.cart_tool.clear_session_cart(_session_id)
                return tool_output
            except Exception as e:
                return f"工具调用失败: {e}"
        else:
            return llm_response

# 模拟 LLM 行为的函数 (与策略二相同，因为prompt结构类似)
def mock_llm_for_external_state(prompt: str, session_id: str) -> str:
    if "两件衬衫" in prompt:
        return f"CALL_TOOL:add_item_to_cart('{session_id}', '衬衫', 2)"
    elif "一条裤子" in prompt:
        return f"CALL_TOOL:add_item_to_cart('{session_id}', '裤子', 1)"
    elif "查看购物车" in prompt:
        return f"CALL_TOOL:view_cart_content('{session_id}')"
    elif "清空购物车" in prompt:
        return f"CALL_TOOL:clear_session_cart('{session_id}')"
    else:
        return "抱歉，我不明白您的意思。"

# 运行 Agent
external_state_store = ExternalStateStore()
agent_with_external_state = AgentWithExternalState(mock_llm_for_external_state, external_state_store)

print("n--- 场景3: Agent框架/外部存储维护状态 ---")
print(agent_with_external_state._process_user_input("我想买两件衬衫"))
print(agent_with_external_state._process_user_input("再加一条裤子"))
print(agent_with_external_state._process_user_input("我的购物车里有什么？"))
print(agent_with_external_state._process_user_input("清空购物车"))
print(agent_with_external_state._process_user_input("我的购物车里有什么？"))

# 模拟另一个会话，验证状态隔离
print("n--- 场景3.1: 另一个用户/会话 ---")
another_agent_external = AgentWithExternalState(mock_llm_for_external_state, external_state_store)
print(another_agent_external._process_user_input("给我买一件夹克"))
print(another_agent_external._process_user_input("查看购物车"))
print(agent_with_external_state._process_user_input("我的购物车里有什么？")) # 验证第一个Agent的购物车未受影响

在这个例子中，ExternalStateStore 是一个独立的类，模拟了外部存储。CartTool 不再拥有自己的 carts 字典，而是通过 ExternalStateStore 的 get_state 和 set_state 方法来读写购物车状态。Agent 依然管理 session_id 并将其传递给工具，工具再用它来与状态存储交互。这种模式解耦了工具和状态的持久化机制，提供了更大的灵活性。

策略四：Session/Conversation ID 模式

这不是一个独立的状态管理策略，而是贯穿于策略二和策略三的一种设计模式。

原理：
Agent 在开始一个复杂任务或会话时生成一个唯一的 Session ID（例如，UUID）。在所有后续的工具调用中，Agent 都会将这个 Session ID 作为参数传递给工具。工具或状态管理层利用此 ID 来存取与该会话相关的状态。

优点：

• 清晰地界定会话范围：每个任务或用户会话都有一个明确的标识符，避免状态混淆。
• 便于追踪和调试：通过 Session ID 可以轻松追踪一个会话的所有相关操作和状态变化。
• 支持多用户并发场景：不同的 Session ID 确保了不同用户或任务的状态隔离。
• 简化 Agent 与工具的接口：工具不再需要知道“我是哪个用户”或“这是哪个任务”，只需处理传入的 Session ID 即可。

缺点：

• 需要 Agent 和所有相关工具都遵循此约定：所有参与有状态交互的组件都必须支持 Session ID 的传递和使用。
• Session ID 的生命周期管理：何时生成、何时过期、何时清理 Session ID 及其关联的状态，需要明确的策略。

我们在策略二和策略三的代码示例中已经广泛使用了 Session ID (self.session_id) 来区分不同的 Agent 实例（模拟不同的用户会话），并确保它们的状态是隔离的。这正是 Session/Conversation ID 模式的应用。

策略	状态存储位置	LLM 上下文影响	优点	缺点	适用场景
LLM Prompt 维护	LLM Prompt 内部 (纯文本)	显著增加	实现最简单，快速原型	易超上下文，成本高，噪音多，解析困难	状态信息少量，短期会话，低并发
工具内部维护	工具实例内部 (内存)	少量（仅传递ID）	封装性好，工具逻辑自洽	状态间共享难，生命周期需管理，并发支持有限	特定工具独占状态，如特定会话的购物车、文件句柄
Agent/外部存储维护	外部存储（DB, Redis, Cache 等）	少量（仅传递ID）	灵活可扩展，状态共享易，支持持久化	实现复杂，需额外基础设施，状态Schema设计	复杂多步骤任务，多工具协作，高并发，需持久化
Session ID 模式	贯穿于工具内部/外部存储（设计模式）	少量（仅传递ID）	隔离会话，便于追踪，支持多用户	需要所有组件遵循约定，Session生命周期管理复杂	所有需要区分会话或用户状态的场景

5. 状态管理的关键考量与最佳实践

在实际项目中实施 Stateful Tool Calls 时，需要考虑一系列关键因素，并遵循一些最佳实践。

1. 状态的粒度与范围

   • 全局状态 (Global State)：整个 Agent 系统共享的状态。应谨慎使用，可能导致难以调试的副作用。
   • 用户状态 (User State)：特定用户的所有会话和任务共享的状态（例如，用户偏好、历史订单）。
   • 会话状态 (Session/Conversation State)：与单个用户会话或特定任务实例绑定的状态。这是最常用的粒度。
   • 工具内部状态 (Tool-Specific State)：仅与特定工具实例相关的状态。
     选择合适的粒度至关重要，它决定了状态的可见性、生命周期和管理复杂性。

2. 状态的生命周期

   • 创建：状态何时被创建？通常在 Agent 启动新会话或新任务时。
   • 更新：状态何时被修改？工具执行后或 Agent 做出决策后。
   • 读取：状态何时被访问？Agent 规划时，工具执行前。
   • 销毁/过期：状态何时被清除？会话结束、用户登出、达到过期时间、任务完成。合理设置过期机制，防止状态无限增长。

3. 状态的持久化

   • 内存 (In-Memory)：最快，但数据在进程重启后丢失。适用于短期、非关键状态。
   • 文件系统 (File System)：简单，但并发访问效率低，不适合高并发。
   • 数据库 (Database)：关系型数据库（如 PostgreSQL, MySQL）或 NoSQL 数据库（如 MongoDB）提供强大的持久化、查询和事务能力。适用于关键、复杂、需持久化的状态。
   • 缓存系统 (Cache Systems)：如 Redis、Memcached，提供快速读写，支持过期策略。适用于高并发、需快速访问但允许一定数据丢失的状态。

4. 并发与一致性

   • 在多用户或多 Agent 场景下，多个并发请求可能同时读写同一个状态。
   • 需要考虑并发控制机制，如锁 (Locking)、事务 (Transactions) 或乐观并发控制 (Optimistic Concurrency Control)，以确保状态的一致性。例如，当多个 Agent 同时尝试修改同一购物车的商品数量时，必须确保最终结果是正确的。

5. 安全性与隐私

   • 状态中可能包含敏感的用户数据（如个人信息、支付信息）。
   • 确保状态存储的安全性，采取加密、访问控制、权限管理等措施。
   • 遵守数据隐私法规（如 GDPR, CCPA）。

6. 错误处理与回滚

   • 状态更新失败时如何处理？是否需要回滚到之前的状态？
   • 设计健壮的错误处理机制，例如，如果支付失败，购物车状态是否应该恢复？
   • 事务机制在复杂流程中尤为重要。

7. 可观察性 (Observability)

   • 如何监控和调试状态？日志记录状态的变化、关键操作。
   • 提供工具或界面来查看当前会话的状态，便于开发和运维。

8. 状态 Schema 设计

   • 状态数据应该以结构化的方式存储，例如 JSON、Protobuf 或数据库 Schema。
   • 设计清晰、可扩展的状态 Schema，确保 Agent 和工具能够轻松理解和存取所需信息。避免在状态中存储冗余或非必要的数据。

9. 状态清理策略

   • 避免状态无限增长，导致存储空间耗尽和性能下降。
   • 实施定期的垃圾回收或基于时间 (TTL – Time To Live) 的过期策略。
   • 例如，用户会话结束后，其购物车状态可以在一定时间内保留，然后自动清理。

6. 实际应用场景示例

Stateful Tool Calls 是构建真正智能和实用 Agent 的基石。它们在许多复杂应用场景中发挥着关键作用：

• 电商购物流程：
  • 购物车：用户添加、删除、修改商品数量，所有操作都基于一个会话级别的购物车状态。
  • 订单生成：从购物车到填写收货地址、选择支付方式、最终确认，每一步都依赖前一步的输入和中间状态。
• 多步骤数据分析：
  • 数据加载：Agent 调用工具加载 CSV 文件。
  • 数据清洗：Agent 调用另一个工具执行缺失值处理、格式转换，并将清洗后的数据作为状态传递。
  • 执行分析：Agent 调用统计工具在清洗后的数据上运行模型。
  • 生成报告：Agent 利用分析结果和之前的数据上下文生成报告。
• 复杂配置管理：
  • 逐步引导用户完成产品配置（例如，定制一台电脑），每个选项的选择都会更新配置状态，并影响后续可选的组件。
• 任务自动化工作流：
  • 审批流程：一个文档从创建、提交、审核到批准，每个阶段的状态（待审批、审批中、已批准、已拒绝）都会流转。
  • IT 运维：执行一系列部署、监控、故障恢复步骤，每一步都依赖于前一步的执行结果和系统当前的状态。

7. 展望未来：Stateful Agent 的发展方向

Stateful Tool Calls 的概念和实现仍在不断演进。未来的发展方向可能包括：

• 更智能的状态推断与管理：Agent 能够自主决定哪些信息应作为状态持久化，哪些是瞬时上下文，甚至能够预测何时需要某种状态。
• 跨模态、跨 Agent 的状态共享：未来的 Agent 可能需要处理文本、语音、图像等多种模态的信息，并在不同专业 Agent 之间共享复杂的状态。
• 结合 RAG (Retrieval Augmented Generation) 与状态管理：将外部知识检索与会话状态相结合，提升 Agent 在特定领域和复杂任务中的知识利用效率。
• 标准化状态管理接口和框架：随着 Agent 技术的成熟，可能会出现更加标准化的状态管理 API 和框架，简化开发和集成。
• 基于区块链或去中心化存储的状态：为状态提供更高的安全性、透明性和不可篡改性，尤其适用于金融、供应链等场景。

8. 智能 Agent 的核心能力

Stateful Tool Calls 是构建强大、智能 Agent 的关键能力。它将 Agent 从“健忘”的单次交互机器，转变为能够理解并连贯执行复杂多步骤任务的智能伙伴。通过合理选择和实施状态管理策略，我们能极大地提升 Agent 的实用性、效率和用户体验，使其在更广泛的实际应用中发挥巨大价值。