• FunctionCalling相关
  • 使用FunctionCalling的方式
• MCP相关
  • 传统做法：一次性传递所有工具
  • 渐进式披露的解决方案
• config配置基类
• Message消息类
• llm基础客户端
• Agent基类
• 工具基类Tool
  • 如何获取子工具
  • 对子工具进行处理
• 注册工具
• 关于调用工具
  • 关于参数的提取与转换
• 获取工具执行信息
  • 实现方式
• 多工具协作
  • 链式调用
  • 工具异步执行
• Agent的记忆系统
  • MemoryTool基础功能
  • MemoryManager管理类
  • 四种记忆类型的设计
  • 四种记忆类型中的时间性计算方式
• RAG 系统的设计
  • 知识库的创建
  • RAG的检索方式
• 上下文工程
  • 为什么使用上下文工程
  • 什么是上下文工程
  • 跨轮次的上下文接力工程
  • 压缩整合的上下文工程
  • 结构化笔记上下文工程NoteTool
• A2A智能体沟通协议

读前须知

文档内容学习的是2025年底到2026年初的Agent相关知识，且是自己调用基础的模型API接口，来实现的各种Agent功能，并未使用流行框架（LangChain和LangGraph）， 主要目的在于理解Agent框架的底层设计，了解Agent的执行流程。

下面列出截止到2026/09/18,笔记记录已经落后的相关内容如下：

FunctionCalling相关

记录笔记时，模型大多数还不具备FunctionCalling功能（也可能是我不知道所以没有学习），所以笔记中实现调用工具的功能是通过先把工具记录为：工具名+工具描述+工具函数本体的方式。

然后把工具记录到字典中，最终传给模型的时候直接把工具列表传过来，结合提示词模板，让模型依据用户信息决定是否需要调用工具，以及调用哪个工具，把参数返回，最后从模型回答中解析参数，再转换为需要的数据类型进行调用工具函数。、

提示词模板示例如下：

# ReAct 提示词模板
REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。

可用工具如下:
{tools}

