# python 3.12.1 Date：2023年12月8日
是最新的维护版本，包含自 3.12.0 以来的 400 多个错误修复、构建改进和文档[更改](https://docs.python.org/3.12/contents.html)。

## 1. typing的优化：
```shell
from typing import Iterable, TypeVar
T = TypeVar('T')

def max[T](args: Iterable[T]) -> T:
    ...
class list[T]:
    def __getitem__(self, index: int, /) -> T:
        ...

    def append(self, item: T) -> None:
       ...
# py_3.10是用显示声明类型别名。Point: TypeAlias = tuple[float, float]
# py_3.9以及下的版本，是用直接赋值定义类型别名，Point = Tuple[float, float]
type Point = tuple[float, float]
# type 语句专门用于声明类型别名，让代码里类型定义的意图更加清晰明确。
type Point[T] = tuple[T, T]

```
### TypeVarTuple 和ParamSpec
```shell
type IntFunc[**P] = Callable[P, int]  # ParamSpec
type LabeledTuple[*Ts] = tuple[str, *Ts]  # TypeVarTuple
type HashableSequence[T: Hashable] = Sequence[T]  # TypeVar with bound
type IntOrStrSequence[T: (int, str)] = Sequence[T]  # TypeVar with constraints
```
以下将结合 Python 的类型注解和类型变量相关知识，对你给出的四个类型别名定义进行详细解释。这些定义主要使用了 Python 的类型系统新特性，如 `ParamSpec`、`TypeVarTuple`、`TypeVar` 等，并且这些特性大多从 Python 3.10 及以上版本开始引入，部分更高级特性如 `TypeVarTuple` 需 Python 3.11 及以上版本支持。

#### 1. `type IntFunc[**P] = Callable[P, int]  # ParamSpec`
```python
from typing import Callable, ParamSpec

P = ParamSpec('P')
type IntFunc[**P] = Callable[P, int]

# 示例函数
def add(a: int, b: int) -> int:
    return a + b

# 使用类型别名注解
func: IntFunc[[int, int]] = add
```
- **含义**：此类型别名 `IntFunc` 借助 `ParamSpec`（`P`）来表示一个可调用对象（函数），该对象可以接受任意数量和类型的参数（`P` 代表参数规范），但返回值类型必须是 `int`。
- **用途**：当你需要定义一系列函数，它们的返回值都是整数，不过参数类型和数量不确定时，就可以使用这个类型别名进行类型注解。

#### 2. `type LabeledTuple[*Ts] = tuple[str, *Ts]  # TypeVarTuple`
```python
from typing import TypeVarTuple

Ts = TypeVarTuple('Ts')
type LabeledTuple[*Ts] = tuple[str, *Ts]

# 示例元组
labeled_int_tuple: LabeledTuple[int] = ('label', 1)
labeled_int_str_tuple: LabeledTuple[int, str] = ('label', 1, 'value')
```
- **含义**：`LabeledTuple` 是一个泛型类型别名，利用 `TypeVarTuple`（`Ts`）表示它可以包含任意数量和类型的元素。这个元组的第一个元素类型固定为 `str`，后续元素的类型由 `Ts` 决定。
- **用途**：当你需要创建带有标签的元组，且标签之后的元素数量和类型不固定时，可使用此类型别名。

#### 3. `type HashableSequence[T: Hashable] = Sequence[T]  # TypeVar with bound`
```python
from collections.abc import Sequence
from typing import TypeVar
from collections.abc import Hashable

T = TypeVar('T', bound=Hashable)
type HashableSequence[T: Hashable] = Sequence[T]

# 示例序列
hashable_list: HashableSequence[int] = [1, 2, 3]
```
- **含义**：`HashableSequence` 是一个泛型类型别名，`TypeVar`（`T`）被约束为 `Hashable` 类型，意味着 `T` 必须是可哈希的类型。该类型别名表示一个序列，序列中的元素类型为可哈希的 `T`。
- **用途**：当你需要处理包含可哈希元素的序列时，可使用此类型别名，确保序列中的元素都能用于需要可哈希对象的场景，如作为字典的键。

#### 4. `type IntOrStrSequence[T: (int, str)] = Sequence[T]`
```python
from collections.abc import Sequence
from typing import TypeVar

T = TypeVar('T', int, str)
type IntOrStrSequence[T: (int, str)] = Sequence[T]

# 示例序列
int_sequence: IntOrStrSequence[int] = [1, 2, 3]
str_sequence: IntOrStrSequence[str] = ['a', 'b', 'c']
```
- **含义**：`IntOrStrSequence` 是一个泛型类型别名，`TypeVar`（`T`）被限制为只能是 `int` 或者 `str` 类型。该类型别名表示一个序列，序列中的元素类型只能是整数或者字符串。
- **用途**：当你需要处理只包含整数或字符串元素的序列时，可使用此类型别名，增强代码的类型安全性和可读性。

这些类型别名定义借助 Python 的类型系统，能够让代码的类型注解更加灵活和精确，提高代码的可读性和可维护性。 

## 2. 更灵活的f字符串解析
主要新功能：
- [更灵活的 f 字符串解析](https://docs.python.org/3.12/whatsnew/3.12.html#pep-701-syntactic-formalization-of-f-strings)，允许许多以前不允许的事情 （PEP 701）。
    
    - 引号重用：
    ```python
    >>> songs = ['Take me back to Eden', 'Alkaline', 'Ascensionism']
    >>> f"This is the playlist: {", ".join(songs)}"
    'This is the playlist: Take me back to Eden, Alkaline, Ascensionism'
    ```
    - 无限制的嵌套
    ```python
    >>> f"""{f'''{f'{f"{1+1}"}'}'''}"""
    '2'
    ```
    - 多行表达式和注释：
    ```python
    >>> f"This is the playlist: {", ".join([
    ...     'Take me back to Eden',  # My, my, those eyes like fire
    ...     'Alkaline',              # Not acid nor alkaline
    ...     'Ascensionism'           # Take to the broken skies at last
    ... ])}"
    'This is the playlist: Take me back to Eden, Alkaline, Ascensionism'
    ```
    - 反斜杠和unicode字符：
    ```python
    >>> print(f"This is the playlist: {"\n".join(songs)}")
    This is the playlist: Take me back to Eden
    Alkaline
    Ascensionism
    >>> print(f"This is the playlist: {"\N{BLACK HEART SUIT}".join(songs)}")
    This is the playlist: Take me back to Eden♥Alkaline♥Ascensionism
    ```
    - 错误更加准确
    
        但错误消息不包括错误在行中的确切位置，并且还人为地将表达式括在括号中。在 Python 3.12 中，由于使用 PEG 解析器解析 f 字符串，因此错误消息可以更精确并显示整行：
    ```python
    >>> my_string = f"{x z y}" + f"{1 + 1}"
      File "<stdin>", line 1
      my_string = f"{x z y}" + f"{1 + 1}"
                   ^^^
    SyntaxError: invalid syntax. Perhaps you forgot a comma?
    ```
## 3. 支持python代码中的缓冲区协议
- 支持 Python 代码中的缓冲区协议 （[PEP 688](https://peps.python.org/pep-0688/)）

## 4. 每个解释器GIL
PEP 684 引入了按解释者指定的 GIL，因此现在可以为每个解释者创建具有唯一 GIL 的子解释器。这允许 Python 程序充分利用多个 CPU 内核。目前只能通过 C-API 使用，但预计 3.13 中将提供 Python API。

从结果来看：多解释器的运行时间，提高了60倍，期待3.13的提升。可以考虑在CPU的密集型任务当中，加入此功能（改成一个装饰器）

```python
import time
import multiprocessing
import ctypes

# 加载 Python 动态链接库
python_lib = ctypes.CDLL(None)

# 定义 PyInterpreterConfig 结构体
class PyInterpreterConfig(ctypes.Structure):
    _fields_ = [
        ("check_multi_interp_extensions", ctypes.c_int),
        ("gil", ctypes.c_int)
    ]

# 定义常量
PyInterpreterConfig_OWN_GIL = 1

# 定义函数原型
python_lib.Py_NewInterpreterFromConfig.argtypes = [ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(PyInterpreterConfig)]
python_lib.Py_NewInterpreterFromConfig.restype = ctypes.c_int

python_lib.PyStatus_Exception.argtypes = [ctypes.c_int]
python_lib.PyStatus_Exception.restype = ctypes.c_int

# 模拟 CPU 密集型任务
def cpu_intensive_task():
    start_time = time.time()
    while time.time() - start_time < 5:
        _ = 1 + 1

# 单解释器运行任务
def single_interpreter():
    cpu_intensive_task()

# 多解释器运行任务
def multi_interpreter():
    config = PyInterpreterConfig()
    config.check_multi_interp_extensions = 1
    config.gil = PyInterpreterConfig_OWN_GIL

    tstate = ctypes.c_void_p()
    status = python_lib.Py_NewInterpreterFromConfig(ctypes.byref(tstate), ctypes.byref(config))
    if python_lib.PyStatus_Exception(status):
        return -1
    cpu_intensive_task()

if __name__ == "__main__":
    num_cores = multiprocessing.cpu_count()

    # 单解释器测试
    single_start_time = time.time()
    processes = [multiprocessing.Process(target=single_interpreter) for _ in range(num_cores)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    single_end_time = time.time()
    single_total_time = single_end_time - single_start_time

    # 多解释器测试
    multi_start_time = time.time()
    processes = [multiprocessing.Process(target=multi_interpreter) for _ in range(num_cores)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    multi_end_time = time.time()
    multi_total_time = multi_end_time - multi_start_time

    print(f"单解释器总时间: {single_total_time} 秒")
    print(f"多解释器总时间: {multi_total_time} 秒")
    if multi_total_time < single_total_time:
        print("多解释器利用多核 CPU 更高效。")
    else:
        print("单解释器和多解释器性能相近或单解释器更优。")


""""
运行结果：
单解释器总时间: 6.34592080116272 秒
多解释器总时间: 0.10944199562072754 秒
多解释器利用多核 CPU 更高效。
"""
```

## 5. 理解式内联

![image.png](/static/img/8b4e481581c95e1aaa28799cc6482cd5.image.webp)

从详细的图片可以看出，之前的版本会在列表推导式、字典推导式，进行创建一个一次性函数对象，用完之后，进行丢弃。

而新的版本是：不需要单独的代码对象，也不再需要创建一次性的函数对象，也不再需要创建和销毁python框架。

**locals包含外部变量**
```python
>>> def f(lst):
...    return [locals() for x in lst]
...
>>> f(1)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 2, in f
TypeError: 'int' object is not iterable
>>> f([1])
[{'lst': [1], 'x': 1}]
>>>
```

对比3.11

```python
Python 3.11.10 (main, Oct  3 2024, 02:37:52) [Clang 14.0.6 ] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> def f(lst):
... ...    return [locals() for x in lst]
  File "<stdin>", line 2
    ...    return [locals() for x in lst]
    ^
IndentationError: expected an indented block after function definition on line 1
>>> def f(lst):
...    return [locals() for x in lst]
...
>>> f([1])
[{'.0': <list_iterator object at 0x10cb4e0b0>, 'x': 1}]
>>>
```

## 6. 与TypeHint相关的新功能
1.  PEP 692： 使用 TypedDict 进行更精确的 **kwargs 类型
在 PEP 484 引入的函数签名中键入 **kwargs 仅在所有 **kwargs 都属于相同类型的情况下才允许有效注释。

PEP 692 指定了一种更精确的键入 **kwargs 的方法，它依赖于类型化的字典：

```python
from typing import TypedDict, Unpack

class Movie(TypedDict):
  name: str
  year: int

def foo(**kwargs: Unpack[Movie]): ...
```


2. PEP 698： 覆盖静态类型的装饰器¶

增加一种，可以检查覆盖父类方法的，是否有调用。如果去掉override，在badchild中他是没有get_color,她就会调用base的方法。但实际上，你写的代码是为了调用badChild的。

so，这个方法还是对于你要覆盖父类方法有很大的用处
```python
from typing import override

class Base:
  def get_color(self) -> str:
    return "blue"

class GoodChild(Base):
  @override  # ok: overrides Base.get_color
  def get_color(self) -> str:
    return "yellow"

class BadChild(Base):
  @override  # type checker error: does not override Base.get_color
  def get_colour(self) -> str:
    return "red"


good = GoodChild()
bad = BadChild()
print(good.get_color())
print(bad.get_color())

```
## 7. canlendar的使用，增加月和日

[链接地址](https://docs.python.org/3.12/library/calendar.html#calendar.Month)
![image.png](/static/img/1e081f384bcad7d4c6c2a6aa88f52a30.image.webp)
## 8. itertools.batched，批量储量
将列表、字符串，批量分割成你规定的大小

```python
flattened_data = ['roses', 'red', 'violets', 'blue', 'sugar', 'sweet']
unflattened = list(batched(flattened_data, 2))
unflattened
[('roses', 'red'), ('violets', 'blue'), ('sugar', 'sweet')]

# batched('ABCDEFG', 3) → ABC DEF G

h
```
反过来：合并的方法是: `itertools.chain(*iterables)`



## 9. os.path.splitroot(path)

将路径名路径拆分为一个 3 项元组 （drive， root， tail），其中 drive 是设备名称或挂载点，root 是驱动器后面的一串分隔符，tail 是 root 之后的所有内容。这些项中的任何一项都可能是空字符串。在所有情况下，drive + root + tail 都将与 path 相同。

看情况，用起来是有点不准的。哈哈

```python
splitroot('C:/Users/Sam')
('C:', '/', 'Users/Sam')
splitroot('//Server/Share/Users/Sam')
('//Server/Share', '/', 'Users/Sam')
```

## 10. unittest,添加--durations,显示N个最慢的测试用例
```python
python3 -m unittest --durations=3 lib.tests.test_threading
.....
Slowest test durations
----------------------------------------------------------------------
1.210s     test_timeout (Lib.test.test_threading.BarrierTests)
1.003s     test_default_timeout (Lib.test.test_threading.BarrierTests)
0.518s     test_timeout (Lib.test.test_threading.EventTests)

(0.000 durations hidden.  Use -v to show these durations.)
----------------------------------------------------------------------
Ran 158 tests in 9.869s

OK (skipped=3)
```

## 11. uuid增加command-line 功能
```python
# generate a random uuid - by default uuid4() is used
$ python -m uuid

# generate a uuid using uuid1()
$ python -m uuid -u uuid1

# generate a uuid using uuid5
$ python -m uuid -u uuid5 -n @url -N example.com
```

将来有时间再加上3.12之前的PEP，还是挺有意思的👂🏻

📅 python3.12版本

## 设计思路

![image.png](/static/img/21912b889df73e1ceff804b4eccc2736.image.webp)
### 1.极简可插拔框架
OpenManus 的核心设计是构建一个非常精简的 Agent 框架，强调模块化和可扩展性。它通过可插拔的工具（Tools）和提示词（Prompt）的组合来定义 Agent 的功能和行为，降低了开发和定制 Agent 的门槛。

- Prompt 决定 Agent 的行为逻辑和思考方式；
- Tools 则提供行动能力（如计算机操作、代码执行、搜索等）。



通过对 Prompt 和 Tools 的自由组合，就能快速"拼装"出新的 Agent，赋予其处理不同类型任务的能力。

### 2. 工具驱动的 ReAct Agent
OpenManus 基于 ReAct（Reason + Act）模式，并以工具为核心驱动 Agent 的行动。Prompt 引导 Agent 的推理和逻辑，而 Tools 则赋予 Agent 行动能力。ToolCall Agent 的引入，进一步提升了工具使用的效率和规范性。

### 3. 规划能力处理复杂任务
OpenManus 延续了 Manus 的多智能体规划优势，将 PlanningTool 用于对用户需求进行高层规划。这种"先规划，后执行"的思路在复杂、长链任务上效果更佳。PlanningTool 将复杂的用户需求分解为线性的子任务计划，这种规划能力是处理现实世界复杂问题的关键。过去的研究表明，在相同模型能力下，如果缺乏系统的分解和规划，许多真实问题的成功率会大打折扣；而加入规划后，成功率会有显著提升。

### 4. 动态 Agent 分配与工具调度
当一个任务拆解出若干子任务后，系统会根据子任务类型，动态将其分配给预先定义或适配的 Agent（有各自的工具集和能力倾向）。这种 *临时分配 + 工具协作*的机制，可以最大化利用多模型、多工具的组合优势，提高应对不同问题场景的灵活度。Agent 预先装备了不同的工具集以应对不同类型的任务，提高了系统的灵活性和效率。


## 工作流程与执行路径
![640.png](/static/img/922813bb08d124d588e98775b113d5bb.640.webp)
### 1. 用户需求输入
用户在前端或命令行中输入复杂的需求，例如"写一段代码完成某种功能，并自动部署到服务器上"。

### 2. PlanningTool 规划
系统先调用 PlanningTool，对需求进行分析与分解，形成一个线性结构的计划或任务序列。比如，会将需求拆解为：
- 分析需求与环境
- 编写初始代码
- 测试并修复错误
- 部署并验证结果

这些子任务被记录在一个 plan 或类似结构中。

![openManus.png](/static/img/ed89b67a3dd0a4c2f1c81afa07e87737.openManus.webp)
### 3. 任务分配与执行

- 如果任务中涉及大规模数据分析或机器学习流程，可能会调用一个具备 Data Interpreter 能力的 Agent；
- 若任务需要复杂的代码修复或文件管理，则会调用另一个能够使用 ComputerUse 工具的 Agent；
- 系统按照顺序从计划中依次取出子任务；
- 根据任务关键字或意图判定，分配给最合适的 Agent。目前 Agent 分配主要基于正则匹配，未来考虑使用 LLM 实现更智能的任务分配。
- 每个 Agent 都会采用 ReAct 循环（Reason + Act）与 Tools 进行交互，以完成自己所负责的子任务。

### 4. 结果汇总与状态更新
当某个子任务执行完毕后，系统会将执行结果、关键上下文信息进行必要的"总结与压缩"（以避免不断增加的冗长 Memory），然后存入当前的"Plan 内存"或全局可访问的共享内存。

- 如果任务完成顺利，进入下一子任务；
- 若出现执行失败或结果异常，系统可进行自动调试或重新规划，视设计实现程度而定。

### 5. 整体产出
当所有子任务执行完毕，系统对整体结果进行汇总并返回给用户，或完成如网页部署、自动执行脚本等操作。


## 技术机构剖析

![image.png](/static/img/9b780551199b7d5c77df0ab7de3c9961.image.webp)

### 工程结构概览
```shell
.
├── app
│   ├── agent
│   │   ├── base.py
│   │   ├── manus.py
│   │   ├── planning.py
│   │   ├── react.py
│   │   ├── swe.py
│   │   └── toolcall.py
│   ├── flow
│   │   ├── base.py
│   │   ├── flow_factory.py
│   │   └── planning.py
│   ├── prompt
│   │   ├── manus.py
│   │   ├── planning.py
│   │   ├── swe.py
│   │   └── toolcall.py
│   └── tool
│       ├── base.py
│       ├── bash.py
│       ├── browser_use_tool.py
│       ├── create_chat_completion.py
│       ├── file_saver.py
│       ├── google_search.py
│       ├── planning.py
│       ├── python_execute.py
│       ├── run.py
│       ├── str_replace_editor.py
│       ├── terminate.py
│       └── tool_collection.py
```

项目依赖也相对简单，主要包括一些用于数据验证（pydantic）、AI 服务调用（openai）、浏览器控制（playwright、browsergym、browser-use）和一些基础工具库：
![image.png](/static/img/09c072581d6201171e1c8691f967b4c2.image.webp)
- pydantic：数据验证和设置管理
- openai：OpenAI API 的客户端库
- browser-use：构建能使用网络浏览器的 AI 代理框架
- browsergym：训练 AI 使用网络浏览器的环境
- playwright：浏览器自动化库
- googlesearch-python：无需 API 密钥进行搜索的库

这样的结构设计使得 OpenManus 在提供强大功能的同时保持了极高的可维护性和可扩展性。

### 4.2 核心系统组件
OpenManus 的架构由四个主要模块构成：
![image.png](/static/img/5f853623c93cf8ac4ca09c4431f2b797.image.webp)
1. 核心多智能体框架（Agent）
Agent 模块采用清晰的继承层次，自底向上逐步增强功能：

示例代码（Manus 实现）：

```python
class Manus(ToolCallAgent):
    """
    A versatile general-purpose agent that uses planning to solve various tasks.
    """
    name: str = "Manus"
    description: str = "A versatile agent that can solve various tasks using multiple tools"
    
    system_prompt: str = SYSTEM_PROMPT
    next_step_prompt: str = NEXT_STEP_PROMPT
    
    # Add general-purpose tools to the tool collection
    available_tools: ToolCollection = Field(
        default_factory=lambda: ToolCollection(
            PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
        )
    )
```
- BaseAgent：定义了智能体的基础属性（name、memory、system_prompt）和基本行为（执行逻辑、状态检查）。
- ReActAgent：实现了经典的 "Reasoning + Acting" 模式，先思考后行动，每一步执行都分为 think 和 act 两个阶段。
- ToolCallAgent：在 ReAct 基础上进一步细化，使 think 阶段专注于工具选择，act 阶段负责执行所选工具。
- Manus：继承 ToolCallAgent，主要通过定制 system_prompt 和 available_tools 来赋予不同能力。

2. Tools（工具层）
工具模块是 OpenManus 的行动能力基础，各类工具均继承自 BaseTool：

其中，planning.py 实现了 Manus 著名的计划功能，用 Markdown 格式管理任务计划并跟踪执行进度。

- ComputerUse：命令行和计算机操作
- BrowserUse：网络浏览和交互
- PythonExecute：执行 Python 代码
- GoogleSearch：网络搜索
- FileSaver：文件读写
- PlanningTool：任务规划与追踪

3. Prompt（提示词模块）
Prompt 模块包含了各种 Agent 使用的指令模板，例如 Planning 的系统提示：

```python
PLANNING_SYSTEM_PROMPT = """
You are an expert Planning Agent tasked with solving complex problems by creating and managing structured plans.
Your job is:
1. Analyze requests to understand the task scope
2. Create clear, actionable plans with the `planning` tool
3. Execute steps using available tools as needed
4. Track progress and adapt plans dynamically
5. Use `finish` to conclude when the task is complete

Available tools will vary by task but may include:
- `planning`: Create, update, and track plans (commands: create, update, mark_step, etc.)
- `finish`: End the task when complete

Break tasks into logical, sequential steps. Think about dependencies and verification methods.
"""
```
而 Manus 的系统提示则更加简洁：
```python
SYSTEM_PROMPT = "You are OpenManus, an all-capable AI assistant, aimed at solving any task presented by the user. 
You have various tools at your disposal that you can call upon to efficiently complete complex requests. 
Whether it's programming, information retrieval, file processing, or web browsing, you can handle it all."
```
4. Flow（执行流程模块）
- Flow 模块负责任务的高层编排和执行流程管理：
- PlanningFlow 的执行流程：
- 每步执行前，系统会生成上下文丰富的提示：
```python
step_prompt = f"""
CURRENT PLAN STATUS:
{plan_status}

YOUR CURRENT TASK:
You are now working on step {self.current_step_index}: "{step_text}"

Please execute this step using the appropriate tools. When you're done, provide a summary of what you accomplished.
"""
```
- BaseFlow：抽象基类，定义了 Agent 管理和执行接口
- PlanningFlow：实现基于规划的执行策略


1. 创建初始计划（_create_initial_plan）
2. 按计划步骤调用适当的 Agent
3. 跟踪计划执行状态并动态调整
![640.jpeg](/static/img/69514c13fdd6c52ead9819d16bebf4b5.640.webp)
### 4.3 系统运行机制

1. 基础版本（Manus）：

- 用户输入需求，调用 Manus agent 的 run 函数

- run 函数循环执行 step 操作（来自 ReActAgent）

- 每个 step 包含 think（选工具）和 act（执行工具）两个环节

- 直接使用基础工具集（Python执行、搜索、浏览器、文件保存等）

2. 高级版本（PlanningFlow）：

- 使用 PlanningTool 对需求进行整体规划

- 针对每个子任务动态生成适合的上下文和指令

- 调用 Manus agent 执行各个子任务

- 维护计划状态和执行进度

值得注意的是，在当前版本中，虽然 PlanningFlow 具备多智能体调度的能力，但实际上只有单一的 Manus 智能体在执行任务。未来版本可引入更多专业化的 Agent 以充分发挥多智能体协作的优势。

### 4.4 Memory 管理与 Agent 分配
与前文描述一致，OpenManus 实现了简单但有效的记忆管理和 Agent 分配机制：

- Memory 管理：每个子任务执行后进行总结压缩，避免上下文过长

- Agent 分配：当前主要基于正则匹配和规则，后续可考虑 LLM 辅助分配


## 未来优化与发展展望


虽然 OpenManus 在短时间内完成了初版本的核心功能，但团队也对后续的发展方向有清晰的思考，包括：

1. 增强规划能力

更深入地将 Tree-of-Thought 等搜索方法引入，或使用自适应的规划方案，让多智能体的任务分解与决策更高效、更精准。

2. 标准化评测 & 基准

在 GAIA、TAU-Bench、SWE-Bench 等多智能体任务集上进行持续评测，为后续工程优化提供量化标准，也能让社区更透明地了解 OpenManus 的改进幅度。

3. 多模型适配与成本优化

    - 从使用单一大模型（如 GPT、Claude）走向集成多个"专用小模型"（如针对代码、搜索、翻译、决策的模型），以更低成本实现超过大模型的效果；
    - 对 Token 消耗进行工程化、缓存化的优化，通过更精细的上下文管理显著降低调用成本。

4. 记忆管理与信息追踪

在多智能体长链任务中，如何高效追踪并管理上下文信息是关键挑战；团队将探索更灵活的 memory 结构，避免信息丢失与无效重复。

5. 更多工具生态 & 智能创造工具

随着工具数量与类型增加，Agent 对于工具选择、调用参数、执行顺序的管理将愈发复杂；如何让 Agent 自动创建或改写工具，也将成为下一阶段研究的热点。

6. 商业化场景深耕

多智能体在代码生成、网站搭建、数据分析等场景已有快速落地趋势。只有在"真实场景任务"和"用户效果"上极致打磨，才能促使用户持续使用 Agent 并产生商业价值。



## 与 MetaGPT 等项目的内在关联


OpenManus 在短时间里得以火速上线，与其团队早期在多智能体领域的系列研究息息相关。MetaGPT、Data Interpreter、AFlow、FACT、SELA、SPO 等项目不仅在论文、开源代码层面做了长期积累，也对多智能体的核心模块——规划、工具调用、搜索与记忆——进行了深入探索。其中：

- MetaGPT：最早的多智能体协作框架，将 SOP（标准操作流程）进行角色化分配，以提升大模型解决复杂任务的成功率。
- Data Interpreter：针对机器学习与数据建模场景，强化了智能体在数据处理、模型训练和工具使用上的能力。
- AFlow：利用蒙特卡洛树搜索（MCTS）在多任务中动态探索最优的 Agent 组合与执行流程。
- FACT：通过改进大模型的检索与记忆机制，增强多事实问答（QA）等任务的准确度。
- SELA：让智能体具有蒙特卡洛式的调试能力，在机器学习任务中自动优化策略。
- SPO：一种自监督的 Prompt 优化工具，适合短样本或质量难以度量的应用场景。

这些项目形成了一个庞大的多智能体技术储备，为 OpenManus 的快速构建提供了坚实的"技术底座"。

OpenManus的玩法

以下是为您整理的向量知识速查指南，包含核心概念、关键参数和调试技巧，帮助您快速上手向量相关开发：

---



### **一、向量搜索核心概念速览**
| **概念**          | **说明**                                                                 | **典型值/示例**                          |
|--------------------|-------------------------------------------------------------------------|----------------------------------------|
| **向量维度**       | 向量数组中元素的个数，决定模型的表达能力                                      | 768 (BERT)、1536 (OpenAI)              |
| **相似度度量**     | 计算向量间相似程度的方法                                                     | 余弦相似度、欧氏距离、内积                 |
| **归一化**         | 将向量转换为单位向量（模长为1）                                                | `vector /= np.linalg.norm(vector)`     |
| **ANN搜索**        | 近似最近邻搜索算法，加速海量向量检索                                            | HNSW、IVF-PQ                           |
| **索引类型**       | 向量数据库的存储结构                                                          | 平面索引（精确）、量化索引（压缩）           |

---

### **二、关键参数详解（以常见向量数据库为例）**

#### **1. 相似度阈值 (`min_score`)**
```python
# 示例：设置最低相似度阈值
results = vector_db.search(
    query_vector, 
    top_k=50, 
    min_score=0.6  # 只返回相似度≥0.6的结果
)
```

| **场景**           | **建议阈值范围** | **说明**                                |
|--------------------|-----------------|----------------------------------------|
| 精准匹配           | 0.8-0.95        | 法律条文、专利检索等需要高置信度的场景        |
| 一般语义搜索       | 0.6-0.8         | 电商商品搜索、文档检索等常见场景              |
| 模糊搜索           | 0.4-0.6         | 创意灵感推荐、相关性扩展等宽松场景            |

#### **2. 其他关键参数**
| **参数名**         | **作用**                                                                 | **调试技巧**                          |
|--------------------|-------------------------------------------------------------------------|--------------------------------------|
| `top_k`            | 控制返回结果数量                                                          | 结合业务需求动态调整，避免过多无效结果      |
| `ef_search`        | HNSW索引的搜索广度（平衡速度与精度）                                        | 值越大精度越高，但会降低搜索速度            |
| `metric_type`      | 指定相似度计算方法                                                         | 必须与索引构建时设置的度量类型一致           |
| `batch_size`       | 批量处理时的数据分片大小                                                    | 根据内存容量调整，过大会导致OOM错误          |

---

### **三、代码调试技巧**

#### **1. 可视化向量分布**
```python
import matplotlib.pyplot as plt
from sklearn.manifold import TSNE

def plot_vectors(vectors, labels):
    # 降维到2D空间
    tsne = TSNE(n_components=2)
    reduced = tsne.fit_transform(vectors)
    
    plt.figure(figsize=(10,6))
    plt.scatter(reduced[:,0], reduced[:,1], c=labels)
    plt.colorbar()
    plt.show()

# 使用示例
plot_vectors(all_vectors, cluster_labels)
```

#### **2. 相似度分布分析**
```python
def score_distribution(scores):
    plt.hist(scores, bins=20, alpha=0.7)
    plt.axvline(x=np.mean(scores), color='r', linestyle='--')
    plt.title("Similarity Score Distribution")
    plt.show()

# 在搜索结果上调用
scores = [res.score for res in search_results]
score_distribution(scores)
```

#### **3. 边界值测试用例**
```python
# 测试极端值输入
test_cases = [
    ("正常文本", "这是一段普通长度的测试文本"), 
    ("空文本", ""),
    ("超长文本", "。".join(["重复句子"]*1000)),
    ("特殊字符", "!@#$%^&*()")
]

for case_name, text in test_cases:
    try:
        vector = model.encode(text)
        print(f"{case_name} 向量形状: {vector.shape}")
    except Exception as e:
        print(f"{case_name} 处理失败: {str(e)}")
```

---

### **四、常见问题速查表**

| **现象**                  | **可能原因**                          | **解决方案**                          |
|---------------------------|--------------------------------------|--------------------------------------|
| 相似度全为1/-1            | 未做向量归一化                         | 添加归一化处理：`vector = vector / np.linalg.norm(vector)` |
| 搜索速度突然变慢          | 索引未优化或数据量超过阈值               | 重建索引并调整参数（如`efConstruction`）     |
| 相同输入得到不同向量       | 浮点数精度问题或模型随机性               | 设置随机种子，检查模型是否启用确定性模式        |
| 阈值过滤失效              | 度量类型不匹配（如用余弦阈值但实际使用内积） | 统一度量类型，进行分数转换：`cos_score = (dot_product + 1) / 2` |

---

### **五、最佳实践总结**
1. **预处理标准化**：
   ```python
   # 文本清洗+向量归一化
   text = clean_text(raw_input)  # 去停用词/特殊符号
   vector = model.encode(text)
   vector = vector / np.linalg.norm(vector)
   ```

2. **动态阈值调整**：
   ```python
   # 基于数据分布的自动阈值计算
   def auto_threshold(scores):
       q75 = np.percentile(scores, 75)
       q25 = np.percentile(scores, 25)
       return q25 - (q75 - q25) * 1.5  # 基于IQR的阈值
   ```

3. **混合搜索策略**：
   ```python
   # 结合关键词与向量搜索
   keyword_results = keyword_search(query)
   vector_results = vector_search(query_embedding)
   
   # 融合排序
   combined = hybrid_rerank(keyword_results, vector_results)
   ```

4. **监控关键指标**：
   ```python
   # 记录搜索质量指标
   log_data = {
       "query_time": search_latency,
       "recall@10": calculate_recall(ground_truth, results[:10]),
       "avg_score": np.mean([res.score for res in results])
   }
   ```

---

通过以上知识框架和代码示例，您可以快速定位和解决向量相关开发中的常见问题。建议在实际调试时结合可视化工具（如向量分布图、分数直方图）进行多维分析。

快速了解向量的相关知识

在处理大模型的计费与向量生成的Token计数时，其逻辑存在显著差异，具体分为以下几个层面：



---

### **1. 计费Token与向量生成Token的本质差异**
| **维度**               | **计费Token**                                   | **向量生成Token**                         |
|-------------------------|------------------------------------------------|------------------------------------------|
| **定义**                | API服务商根据输入输出内容计算的收费单位         | 模型处理输入文本时实际消耗的计算单位       |
| **计算范围**            | 通常包含输入和输出的总Token数（如GPT系列）      | 仅包含输入文本的Token数（如Embedding模型）|
| **标准化程度**          | 由服务商定义（如OpenAI使用`tiktoken`库）        | 依赖模型自身的分词器实现                  |
| **业务影响**            | 直接影响API调用成本                             | 反映模型计算复杂度                        |



**典型场景对比**：
```python
# OpenAI文本补全模型（计费包含输入输出）
response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}]
)
# 计费Token = 输入5 Tokens + 输出7 Tokens = 12 Tokens

# OpenAI Embedding模型（仅计输入）
response = openai.Embedding.create(
    input="Hello",
    engine="text-embedding-3-small"
)
# 计费Token = 输入5 Tokens（输出向量不占Token）
```

---

### **2. 不同模型架构的Token计算差异**
#### **(1) 基于Transformer的模型**
- **分词方式**：使用子词分词（如WordPiece/BPE）
- **特殊标记**：添加[CLS]/[SEP]等控制符
- **示例**：
  ```python
  from transformers import AutoTokenizer
  tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
  text = "Hello world"
  
  # BERT的分词结果（添加特殊标记）
  tokens = tokenizer.tokenize(text)
  # ['hello', 'world'] → 实际输入: [CLS] hello world [SEP] → Token数=4
  ```

#### **(2) 商业API模型**
- **OpenAI的分词器**：
  ```python
  import tiktoken
  enc = tiktoken.encoding_for_model("text-embedding-3-small")
  
  # 相同文本的Token计算
  len(enc.encode("Hello world"))  # 输出: 2（无特殊标记）
  ```

#### **(3) 传统词袋模型**
- **纯空格分词**：
  ```python
  text = "Hello world"
  tokens = text.split()  # ['Hello', 'world'] → Token数=2
  ```

---

### **3. 成本优化关键策略**
#### **(1) 精确Token预测**
```python
def estimate_openai_cost(text, model):
    """OpenAI计费Token预测工具"""
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

# 使用示例
text = "生成一段关于机器学习的说明"
cost_tokens = estimate_openai_cost(text, "text-embedding-3-small")
print(f"预估消耗Token: {cost_tokens}")
```

#### **(2) 多模型成本对比表**
| 模型类型               | 每百万Token成本 | 典型维度 | 适合场景              |
|-----------------------|----------------|---------|----------------------|
| OpenAI text-embedding-3-small | $0.02         | 1536    | 通用语义搜索          |
| Cohere embed-english-v3.0     | $0.15         | 1024    | 高精度语义匹配        |
| 自托管all-mpnet-base-v2       | $0（硬件成本） | 768     | 数据隐私敏感场景      |

#### **(3) 降维压缩技术**
```python
from sklearn.decomposition import PCA

def reduce_dimension(vec, target_dim):
    """向量降维工具"""
    pca = PCA(n_components=target_dim)
    return pca.fit_transform(vec.reshape(1, -1))[0]

# 使用示例
original_vec = np.random.rand(1536)  # OpenAI生成的高维向量
compressed_vec = reduce_dimension(original_vec, 512)  # 降维到512
```

---

### **4. 生产环境最佳实践**
#### **(1) Token监控系统**
```python
class TokenMonitor:
    """Token用量监控组件"""
    def __init__(self):
        self.usage = defaultdict(int)
    
    def track(self, model: str, tokens: int):
        self.usage[model] += tokens
    
    def generate_report(self):
        for model, total in self.usage.items():
            print(f"{model}: {total} Tokens (≈${total*0.02/1000:.2f})")

# 集成到向量服务
handler = VectorModelHandler()
monitor = TokenMonitor()

result = handler.get_embeddings(text, model)
monitor.track(model, result['tokens'])
```

#### **(2) 混合模型路由**
```python
def smart_router(text):
    """根据文本长度自动选择模型"""
    token_count = len(tokenizer.encode(text))
    
    if token_count < 128:
        return "openai/text-embedding-3-small"  # 低成本模型
    elif token_count < 512:
        return "hf/all-mpnet-base-v2"           # 平衡模型
    else:
        return "cohere/embed-english-v3.0"      # 长文本优化模型
```

#### **(3) 性能与成本的平衡公式**
$$
\text{模型选择权重} = \alpha \cdot \frac{\text{准确率}}{\text{成本}} + \beta \cdot \frac{\text{延迟}}{\text{基准}}
$$
其中：
- $\alpha$ = 0.7（业务精度权重）
- $\beta$ = 0.3（响应速度权重）

---

### **总结**
大模型的计费Token与向量生成的Token计算逻辑存在本质区别：
- **计费Token**：服务商定义的商业化计量单位，需关注输入输出总和
- **向量Token**：技术层面的处理单元，仅反映模型输入复杂度

实际应用中建议：
1. 通过`tiktoken`等工具精确预测商业API成本
2. 对自托管模型建立独立监控体系
3. 使用降维技术控制向量存储成本
4. 根据业务场景实施动态模型路由策略

大模型计费的token和向量模型的token有什么区别？

## commit的 Message的格式
```shell
<type>(<scope>): <subject>
<BLANK LINE>
<body>
<BLANK LINE>
<footer>
```



- type： commit的类别
- scope： commit影响的范围
- subject: 简短的描述



其中，header 是必需的，body 和 footer 可以省略。
不管是哪一个部分，任何一行都不得超过72个字符（或100个字符）。这是为了避免自动换行影响美观。

## Header
Header部分只有一行，包括三个字段：type（必需）、scope（可选）和subject（必需）。

### Type
用于说明 commit 的类别，只允许使用下面7个标识。

- feat：新功能（feature）
- fix：修补bug
- docs：文档（documentation）
- style： 格式（不影响代码运行的变动）
- refactor：重构（即不是新增功能，也不是修改bug的代码变动）
- test：增加测试
- chore：构建过程或辅助工具的变动
- perf: 性能（提高代码性能的改变）
- build: 影响构建系统或外部依赖项的更改（maven,gradle,npm等等）
- ci： 对CI配置文件和脚本的更改
- chore： 对非src和test目录的修改
- revert： Revert a commit


如果type为feat和fix，则该 commit 将肯定出现在 Change log 之中。

其他情况（docs、chore、style、refactor、test）由你决定，要不要放入 Change log，建议是不要。


### scope
scope用于说明 commit 影响的范围，比如`数据层、控制层、视图层`等等，视项目不同而不同。

例如在Angular，可以是$location, $browser, $compile, $rootScope, ngHref, ngClick, ngView等。

如果你的修改影响了不止一个scope，你可以使用*代替。

### subject
subject是 commit 目的的简短描述，不超过50个字符。

其他注意事项：

- 以动词开头，使用第一人称现在时，比如change，而不是changed或changes
- 第一个字母小写
- 结尾不加句号（.）


## Body
Body 部分是对本次 commit 的详细描述，可以分成多行。下面是一个范例。

```shell
如有必要，更详细的说明文本。包装它
大概72个字左右。

后面的段落在空行之后。

-要点也可以
-使用悬挂缩进

```
有两个注意点:

使用第一人称现在时，比如使用change而不是changed或changes。

- 永远别忘了第2行是空行
- 应该说明代码变动的动机，以及与以前行为的对比。

## Footer

Footer 部分只用于以下两种情况：

### 不兼容变动
如果当前代码与上一个版本不兼容，则 Footer 部分以BREAKING CHANGE开头，后面是对变动的描述、以及变动理由和迁移方法。
```
BREAKING CHANGE: isolate scope bindings definition has changed.

    To migrate the code follow the example below:

    Before:

    scope: {
      myAttr: 'attribute',
    }

    After:

    scope: {
      myAttr: '@',
    }

    The removed `inject` wasn't generaly useful for directives so there should be no code using it.
```
### 关闭 Issue
如果当前 commit 针对某个issue，那么可以在 Footer 部分关闭这个 issue 。
```
Closes #234
```


![image.png](/static/img/eb677a049b73a51ada9febe92d31ed57.image.webp)
## Revert
还有一种特殊情况，如果当前 commit 用于撤销以前的 commit，则必须以revert:开头，后面跟着被撤销 Commit 的 Header。
```git
revert: feat(pencil): add 'graphiteWidth' option

This reverts commit 667ecc1654a317a13331b17617d973392f415f02.
```
Body部分的格式是固定的，必须写成`This reverts commit &lt;hash>.`，其中的hash是被撤销 commit 的 SHA 标识符。

如果当前 commit 与被撤销的 commit，在同一个发布（release）里面，那么它们都不会出现在 Change log 里面。如果两者在不同的发布，那么当前 commit，会出现在 Change log 的Reverts小标题下面。

维度	计费Token	向量生成Token
定义	API服务商根据输入输出内容计算的收费单位	模型处理输入文本时实际消耗的计算单位
计算范围	通常包含输入和输出的总Token数（如GPT系列）	仅包含输入文本的Token数（如Embedding模型）
标准化程度	由服务商定义（如OpenAI使用`tiktoken`库）	依赖模型自身的分词器实现
业务影响	直接影响API调用成本	反映模型计算复杂度