请严格按照以下格式进行回应:

Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一:
- `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。
- `Finish[最终答案]`:当你认为已经获得最终答案时。
- 当你收集到足够的信息，能够回答用户的最终问题时，你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。

现在，请开始解决以下问题:
Question: {question}
History: {history}
"""

使用FunctionCalling的方式

模型目前都具有FunctionCalling功能，在与模型进行交互时，工具信息不需要再填入到提示词模板中进行传输。

模型有特定的字段来接收，具体示例(以OpenAI风格API为例)如下：

{
  "model": "gpt-4.1",
  "messages": [
    {"role": "user", "content": "北京天气怎么样？"}
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "获取指定城市的天气",
        "parameters": {
          "type": "object",
          "properties": {
            "city": {
              "type": "string",
              "description": "城市名称"
            },
            "unit": {
              "type": "string",
              "enum": ["celsius", "fahrenheit"],
              "description": "温度单位"
            }
          },
          "required": ["city"]
        }
      }
    }
  ]
}

我们传递工具信息时，直接作为字段属性传给大模型即可，模型有专门接收工具信息的json字段。模型的名称，描述以及参数信息都可以直接传给大模型。

工具类信息大致如下：

class SearchTool:
    name = "search"
    description = "搜索最新信息"

    def to_openai_schema(self):
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "搜索关键词"
                        }
                    },
                    "required": ["query"]
                }
            }
        }

    def run(self, query):
        # 真正执行搜索
        return f"搜索结果：{query}..."

同样，模型后续需要进行调用某个工具时，会把工具名称，工具参数也放到对应的json字段中返回，我们直接在回复字段中取出对应参数即可调用相关工具。

模型返回示例如下：

{
  "choices": [
    {
      "message": {
        "role": "assistant",
        "tool_calls": [
          {
            "id": "call_abc123",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": "{\"city\":\"北京\",\"unit\":\"celsius\"}"
            }
          }
        ]
      }
    }
  ]
}

之前是直接把工具信息填入到提示词模板中传给大模型的，这样的方式属于硬编码，后续调整很不方便，耦合度太高，同样返回时的参数从模型回答中进行提取，完全依赖于模型回复格式，若上下文变大，模型出现幻觉，未按固定格式输出，参数提取会失败，调用就会失败。

使用FunctionCalling之后，工具信息直接依据模型提供方固定的接收方式进行传递，工具调用及参数信息也由模型的固定输出字段进行提取，调用稳定性大幅提高，同时后续进行工具微调，业务逻辑调整等都十分方便

下面把具体调用示例步骤显示如下：

第一步：请求模型

import json
from openai import OpenAI

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

messages = [
    {"role": "system", "content": "你是一个可以使用工具的AI助手。"},
    {"role": "user", "content": "北京天气怎么样？"}
]

response = client.chat.completions.create(
    model="gpt-4.1",
    messages=messages,
    tools=tools
)

第二步：模型返回 tool_calls 模型不会直接执行函数，而是返回：

{
  "role": "assistant",
  "tool_calls": [
    {
      "id": "call_abc123",
      "type": "function",
      "function": {
        "name": "get_weather",
        "arguments": "{\"city\":\"北京\",\"unit\":\"celsius\"}"
      }
    }
  ]
}

注意：arguments 是一个 JSON 字符串，不是已经解析好的对象

第三步：解析参数并执行本地工具

msg = response.choices[0].message

if msg.tool_calls:
    for tool_call in msg.tool_calls:
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)

        if name == "get_weather":
            result = get_weather(**args)   # 真正执行你的函数
        else:
            result = f"未知工具: {name}"

        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False)
        })

第四步：把工具结果回传模型

messages.append(response.choices[0].message)  # 把 assistant 的 tool_calls 消息也加进去

response2 = client.chat.completions.create(
    model="gpt-4.1",
    messages=messages,
    tools=tools
)

print(response2.choices[0].message.content)
# 北京当前天气晴，气温 25 摄氏度。

如果模型觉得还需要调用别的工具，它会继续返回 tool_calls。 所以整体是一个循环

调用 LLM → 返回 tool_calls → 执行工具 → 回传 tool 结果 → 再调用 LLM
直到模型返回普通 content，作为最终回答。

Function Calling 中，工具信息不是拼进 ReAct 提示词模板的 {tools} 文本里，而是作为 API 请求的 tools 字段，以 JSON Schema 列表的形式传给模型。模型返回结构化的 tool_calls，其中包含函数名和 JSON 字符串形式的参数。应用侧解析参数、执行本地工具，再把结果作为 role=tool 消息回传模型，循环直到模型给出最终回答

MCP相关

原本的MCP实现是我们拉去MCP服务到本地之后，启动项目时，我们直接向MCP服务发起请求（例如查询有哪些工具，或是调用需要执行的工具），然后MCP内部执行对应工具后把结果返回。

我们需要做的就是把需要调用的信息（工具信息）发给MCP，然后等待接收结果即可。
具体的工具内部执行和HTTP请求，例如调用高德地图API需要发送HTTP请求，是在MCP内部封装好的，由MCP去执行。

MCP中包含多个工具，同时每个工具都有自己对应的FunctionCalling，描述了该工具具体可以做什么。

但是我们没必要在每次调用模型时都把MCP中所有的工具信息都发给模型，这样的话会造成上下文过长，信息一多模型的注意力便会分散。

传统做法：一次性传递所有工具

早期的 MCP 实现中，客户端会在连接 MCP 服务器后，立即发送 tools/list 请求，获取该服务器上所有可用工具的完整定义（包括名称、描述、JSON Schema），然后一次性全部传给 LLM。

这个过程的核心机制是：

1. 客户端连接 MCP 服务器（本地启动的进程或远程服务）。
2. 发送 tools/list 请求，服务器返回该 MCP 中所有工具的定义列表，每个工具都包含 name、description 和 inputSchema。
3. 客户端将这些工具定义转换成 LLM 的 Function Calling 格式（比如 OpenAI 的 tools 字段格式），随请求一起发给模型。
4. 模型返回 tool_calls，客户端再通过 MCP 的 tools/call 请求去实际执行工具。

所以在这种模式下，一个 MCP 服务器对应的是“一批工具”，客户端会把这一批工具全部传给模型，而不是“一个 MCP 对应一个描述”。

带来问题：上下文膨胀

当 MCP 服务器暴露的工具数量很多时（比如几十上百个），把所有工具定义一次性塞进模型的上下文窗口，会带来严重问题：

• Token 消耗巨大：每个工具的 JSON Schema 都要占用 token，工具多了之后光是工具定义就可能吃掉大量上下文预算。
• 模型选择困难：工具太多时，模型反而容易选错或混淆相似的工具。
• 延迟增加：传输和处理大量工具定义会增加请求延迟

渐进式披露的解决方案

渐进式披露的核心思想是：不要把 MCP 里所有工具的完整定义一次性全部传给模型，而是分层、按需地暴露工具信息。

Anthropic 的官方指引明确指出，MCP 服务器应该渐进式地揭示能力，而不是把所有工具定义都倾倒进模型上下文。

具体实现方式有几种：

方式一：先给“分类”，再给“具体工具”

客户端先只告诉模型 MCP 服务器有哪些工具类别或能力概述（比如“这个 MCP 提供文件操作、数据库查询、网络请求三类工具”）。当模型判断需要某一类工具时，再发送 tools/list 获取该类别的具体工具定义，并追加到对话上下文中。这样模型的第一步决策只需要基于少量信息，等真正需要时才加载详细 Schema

方式二：用“搜索工具”代替“全部工具”

Hermes Agent 采用了这种模式：它把 MCP 和插件工具替换成三个“桥接工具”，模型只看到这三个桥接工具。当模型需要某个具体工具时，通过桥接工具去按需加载对应的 Schema。

ProDisco 也类似：MCP 服务器只暴露 searchTools 和 runSandbox 两个工具，模型先用搜索工具找到需要的 API，再在沙箱中执行，最终只把精简结果返回给模型。

方式三：Anthropic Agent Skills 的三级披露、

Anthropic 的 Agent Skills 采用了更精细的三级渐进披露架构：

• Level 1：只加载技能的 name 和简短描述（始终在系统提示中）。
• Level 2：当模型判断需要某个技能时，加载该技能的详细文档。
• Level 3：只有真正执行时才加载完整的代码或资源。

这种模式让技能数量可以无限扩展，而不会撑爆上下文窗口。

示例如下：

以 Python 的 MCP 客户端为例，传统方式大概是：

# 连接 MCP 服务器后，一次性获取所有工具
tools = await session.list_tools()  # 返回该 MCP 上所有工具

# 全部转换成 OpenAI Function Calling 格式
openai_tools = [mcp_tool_to_openai(t) for t in tools.tools]

# 发给模型
response = client.chat.completions.create(
    model="gpt-4.1",
    messages=messages,
    tools=openai_tools
)

而渐进式披露的客户端会这样处理：

# 先只获取工具的分类/摘要（可能由 MCP 服务器提供一个"元工具"）
summary = await session.call_tool("get_tool_categories", {})

# 把摘要告诉模型，模型决定需要哪类工具
# 然后才按需加载具体工具
if need_weather_tools:
    weather_tools = await session.list_tools(category="weather")
    # 只把 weather 相关的工具加入 tools 字段

问题	回答
MCP 会把所有工具一次性传给模型吗？	传统做法会，客户端连接后立即 tools/list 并全部传给模型。但这不是唯一方式，也不是推荐方式。
一个 MCP 对应一个描述吗？	一个 MCP 服务器通常暴露多个工具，客户端可以通过 tools/list 获取全部，也可以按需部分获取
渐进式披露怎么实现？	客户端先给模型分类/摘要，或提供搜索工具，模型判断需要时才按需加载具体工具的 JSON Schema。Anthropic 的 Agent Skills 采用三级披露架构。

之前理解的 Function Calling 机制（工具定义结构化传给模型、模型返回 tool_calls）在 MCP 场景下依然成立，只是“传哪些工具”从“全部”变成了“按需的部分”。这也是当前 Agent 框架在工具规模增大后必须面对的核心工程问题之一

config配置基类

给出固定默认值，同时若未传入参数，从环境变量中进行读取

"""配置管理"""
import os
from typing import Optional, Dict, Any
from pydantic import BaseModel

class Config(BaseModel):
    """HelloAgents配置类"""
    
    # LLM配置
    default_model: str = "gpt-3.5-turbo"
    default_provider: str = "openai"
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    
    # 系统配置
    debug: bool = False
    log_level: str = "INFO"
    
    # 其他配置
    max_history_length: int = 100
    
    @classmethod
    def from_env(cls) -> "Config":
        """从环境变量创建配置"""
        return cls(
            debug=os.getenv("DEBUG", "false").lower() == "true",
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            temperature=float(os.getenv("TEMPERATURE", "0.7")),
            max_tokens=int(os.getenv("MAX_TOKENS")) if os.getenv("MAX_TOKENS") else None,
        )
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return self.dict()

Message消息类

通过 typing.Literal 将 role 字段的取值严格限制为 "user", "assistant", "system", "tool" 四种，这直接对应 OpenAI API 的规范，保证了类型安全

"""消息系统"""
from typing import Optional, Dict, Any, Literal
from datetime import datetime
from pydantic import BaseModel

# 定义消息角色的类型，限制其取值
MessageRole = Literal["user", "assistant", "system", "tool"]

class Message(BaseModel):
    """消息类"""
    
    content: str
    role: MessageRole
    timestamp: datetime = None
    metadata: Optional[Dict[str, Any]] = None
    
    def __init__(self, content: str, role: MessageRole, **kwargs):
        super().__init__(
            content=content,
            role=role,
            timestamp=kwargs.get('timestamp', datetime.now()),
            metadata=kwargs.get('metadata', {})
        )
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式（OpenAI API格式）"""
        return {
            "role": self.role,
            "content": self.content
        }
    
    def __str__(self) -> str:
        return f"[{self.role}] {self.content}"

llm基础客户端

主要实现基础的模型客户端，是所有Agent实现交流的基础，功能如下：

1. 依据传入的apikey,baseurl等信息进行创建模型客户端，进行大模型的调用

2. 依据provider进行不同模型的自适应，主要是通过模型的apikey以及baseurl进行匹配，然后返回对应的模型提供商，得知使用的供应商后就去获取对应供应商的环境变量信息（这里匹配还是依据不同供应商配置的不同名称的环境变量，太依靠变量名称了，名称一换就匹配不到了）

         """" 自动检测LLM提供商
   
       检测逻辑：
       1. 优先检查特定提供商的环境变量
       2. 根据API密钥格式判断
       3. 根据base_url判断
       4. 默认返回通用配置"""
       # 上面是得出使用的哪个供应商，下面是使用对应供应商的信息
       if self.provider == "openai":
           resolved_api_key = api_key or os.getenv("OPENAI_API_KEY") or os.getenv("LLM_API_KEY")
           resolved_base_url = base_url or os.getenv("LLM_BASE_URL") or "https://api.openai.com/v1"
           return resolved_api_key, resolved_base_url

3. 实现了一些基础功能，例如流式非流式调用等

Agent基类

• 通过接收基础的模型客户端llm，在此基础上进行封装，对除基本问答功能与自动识别供应商外的拓展功能进行统一
• 借助抽象类的特性，继承python中的ABC模式，结合@abstractmethod注解，实现对agent类型的结构功能统一。
• 基类创建好基础的属性以及方法，同时规定子类必须统一实现相同结构的run方法，这样后续不同类型智能体都通过继承方式，实现基础功能统一，若是需要添加独有功能，直接进行添加即可。
• 后续实现不同类型的Agent，都会继承该Agent基类，保证不同类型Agent的调用都是同一种模式，实现无感切换。封装内部实现细节。

"""Agent基类"""
from abc import ABC, abstractmethod
from typing import Optional, Any
from .message import Message
from .llm import HelloAgentsLLM
from .config import Config

class Agent(ABC):
    """Agent基类"""
    
    def __init__(
        self,
        name: str,
        llm: HelloAgentsLLM,
        system_prompt: Optional[str] = None,
        config: Optional[Config] = None
    ):
        self.name = name
        self.llm = llm
        self.system_prompt = system_prompt
        self.config = config or Config()
        self._history: list[Message] = []
    
    @abstractmethod
    def run(self, input_text: str, **kwargs) -> str:
        """运行Agent"""
        pass
    
    def add_message(self, message: Message):
        """添加消息到历史记录"""
        self._history.append(message)
    
    def clear_history(self):
        """清空历史记录"""
        self._history.clear()
    
    def get_history(self) -> list[Message]:
        """获取历史记录"""
        return self._history.copy()
    
    def __str__(self) -> str:
        return f"Agent(name={self.name}, provider={self.llm.provider})"

工具基类Tool

• 这里同样采用抽象类进行结构统一，方便后续注册工具时进行统一管理
• 每一个工具都包含：
  1. 工具名称：name
  2. 工具描述：description，用来告诉大模型该工具的作用，让模型依据该描述来判断是否需要调用这个工具
  3. 是否可展开：expandable，工具是否是可以进行展开的，也就是该工具是否还有子工具。 子工具会使用装饰器进行装饰，给工具附上属性（工具其实就是一个个函数，装饰器就是给函数加上属性，判断时依据属性来判断是否是子工具）

     对工具进行了整合，类似于汽车中包含发动机，这种关联性比较强的工具直接整合到一起，作为子工具出现

  4. 获取工具的参数列表：一般是通过从方法签名或是方法描述文档中获取。

如何获取子工具

先来了解一下装饰器的作用:

1. Python 装饰器：是一个在代码加载时立即执行的函数，它接收被装饰的函数作为参数,给函数贴上属性标签后返回，返回一个新的函数（或原函数）
2. 使用方式

@tool_action("memory_add", "添加新记忆")
def _add_memory(...): ...

• 先调用 tool_action("memory_add", ...)，得到内部的 decorator 函数
• 再把 _add_memory 传给 decorator
• decorator 给函数贴上属性标签（func._is_tool_action = True）后返回

通过装饰器我们可以对函数进行属性的添加，类似于贴上标签，后续可以依据标签实现差异化处理。
例如使用装饰器装饰的函数（或者说是带标签，带属性的）是子工具，未使用装饰器的就是其他辅助函数，类似格式转换，数据解析函数等

装饰器示例代码：

def tool_action(name: str = None, description: str = None):
    """装饰器：标记一个方法为可展开的工具 action

    用法:
        @tool_action("memory_add", "添加新记忆")
        def _add_memory(self, content: str, importance: float = 0.5) -> str:
            '''添加记忆

            Args:
                content: 记忆内容
                importance: 重要性分数
            '''
            ...

    Args:
        name: 工具名称（如果不提供，从方法名自动生成）
        description: 工具描述（如果不提供，从 docstring 提取）
    """
    def decorator(func: Callable):
        func._is_tool_action = True
        func._tool_name = name
        func._tool_description = description
        return func
    return decorator

获取子工具方式

  def get_expanded_tools(self) -> Optional[List['Tool']]:
        """获取展开后的子工具列表

        默认实现：自动从标记了 @tool_action 的方法生成子工具
        子类可以重写此方法提供自定义的展开逻辑

        Returns:
            如果工具支持展开，返回子工具列表；否则返回 None
        """
        if not self.expandable:
            return None

        # 自动从装饰器标记的方法生成工具
        tools = []
        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
            if hasattr(method, '_is_tool_action'):
                tool = AutoGeneratedTool(
                    parent=self,
                    method=method,
                    name=method._tool_name,
                    description=method._tool_description
                )
                tools.append(tool)

        return tools if tools else None

1. 依据inspect.getmembers来获取工具的所有属性，包括方法，同时使用predicate=inspect.ismethod来过滤，得到方法所有的实例方法列表
2. 过滤只要使用装饰器装饰过的方法，因为工具内部还有一些是辅助类方法，像对格式进行转换，整理数据这类方法不可能是子工具，所以就不需要。
3. 过滤得到的数据结构是字典类型，键值对的形式。

对子工具进行处理

由于提取出之后的子工具本质还是方法，我们要把子工具解析为工具Tool的统一格式，工具名称，工具名，然后统一交给大模型，方便使用。

• 主要就是，父工具绑定（要知道当前子工具是哪个工具下的），子工具名，子工具描述，子工具的参数获取

注册工具

注册工具时，有两种方式进行注册：

1. 直接注册为Tool对象，注册到工具列表中

   注册为Tool对象之后，可进行调用工具中的方法获取基本信息，增强了工具的自述性
   例如获取工具的参数信息，子工具信息，子工具参数信息等

2. 注册为一个函数，把一个个函数放到列表中

     def create_calculator_registry():
       """创建包含计算器的工具注册表"""
       registry = ToolRegistry()
   
       # 注册计算器函数
       registry.register_function(
           name="my_calculator",
           description="简单的数学计算工具，支持基本运算(+,-,*,/)和sqrt函数",
           func=my_calculate
       )
   
       return registry
       #使用示例
       # 创建包含计算器的注册表
       registry = create_calculator_registry()

关于调用工具

发消息给大模型之前会连同当前可用工具的列表一并发送，工具列表中信息为工具的名称，工具的描述

• 注册工具：向工具列表中添加工具信息，注意若是有可展开的子工具则进行展开处理
• 提取参数: 从上下文中进行提取参数信息，要注意参数的类型转换
• 执行工具：依据工具名获取到工具的本体进行执行对应函数

关于参数的提取与转换

1. 先解析传来的参数信息，可能是键值对，可能是json，都进行解析得到参数字典
2. 结合工具的参数的类型信息，对参数进行类型转换
3. 把转换好的参数字典传给工具，执行工具调用，得到结果

获取工具执行信息

既然我们想让智能体执行对应工具，那我们就要能够知道智能体每一步执行了哪个工具，结果是什么，而不是只知道智能体执行了工具其他一概不知。

实现方式

实现的方法如下：

1. 通过提示词规定模型返回的文本中要包含工具调用日志信息，工具名，工具参数，执行结果，对结构进行设定。
2. 得到结果后对内容进行提取，得到我们需要的详细信息然后进行记录或是直接输出。
3. 提示词以及解析方式如下：
   提示词示例如下：

     ## 可用工具
     你可以使用以下工具来帮助回答问题
     {tools_description} 
   
     ## 工具调用格式
     当需要使用工具时，请使用以下格式：
     [TOOL_CALL:{tool_name}:{parameters}]

   解析方式示例如下：

   def _parse_tool_calls(self, text: str) -> list:
       """解析文本中的工具调用"""
       pattern = r'\[TOOL_CALL:([^:]+):([^\]]+)\]'
       matches = re.findall(pattern, text)
       
       tool_calls = []
       for tool_name, parameters in matches:
           tool_calls.append({
               'tool_name': tool_name.strip(),
               'parameters': parameters.strip(),
               'original': f'[TOOL_CALL:{tool_name}:{parameters}]'
           })
       
       return tool_calls

多工具协作

链式调用

Agent完成某个任务时可能需要多个工具进行协作，为此建立了一个工具的链式调用机制。

工具系统开发的核心理念：在设计层面，每个工具都应该遵循单一职责原则，专注于特定功能的同时保持接口的统一性，并将完善的异常处理和安全优先的输入验证作为基本要求。在性能优化方面，利用异步执行提高并发处理能力，同时合理管理外部连接和系统资源。

执行时，先把需要使用的工具注册到工具管理器中，然后注册到过滤器链中，让工具按特定顺序执行(本质上是使用列表进行存储工具的顺序，遍历时作为执行依据)

示例如下：

def create_research_chain() -> ToolChain:
    """创建一个研究工具链：搜索 -> 计算 -> 总结"""
    chain = ToolChain(
        name="research_and_calculate",
        description="搜索信息并进行相关计算"
    )

    # 步骤1：搜索信息
    chain.add_step(
        tool_name="search",
        input_template="{input}",
        output_key="search_result"
    )

    # 步骤2：基于搜索结果进行计算
    chain.add_step(
        tool_name="my_calculator",
        input_template="2 + 2",  # 简单的计算示例
        output_key="calc_result"
    )

    return chain

工具异步执行

采用线程池配合Python的协程结合实现工具的异步执行，支持多个工具的并行异步

Agent的记忆系统

采用多种记忆模式设计记忆系统：

1. 长期记忆（语义记忆）：它存储的是更为抽象的知识、概念和规则。例如，通过对话了解到的用户偏好、需要长期遵守的指令或领域知识点，都适合存放在这里。这部分记忆具有高度的持久性和重要性，是智能体形成“知识体系”和进行关联推理的核心
2. 短期的工作记忆：一般只针对当前对话，并有最大窗口限制，扮演着智能体“短期记忆”的角色，主要用于存储当前对话的上下文信息。为确保高速访问和响应，其容量被有意限制（例如，默认50条），并且生命周期与单个会话绑定，会话结束后便会自动清理
3. 情景记忆：它负责长期存储具体的交互事件和智能体的学习经历。与工作记忆不同，情景记忆包含了丰富的上下文信息，并支持按时间序列或主题进行回顾式检索，是智能体“复盘”和学习过往经验的基础。
4. 感知记忆:该模块专门处理图像、音频等多模态信息，并支持跨模态检索。其生命周期会根据信息的重要性和可用存储空间进行动态管理。

采用统一入口状态机进行区分调用的记忆操作类型：

def execute(self, action: str, **kwargs) -> str:
    """执行记忆操作

    支持的操作：
    - add: 添加记忆（支持4种类型: working/episodic/semantic/perceptual）
    - search: 搜索记忆
    - summary: 获取记忆摘要
    - stats: 获取统计信息
    - update: 更新记忆
    - remove: 删除记忆
    - forget: 遗忘记忆（多种策略）
    - consolidate: 整合记忆（短期→长期）
    - clear_all: 清空所有记忆
    """

    if action == "add":
        return self._add_memory(**kwargs)
    elif action == "search":
        return self._search_memory(**kwargs)
    elif action == "summary":
        return self._get_summary(**kwargs)
    # ... 其他操作

通过action参数指定具体操作，使用kwargs允许每个操作有不同的参数需求,这里kwargs接受多个参数，一般是键值对的形式（也就是关键字参数）方法需要参数则对应传入，不需要则可以进行按需过滤，或者不做处理单纯忽略掉不需要的参数。实现按需接收。

存储记忆时采用同步存入记忆对应的importance重要程度（依据分类来区分，例如属于短期记忆的内容重要程度会低一些），用来作为搜索记忆时的权重，越低的越不重要，排序时尽量靠后。

---

MemoryTool基础功能

这是暴露给大模型的几个基础功能：

1. add：add操作是记忆系统的基础，它模拟了人类大脑将感知信息编码为记忆的过程。在实现中，我们不仅要存储记忆内容，还要为每个记忆添加丰富的上下文信息，这些信息将在后续的检索和管理中发挥重要作用
2. search：search操作是记忆系统的核心功能，它需要在大量记忆中快速找到与查询最相关的内容。它涉及语义理解、相关性计算和结果排序等多个环节。
3. forget：遗忘机制模拟人类大脑的选择性遗忘过程，支持三种策略：基于重要性（删除不重要的记忆）、基于时间（删除过时的记忆）和基于容量（当存储接近上限时删除最不重要的记忆）
4. consolidate：模拟人类大脑将短期记忆转化为长期记忆的过程。默认设置是将重要性超过0.7的工作记忆转换为情景记忆，这个阈值确保只有真正重要的信息才会被长期保存。整个过程是自动化的，用户无需手动选择具体的记忆，系统会智能地识别符合条件的记忆并执行类型转换。

MemoryTool构建了一个完整的记忆生命周期管理体系。从记忆的创建、检索、摘要到遗忘、整合和管理，形成了一个闭环的智能记忆管理系统，让Agent真正具备了类人的记忆能力

---

MemoryManager管理类

MemoryManager作为记忆系统的核心协调者，负责管理不同类型的记忆模块(上述提到的四个功能模块)，并提供统一的操作接口，并指定使用的配置信息。具体的存储与检索能力由各记忆类型在内部实现

MemoryTool在初始化时会创建一个MemoryManager实例，并根据配置启用不同类型的记忆模块。这种设计让用户可以根据具体需求选择启用哪些记忆类型

四种记忆类型的设计

1. 工作记忆

工作记忆是记忆系统中最活跃的部分，它负责存储当前对话会话中的临时信息。工作记忆的设计重点在于快速访问和自动清理，这种设计确保了系统的响应速度和资源效率。

工作记忆的存储：

工作记忆采用了纯内存存储方案(直接存到集合中)，配合TTL（Time To Live）机制进行自动清理。这种设计的优势在于访问速度极快，但也意味着工作记忆的内容在系统重启后会丢失。这种特性正好符合工作记忆的定位，存储临时的、易变的信息

工作记忆的检索

   base_relevance = vector_score * 0.7 + keyword_score * 0.3 if vector_score > 0 else keyword_score
            time_decay = self._calculate_time_decay(memory.timestamp)
            importance_weight = 0.8 + (memory.importance * 0.4)
            
            final_score = base_relevance * time_decay * importance_weight

• 这里的base_relevance是综合向量得分占0.7关键词得分占0.3然后得出的总体值

• 这里的timedecay则是时间越久远分值就越低，

• 然后最后的importance_weight则是给前两个分数一个整体的权重值，如果是1就不对他们造成影响，如果是小于1就代表让整体得分变低，排序就更靠后，自然降低了排序率，

然后importance_weight内部的话，一般保存记忆时会关联带的有记忆重要性分值也就是memory_importance的值（0-1），这样的话若是0.5则刚刚好乘以0.4再加0.8等于1，不对整体分值照成影响，若是大于0.5则importance_weight会大于1，这样提高分值，若是小于0.5则会小于1刚好降低相关性。

importance_weight是针对于记忆的基础记忆重要性分值，然后timedeacy则是依据记忆新鲜度，第一个baserelevacnce则是关于语义相似度的分值，三者结合综合了语义相关性，新鲜度，记忆本身的重要性得出一个综合分数进行搜寻相关的内容

最后根据排序取关联性最高同时时效性最高的记忆

2. 情景记忆

情景记忆存储的是记忆与事件的关联，存储具体的事件和经历，它偏向于保持事件的完整性和时间序列关系。情景记忆采用了SQLite+Qdrant的混合存储方案，SQLite负责结构化数据的存储和复杂查询，Qdrant负责高效的向量检索。

• SQLite: 结构化存储记忆信息，包括记忆相关的元数据信息
• Qdrant: 向量存储记忆信息，用于语义搜索

情景记忆存储

情景记忆的存储是SQLite结构化存储和Qdrant向量存储两种方式。

1. 把记忆进行结构化之后存储进SQLite中
2. 记忆向量化后存入Qrdant中 检索时，先依据需求去SQLite中查询符合要求的记忆列表，然后去向量数据库中寻找语义相近的记忆列表，最后得到两部分的交集记忆列表。
3. 依据权重计算得分总值：(向量相似度 × 0.8 + 时间近因性 × 0.2) × (0.8 + 重要性 × 0.4)

这样确保检索结果既语义相关又时间相关

情景记忆的检索方式

1. 利用 SQLite 处理精确查询的接口。在检索时，先去调用该方法，根据传入的时间范围、重要性等级、用户ID等硬性条件，把符合结构化规则的事件 ID 先筛选出来
2. 然后去Qrdant中查询语义相关的记忆列表利用 Qdrant 处理语义查询的接口。它将用户的自然语言问题转为向量，在海量历史记忆中找出“意思最像”的 Top N 个结果
3. 结合前两步的结果，求两者的交集数据，作为这次查询相关的记忆结果
4. 最后依据权重计算得分总值：(向量相似度 × 0.8 + 时间近因性 × 0.2) × (0.8 + 重要性 × 0.4)

确保检索结果既语义相关又时间相关

3. 语义记忆

语义记忆负责存储抽象的概念、规则和知识。语义记忆的设计重点在于知识的结构化表示和智能推理能力。语义记忆采用了Neo4j图数据库和Qdrant向量数据库的混合架构，这种设计让系统既能进行快速的语义检索，又能利用知识图谱进行复杂的关系推理。

结合语义记忆的原理，我们可以在这个基础上进行智能体的增强，让智能体把聊天过程中的信息进行整合，最终变 成知识库，可以是常用知识点的总结（如何判断呢？加一个字段作为知识点的出现频繁程度，若是特别频繁就进行存储为对应类型的知识库，判断出现次数时需要进行语义的判断），也可以是某种错误事件的经验总结，若是之前ai犯错，那就把事件类型，原因总结起来作为错误经验知识库，让智能体可以自动成长。

语义记忆的存储

1. 写入记忆时，使用nlp从非结构化文本中自动抽取出 Entity 和 Relation，把记忆存储到图数据库。
2. 记忆转换为向量，结合元数据存储到向量数据库中

   def add(self, memory_item: MemoryItem) -> str:
       """添加语义记忆"""
       # 1. 生成文本嵌入
       embedding = self.embedding_model.encode(memory_item.content)
       
       # 2. 提取实体和关系
       entities = self._extract_entities(memory_item.content)
       relations = self._extract_relations(memory_item.content, entities)
       
       # 3. 存储到Neo4j图数据库
       for entity in entities:
           self._add_entity_to_graph(entity, memory_item)
       
       for relation in relations:
           self._add_relation_to_graph(relation, memory_item)
       
       # 4. 存储到Qdrant向量数据库
       metadata = {
           "memory_id": memory_item.id,
           "entities": [e.entity_id for e in entities],
           "entity_count": len(entities),
           "relation_count": len(relations)
       }
       
       self.vector_store.add_vectors(
           vectors=[embedding.tolist()],
           metadata=[metadata],
           ids=[memory_item.id]
       )

语义记忆的检索

语义记忆的检索实现了混合搜索策略，结合了向量检索的语义理解能力和图检索的关系推理能力

先分别去向量数据库和图数据库去查询相关的记忆列表，然后对记忆列表进行整合，最后依据权重计算总得分，返回排序后的记忆结果。

语义记忆的评分公式为：(向量相似度 × 0.7 + 图相似度 × 0.3) × (0.8 + 重要性 × 0.4)。这种设计的核心思想是：

• 向量检索权重（0.7）：语义相似度是主要因素，确保检索结果与查询语义相关
• 图检索权重（0.3）：关系推理作为补充，发现概念间的隐含关联
• 重要性权重范围[0.8, 1.2]：避免重要性过度影响相似度排序，保持检索的准确性

def _combine_and_rank_results(self, vector_results, graph_results, query, limit):
    """混合排序结果"""
    combined = {}
    
    # 合并向量和图检索结果
    for result in vector_results:
        combined[result["memory_id"]] = {
            **result,
            "vector_score": result.get("score", 0.0),
            "graph_score": 0.0
        }
    
    for result in graph_results:
        memory_id = result["memory_id"]
        if memory_id in combined:
            combined[memory_id]["graph_score"] = result.get("similarity", 0.0)
        else:
            combined[memory_id] = {
                **result,
                "vector_score": 0.0,
                "graph_score": result.get("similarity", 0.0)
            }
    
    # 计算混合分数
    for memory_id, result in combined.items():
        vector_score = result["vector_score"]
        graph_score = result["graph_score"]
        importance = result.get("importance", 0.5)
        
        # 基础相似度得分
        base_relevance = vector_score * 0.7 + graph_score * 0.3
        
        # 重要性权重 [0.8, 1.2]
        importance_weight = 0.8 + (importance * 0.4)
        
        # 最终得分：相似度 * 重要性权重
        combined_score = base_relevance * importance_weight
        result["combined_score"] = combined_score
    
    # 排序并返回
    sorted_results = sorted(
        combined.values(),
        key=lambda x: x["combined_score"],
        reverse=True
    )
    
    return sorted_results[:limit]

语义记忆的评分公式为：(向量相似度 × 0.7 + 图相似度 × 0.3) × (0.8 + 重要性 × 0.4)。这种设计的核心思想是：

向量检索权重（0.7）：语义相似度是主要因素，确保检索结果与查询语义相关 图检索权重（0.3）：关系推理作为补充，发现概念间的隐含关联 重要性权重范围[0.8, 1.2]：避免重要性过度影响相似度排序，保持检索的准确性

---

4. 感知记忆

感知记忆支持文本、图像、音频等多种模态的数据存储和检索。它采用了模态分离的存储策略，为不同模态的数据创建独立的向量集合，这种设计避免了维度不匹配的问题，同时保证了检索的准确性

存入数据时，依据后缀进行对应数据类型的处理，文本，图片，音频各自使用对应的处理策略

• 图片直接通过视觉编码器变成向量
• 音频直接通过音频编码器变成向量
• 视频则是使用专门的video模态或者使用以下方式：
  1. 视频是一帧一帧图片加上音频组成的
  2. 分别使用视觉和听觉编码器把视频向量化，然后保存到同一组数据中，作为该视频的向量，存储时把视频拆成“关键帧向量(CLIP)”+“音频向量(CLAP)”+“字幕向量(Text)”，检索时分别检索后做加权融合（例如 RRF 算法）
• 使用图片或者视频，音频模型把文件进行转换为文本描述向量再进行存入向量库（该项目未使用这种方法）

文本类问题怎么对应到图片与视频呢

例如我们发出的问题分别是给我一首关于小猫的音频，小猫的图片，学习python的视频。

我们使用的视觉和音频编码器训练时，让“猫的图片向量”和“猫的文字向量”在数学上距离很近。CLIP有独立的文本编码器和图像编码器，但训练目标是把配对数据（图-文）拉近。CLAP同理（音-文）

所以，当你输入文本 "一张关于小猫的图片" 时：

1. 文本通过 CLIP的文本编码器 变成向量 V_text。
2. 系统去图片向量库（perceptual_image）中搜索。
3. 因为训练时对齐过，V_text 会与库中“小猫图片”的向量 V_image 余弦相似度极高，直接匹配成功

针对本项目中的感知记忆如何确定一次检索需要对应的模态类型呢？

1.该项目采用的是传入查询参数时，就指定对应的查询模态库：

query_vector = self._encode_data(query, query_modality)  # 假设 query_modality="text"
store = self._get_vector_store_for_modality(target_modality or query_modality) 
# 假设 target_modality="audio"，拿到的 store 是 perceptual_audio，维度是 self._audio_dim

2.对于查询使用的文本，查哪个库，就用哪个模型去编码文本

• 如果 _encode_data 用的是普通的 text_embedder（输出维度可能是 768 或 1536），即转换后的查询文本向量维度是768或1536。
• 但存储音频的知识库 perceptual_audio 存储是基于 CLAP 的，维度可能是 512（不同模型维度不同）
• 导致查询结果错误！

例如去文本库查数据要使用文本编码器进行编码，图片库使用图片编码器进行编码，视频音频使用对应的编码器进行编码：

def _encode_data(self, query, target_modality):
    if target_modality == "image":
        return self._clip_model.encode_text(query)  # CLIP文本编码器
    elif target_modality == "audio":
        return self._clap_model.encode_text(query)  # CLAP文本编码器
    else:
        return self.text_embedder.encode(query)     # 通用文本编码器

感知记忆的评分公式为：(向量相似度 × 0.8 + 时间近因性 × 0.2) × (0.8 + 重要性 × 0.4)。感知记忆的评分机制还支持跨模态检索，通过统一的向量空间实现文本、图像、音频等不同模态数据的语义对齐。当进行跨模态检索时，系统会自动调整评分权重，确保检索结果的多样性和准确性。此外，感知记忆中的时间近因性计算采用了指数衰减模型

---

四种记忆类型中的时间性计算方式

自然指数函数作为衰减模型

自然指数曲线图如图所示：

自然指数指的是数学中的e^x（e 约等于 2.71828）:

当x>0时：

1. x趋近于正无穷，e^x的值就越大，趋近于正无穷
2. x趋近于负无穷，e^x值越小，趋近于0

当x<0时：

1. x趋近于正无穷，e^x的值趋近于0
2. x趋近于负无穷，e^x值越大趋近于正无穷

以此做为计算记忆的时间得分。公式为：e^(-衰减因子*天数) 即 e^-(衰减因子*天数)

1. 用当前查询当前时间（当时取记忆的时间）减去存储记忆的初始时间，得到小时数后用小时数/24得到记忆的天数。

2. 衰减因子是定义衰减程度的阈值，值越大，与天数的乘积自然越大，加上前面的负号，最终e^-x的值的情况：

   • x越大(就是衰减因子*天数的值越大)，最终结果越趋近于0，记忆的时间得分越低
   • x越小（就是衰减因子*天数的值越小），最终结果越大，记忆时间得分越大

由于是用记忆初始存储时间到当前查询记忆时间的小时数除以24的到的天数，所以天数最小为0，e^-x的值最大为1，时间得分的范围为0-1

---

时间衰减的计算

def _calculate_recency_score(self, timestamp: str) -> float:
    """计算时间近因性得分"""
    try:
        memory_time = datetime.fromisoformat(timestamp)
        current_time = datetime.now()
        age_hours = (current_time - memory_time).total_seconds() / 3600
        
        # 指数衰减：24小时内保持高分，之后逐渐衰减
        decay_factor = 0.1  # 衰减系数
        # 使用自然指数函数，以e为底，变量decay_factor为衰减系数进行计算，exp就是方式计算函数
        # math.exp(x) 不是绝对值，它指的是自然指数函数，即数学中的e^x（e 约等于 2.71828）
        # 若想增大衰减程度，增大decay_factory系数即可
        recency_score = math.exp(-decay_factor * age_hours / 24)
        
        return max(0.1, recency_score)  # 最低保持0.1的基础分数
    except Exception:
        return 0.5  # 默认中等分数

math.exp(x)它指的是自然指数函数，即数学中的 ( e^x )（( e ) 约等于 2.71828）。

这里专门写成 math.exp(-decay_factor * age_hours / 24)，是利用 ( e ) 的负幂次方 来生成一条从 1 平滑下降到 0 的衰减曲线。

1. 定义“衰减速度”

代码中的 decay_factor = 0.1 是速率常数，而 age_hours / 24 是把小时换算成“天”。所以，指数部分（即 ( x )）实际上等于 ( 0.1 \times 天数 )。

2. exp(-x) 的意义

exp(-x) 等价于 ( e^{-x} )，也就是 ( \frac{1}{e^{x}} )。

• 当 age=0（刚发生）时：( x = 0 )，exp(0) = 1（满分）。
• 当 age=24小时（1天） 时：( x = 0.1 )，exp(-0.1) ≈ 0.905。这意味着即使过了一天，分数依然有 90.5%，衰减非常缓慢（这是“24小时内保持高分”的体现）。
• 当 age=240小时（10天） 时：( x = 1.0 )，exp(-1) ≈ 0.368。分数降到了 36.8%。
• 当 age=无限大 时：分数无限趋近于 0（但因为最后有 max(0.1, ...)，所以最低会卡在 0.1）。

总结：exp(-x) 就是标准的指数衰减，利用 ( e ) 的幂次方特性，让分数随着时间增长按比例平滑下降，而不是直线下降。可以把它理解为“记忆的半衰期”模型——越久远的事情，权重越低，但永远不会直接归零。

如果觉得 0.1 系数衰减太慢（10天后还有36%），想让它“忘得更快”，只需要把 decay_factor 调大即可（比如改成 0.5，那么1天后就只剩 60.6% 了）。

这种时间衰减模型模拟了人类记忆中的遗忘曲线，确保了感知记忆系统能够优先检索到时间上更相关的记忆内容。

RAG 系统的设计

RAG是用来进行补充检索需要的信息的，检索增强生成指的就是他。

具体步骤是我们先对传入的知识库文件进行分段存储，作为知识库，后续智能体需要数据时来对应知识库中进行搜寻需要的内容，得到之后进行整理，最终返回答案。

这里的每一层处理都是分开进行模块化设计的，方便后续进行拓展甚至是替换

用户层：RAGTool统一接口
  ↓
应用层：智能问答、搜索、管理
  ↓  
处理层：文档解析、分块、向量化
  ↓
存储层：向量数据库、文档存储
  ↓
基础层：嵌入模型、LLM、数据库

知识库的创建

1.首先把所有类型的文件转换为markdown类型的文本

任意格式文档 → MarkItDown转换 → Markdown文本 → 智能分块 → 向量化 → 存储检索

多模态文档转换载入： 使用MarkItDown作为统一的文档转换引擎，支持几乎所有常见的文档格式。MarkItDown是微软开源的通用文档转换工具，它负责将任意格式的文档统一转换为结构化的Markdown文本。无论输入是PDF、Word、Excel、图片还是音频，最终都会转换为标准的Markdown格式，然后进入统一的分块、向量化和存储流程

def _convert_to_markdown(path: str) -> str:
    """
    Universal document reader using MarkItDown with enhanced PDF processing.
    核心功能：将任意格式文档转换为Markdown文本
    
    支持格式：
    - 文档：PDF、Word、Excel、PowerPoint
    - 图像：JPG、PNG、GIF（通过OCR）
    - 音频：MP3、WAV、M4A（通过转录）
    - 文本：TXT、CSV、JSON、XML、HTML
    - 代码：Python、JavaScript、Java等
    """
    if not os.path.exists(path):
        return ""
    
    # 对PDF文件使用增强处理
    ext = (os.path.splitext(path)[1] or '').lower()
    if ext == '.pdf':
        return _enhanced_pdf_processing(path)
    
    # 其他格式使用MarkItDown统一转换
    md_instance = _get_markitdown_instance()
    if md_instance is None:
        return _fallback_text_reader(path)
    
    try:
        result = md_instance.convert(path)
        markdown_text = getattr(result, "text_content", None)
        if isinstance(markdown_text, str) and markdown_text.strip():
            print(f"[RAG] MarkItDown转换成功: {path} -> {len(markdown_text)} chars Markdown")
            return markdown_text
        return ""
    except Exception as e:
        print(f"[WARNING] MarkItDown转换失败 {path}: {e}")
        return _fallback_text_reader(path)

2.对Markdown文本内容进行分块保存

Markdown结构感知的分块流程：

标准Markdown文本 → 标题层次解析 → 段落语义分割 → Token计算分块 → 重叠策略优化 → 向量化准备
       ↓                ↓              ↓            ↓           ↓            ↓
   统一格式          #/##/###        语义边界      大小控制     信息连续性    嵌入向量
   结构清晰          层次识别        完整性保证    检索优化     上下文保持    相似度匹配

由于所有文档都已转换为Markdown格式，系统可以利用Markdown的标题结构（#、##、###等）进行精确的语义分割

2.1 先进行数据清洗

1. 依据段落先进行数据清洗（就是先按段落拆分），同时每个段落都包含对应的所属标题信息

    paragraphs.append({
           "content": content,
           "heading_path": " > ".join(heading_stack) if heading_stack else None,
           "start": max(0, end_pos - len(content)),
           "end": end_pos,})

2. 判断当前行文本是否是空行，若是空行说明已经是段落结束了，缓冲区进行存储，然后清空缓冲区，进行下一段落的存储

2.2 依据固定token进行切分

1. 确定切分依据的chunk_tokens大小

2. 为防止切分成块（chunk）时相近段落被拆分造成语义割裂，采用了相邻分块重叠的解决方案

3. 依次遍历上一步清洗后的段落数据列表（清洗后数据文档已经按段落分割为一个个列表），然后计算各自的预估token数量（这里是使用的简便的中文一个token，英文单词一个token），计算段落token相加的数量是否大于设定的chunk_tokens数量

4. 这里计算重叠时，是结合段落以及设定的重叠overlap_tokens大小进行计算的。

   4.1 如果上一分块最后一个段落的token量大于设定的overlap_tokens那就不进行重叠

   4.2 如果最后一段的token数量小于设定的重叠token值overlap_tokens那就记录当前段落的token值，继续倒叙循环下一个段落判断token累计值，直到大于或者等于设定的重叠值overlap_token就不进行遍历。

   4.3 遍历到的段落就是下一个分块与当前分块区域的重叠值，并记录token累计值，在这个基础上继续下一步的切分。

3. 把分块存入向量库

1. 先对上述分块chunks进行数据清洗，去除多余的空格，符号等

2. 然后指定分组大小，批量把分块转换为向量 ```python # 批量编码 vecs: List[List[float]] = [] for i in range(0, len(processed_texts), batch_size): part = processed_texts[i:i+batch_size] try: # 使用统一嵌入器（内部处理缓存） part_vecs = embedder.encode(part)

                # 标准化为List[List[float]]格式
                if not isinstance(part_vecs, list):
                    if hasattr(part_vecs, "tolist"):
                        part_vecs = [part_vecs.tolist()]
                    else:
                        part_vecs = [list(part_vecs)]
                
                # 处理向量格式和维度
                for v in part_vecs:
                    try:
                        if hasattr(v, "tolist"):
                            v = v.tolist()
                        v_norm = [float(x) for x in v]
                        
                        # 维度检查和调整
                        if len(v_norm) != dimension:
                            print(f"[WARNING] 向量维度异常: 期望{dimension}, 实际{len(v_norm)}")
                            if len(v_norm) < dimension:
                                v_norm.extend([0.0] * (dimension - len(v_norm)))
                            else:
                                v_norm = v_norm[:dimension]
                        
                        vecs.append(v_norm)
                    except Exception as e:
                        print(f"[WARNING] 向量转换失败: {e}, 使用零向量")
                        vecs.append([0.0] * dimension)
    ```
   

3. 得到对应的向量数据后，要对向量数据进行转换，这里的判断向量维度是否相等只是兜底逻辑，实际并不会走到这里，且 v_norm = v_norm[:dimension]截断后数据也发生变化并不是原本数据，牺牲一致性换来稳定性（防止维度出错之后程序直接报错，例如模型升级维度发生变化）。不过后续要再这里添加异常以及日志信息，方便出问题之后进行排查。

4. 设定重试逻辑防止异常失败后数据丢失。提升稳定性

---

RAG的检索方式

实际应用中，用户的查询表述与文档中的实际内容可能存在用词差异，导致相关文档无法被检索到。为了解决这个问题，采用了三种互补的检索策略：多查询扩展、假设文档嵌入和统一的扩展检索框架

1. 多查询拓展

通过生成语义等价的多样化查询来提高检索召回率的技术：同一个问题可以有多种不同的表述方式，而不同的表述可能匹配到不同的相关文档。例如，"如何学习Python"可以扩展为"Python入门教程"、"Python学习方法"、"Python编程指南"等多个查询。通过并行执行这些扩展查询并合并结果，系统能够覆盖更广泛的相关文档，避免因用词差异而遗漏重要信息

优势在于它能够自动理解用户查询的多种可能含义，特别是对于模糊查询或专业术语查询效果显著

2.假设文档嵌入

核心思想是"用答案找答案"。传统的检索方法是用问题去匹配文档，但问题和答案在语义空间中的分布往往存在差异——问题通常是疑问句，而文档内容是陈述句。通过让LLM先生成一个假设性的答案段落，然后用这个答案段落去检索真实文档，从而缩小了查询和文档之间的语义鸿沟

假设答案与真实答案在语义空间中更加接近，因此能够更准确地匹配到相关文档。即使假设答案的内容不完全正确，它所包含的关键术语、概念和表述风格也能有效引导检索系统找到正确的文档。特别是对于专业领域的查询，能够生成包含领域术语的假设文档，提升检索精度。

3.扩展检索框架

项目将多查询扩展和假设文档嵌入两种策略整合到统一的扩展检索框架中。系统通过enable_mqe和enable_hyde参数让用户可以根据具体场景选择启用哪些策略：对于需要高召回率的场景可以同时启用两种策略，对于性能敏感的场景可以只使用基础检索。

扩展检索的核心机制是"扩展-检索-合并"三步流程。首先，系统根据原始查询生成多个扩展查询（包括多查询扩展生成的多样化查询和假设性文档嵌入生成的假设文档）；然后，对每个扩展查询并行执行向量检索，获取候选文档池；最后，通过去重和分数排序合并所有结果，返回最相关的top-k文档。这种设计的巧妙之处在于，它通过candidate_pool_multiplier参数（默认为4）扩大候选池，确保有足够的候选文档进行筛选，同时通过智能去重避免返回重复内容

对于具体的拓展检索方式不必过于深究，实际情况下需依据业务需求选择适合的方式，这里只是列举两种基本的查询方式而已

上下文工程

首先我们需要理解为什么需要上下文工程，以及什么是上下文工程。

为什么使用上下文工程

目前系统只是具有基础的记忆以及RAG检索增强，流程只是，我们与模型进行交流时，记忆只涉及存储，RAG则是依据用户的提问，去知识库中进行搜寻相关的资料内容进行返回给模型，模型再依据所得到的信息进行总结然后回复。

目前的问题：
主要是模型搜寻到信息之后会不加限制的全部返回给大模型，而上下文工程则是对得到的信息进行处理，例如进行总结，压缩，记录，让大模型更易于理解的同时，节省上下文空间，消除无关（或是不重要）信息的影响，减少幻觉。

上下文工程的重点不是“把长文变短文”，而是“从一堆信息里挑出最值得塞进窗口的那几段”。只有窗口实在塞不下时，才会退而求其次做压缩

什么是上下文工程

上下文工程是在基础记忆与 RAG 检索之后，引入的一套精细化加工流水线。它不仅解决“一股脑全塞给模型”的问题，更通过三个核心动作提升交互质量：

• 智能筛选（选）：依据相关性与新近性打分，丢弃低质或无关信息，减少幻觉诱因；

• 预算管控（压）：通过贪心算法截断超长内容，强制预留系统指令空间，确保模型行为不失控（避免最初的系统提示词被挤出上下文窗口）；

• 结构化重组（排）：将零散信息按角色、证据、背景分区块排列，降低模型的理解难度，从而让回答更精准、更遵从指令。

跨轮次的上下文接力工程

上面描述的是单轮会话中的上下文工程处理，现在我们来长时以及多轮对话的上下文工程如何处理：

如果说基础的上下文工程是解决“一次对话中，怎么把资料整理好递给模型”，那么下面讲述的就是“对话长达几小时甚至几天，窗口满了之后，怎么让模型不‘失忆’”。

1. 压缩整合策略：窗口快满时，调用 LLM 把旧对话浓缩成摘要，用摘要开启新窗口。

   • 定义：当对话接近上下文上限时，对其进行高保真总结，并用该摘要重启一个新的上下文窗口，以维持长程连贯性。
   • 实践：让模型压缩并保留架构性决策、未解决缺陷、实现细节，丢弃重复的工具输出与噪声；新窗口携带压缩摘要 + 最近少量高相关工件（如“最近访问的若干文件”）。

2. 结构化笔记：主动写入外部文件（如 NOTES.md），需要时再主动读取。

   • 定义：也称“智能体记忆”。智能体以固定频率将关键信息写入上下文外的持久化存储，在后续阶段按需拉回。
   • 价值：以极低的上下文开销维持持久状态与依赖关系。例如维护 TODO 列表、项目 NOTES.md、关键结论/依赖/阻塞项的索引，跨数十次- 工具调用与多轮上下文重置仍能保持进度与一致性。
   • 说明：在非编码场景中同样有效（如长期策略性任务、游戏/仿真中的目标管理与统计计数）。结合记忆工具 MemoryTool，可轻松实现文件式/向量式的外部记忆并在运行时检索。

3. 子代理架构：主模型只负责分配任务，子模型在独立的干净窗口里干活，只汇报精简结果。

   • 思想：由主代理负责高层规划与综合，多个专长子代理在“干净的上下文窗口”中各自深挖、调用工具并探索，最后仅回传凝练摘要（常见 1,000–2,000 tokens）。
   • 好处：实现关注点分离。庞杂的搜索上下文留在子代理内部，主代理专注于整合与推理；适合需要并行探索的复杂研究/分析任务。
   • 经验：公开的多智能体研究系统显示，该模式在复杂研究任务上相较单代理基线具有显著优势。

方法取舍可以遵循以下法则：

• 压缩整合：适合需要长对话连续性的任务，强调上下文的“接力”。
• 结构化笔记：适合有里程碑/阶段性成果的迭代式开发与研究。
• 子代理架构：适合复杂研究与分析，能从并行探索中获益。

压缩整合的上下文工程

项目采用的上下文结构如下：

1. 统一入口：将"获取(Gather)- 选择(Select)- 结构化(Structure)- 压缩(Compress)"抽象为可复用流水线，减少在 Agent 实现中的重复模板代码。这种统一的接口设计让开发者无需在每个 Agent 中重复编写上下文管理逻辑。

• 稳定形态：输出固定骨架的上下文模板，便于调试、A/B 测试与评估。我们采用了分区组织的模板结构：

  • [Role & Policies]：明确 Agent 的角色定位和行为准则
  • [Task]：当前需要完成的具体任务
  • [State]：Agent 的当前状态和上下文信息
  • [Evidence]：从外部知识库检索的证据信息
  • [Context]：历史对话和相关记忆
  • [Output]：期望的输出格式和要求

2. 预算守护：在 token 预算内尽量保留高价值信息，对超限上下文提供兜底压缩策略。这确保了即使在信息量巨大的场景下，系统也能稳定运行。

3. 最小规则：不引入来源/优先级等分类维度，避免复杂度增长。实践表明，基于相关性和新近性的简单评分机制，在大多数场景下已经足够有效。

1. 多源信息搜集

第一阶段是从多个来源汇集候选信息。这个阶段的关键在于容错性和灵活性

依据用户最新一次的查询，去记忆以及RAG知识库中查询相关的信息，同时提取离当前最近的三十轮对话，加上系统提示词一起组合为一个基础信息包

2.智能信息选择

第二阶段是根据相关性和新近性对候选信息进行评分和选择。这是整个流水线的核心，直接决定了最终上下文的质量

1. 先从信息包中提取出提示词信息，用设定的压缩窗口大小减去提示词的token大小得到目前可用的token窗口大小

2. 计算各个信息的相关性分数以及时效分数，按倒序排序，在可用token窗口大小的限制下进行贪婪匹配，尽可能多的把信息存到可用的窗口中。

   这里的窗口不是模型的最大上下文窗口，它指的是我们配置的用来进行压缩时使用的最大上下文窗口大小

3. 若是当前剩余窗口大小不足以装下下一个信息时，直接丢弃即可

4. 至此获取到需要压缩的信息（系统提示词，相关记忆，相关知识库，相关的对话历史），准备结构化输出内容

3. 结构化输出

第三阶段是将选中的信息组织成结构化的上下文模板。

具体结构如下：

• [Role & Policies]：明确 Agent 的角色定位和行为准则
• [Task]：当前需要完成的具体任务
• [State]：Agent 的当前状态和上下文信息
• [Evidence]：从外部知识库检索的证据信息
• [Context]：历史对话和相关记忆
• [Output]：期望的输出格式和要求

结构化阶段将散乱的信息包组织成清晰的分区，这种设计有几个优势：

• 可读性：清晰的分区让人类和模型都更容易理解上下文结构
• 可调试性：问题定位更容易，可以快速识别哪个区域的信息有问题
• 可扩展性：添加新的信息源只需要创建新的分区

4. 兜底压缩

第四阶段是对超限上下文进行压缩处理

经历前三步处理后，这里的内容就已经是整理好的数据了，压缩是指的是第二步的智能信息选择之后会进行结构化处理，这第四步的压缩其实就是防止结构化之后会出现token数超过设定的token窗口大小，是对结构化之后的结果进行压缩处理。

目前的压缩策略是遍历第三步的结构化列表，对于其中的信息token数进行累计，后续若是剩余token数已不足以存入片段，就进行概要总结存入压缩后列表。

结构化笔记上下文工程NoteTool

A2A智能体沟通协议

这里主要记对协议的理解，具体实现可看官方文档

A2A是用于智能体之间进行交流的协议，每个Agent会作为一个服务存在，具备skill属性，属性内容是该Agent所具备的能力，以便其他Agent理解并进行协作。

主要运用场景是不同类型的Agent之间进行协作，例如我们需要调用其他团队写的Agent，或者是部署到某个服务器上的Agent，都是以A2A为桥梁来进行连接的。

运用场景如下：

Agent 由不同团队/不同语言实现

Python 写的行程规划 Agent
   ↕ A2A
Java 写的酒店库存 Agent
   ↕ A2A
Go 写的支付 Agent

Agent 要独立部署、独立扩容

景点搜索 Agent  → 部署 3 个实例，负载均衡
天气查询 Agent  → 部署 1 个实例
酒店推荐 Agent  → 部署 2 个实例

Agent 做成独立服务，用 A2A 或 HTTP 暴露

需要动态发现和协商

agent2 提议 deadline=5
agent1 拒绝，反提议 deadline=7
agent2 接受或再提

Agent 网络规模大 多个 Agent，需要服务发现、路由、负载均衡，使用ANP进行设计。

ANP负责搭建Agent网络，A2A负责Agent之间的交流。