AJ教程_站长资源网

为何最终我放弃了 Go 的 sync.Pool

Wed, 26 Nov 2025 17:22:04 +0800

声明: 本文并非否定 sync.Pool，而是分享技术选型的思考过程，帮助大家更准确地使用它

一、使用场景

一句话总结：保存和复用临时对象，减少内存分配，降低GC压力

1.1、引入：

举个简单的例子：

type User struct {
    ID       int64  `json:"id"`
    Username string `json:"username"`
    Email    string `json:"email"`
    Profile  [512]byte `json:"profile_data"` // 简介
}

var buf, _ = json.Marshal(
		User{
			ID: 1, 
			Username: "john_doe", 
			Email: "john@example.***",
		   },
		)

 user := &User{}
 json.Unmarshal(buf, user)

json的反序列化在数据解析和网络通信中非常常见，当程序并发度非常高的情况下，
短时间内需要创建大量临时对象。而这些临时对象都是分配在堆上的，会给GC造成很大的压力，严重影响程序的性能。
所以可以通过sync.Pool来解决。

1.2、什么是sync.pool？

Go语言，从1.3版本开始提供对象重用机制，即 sync.Pool。
sync.Pool 是 sync 包下的一个组件，可以作为保存临时取还对象的一个“池子”。
同时sync.Pool是可伸缩且并发安全的，他的大小受限于内存的大小。sync.Pool用于存储那些被分配了但是没有被使用，而未来还会使用的值。

这样就不用再次经过内存分配，而是直接复用对象，减轻GC压力，从而提升性能。

但个人觉得它的命名可能造成误解，因为 Pool 里装的对象可以被无通知地被回收，可能 sync.Cache(临时缓存) 是一个更合适的名字。

二、如何使用

sync.Pool 的使用方式非常简单：

2.1、声明对象池

只需要实现New函数即可，当对象池(sync.Pool)中没有对象时，就会自动调用New函数进行。

var userPool = sync.Pool{
    New: func() interface{} { 
        return new(User) 
    },
}

2.2、GET & PUT

// 取出
user := userPool.Get().(*User) 
json.Unmarshal(buf,user)
// 放回
userPool.Put(user)

Get() 用于从对象池中获取对象，因为返回值是 interface{}，因此需要类型转换。
Put() 则是在对象使用完毕后，返回对象池。

三、实例：

3.1、标准库中的应用

3.1.1: fmt.Printf

Go语言标准库大量使用了sync.Pool，例如: fmt和encoding/json
以下是fmt.Printf的源代码(go/src/fmt/print.go) - 你也可以到本地Go源码自行查看

// go 1.13.6

// pp is used to store a printer's state and is reused with sync.Pool to avoid allocations.
// pp用于存储打印机的状态，并与sync.Pool一起重用。以避免分配。
type pp struct {
    buf buffer
    ...
}

var ppFree = sync.Pool{
	New: func() interface{} { return new(pp) },
}

// newPrinter allocates a new pp struct or grabs a cached one.
// newPrinter分配了一个新的pp结构体或获取一个缓存的pp结构体。
func newPrinter() *pp {
	p := ppFree.Get().(*pp)
	p.panicking = false
	p.erroring = false
	p.wrapErrs = false
	p.fmt.init(&p.buf)
	return p
}

// free saves used pp structs in ppFree; avoids an allocation per invocation.
// 在ppFree中保存使用过的pp结构体；避免每次调用分配。
func (p *pp) free() {
	if cap(p.buf) > 64<<10 {
		return
	}

	p.buf = p.buf[:0]
	p.arg = nil
	p.value = reflect.Value{}
	p.wrappedErr = nil
	ppFree.Put(p)
}

func Fprintf(w io.Writer, format string, a ...interface{}) (n int, err error) {
	p := newPrinter()
	p.doPrintf(format, a)
	n, err = w.Write(p.buf)
	p.free()
	return
}

// Printf formats a***ording to a format specifier and writes to standard output.
// Printf根据格式说明符进行格式化，并写入标准输出。
// It returns the number of bytes written and any write error encountered.
// 返回写入的字节数和遇到的任何写入错误。
func Printf(format string, a ...interface{}) (n int, err error) {
	return Fprintf(os.Stdout, format, a...)
}

fmt.Printf 的调用是非常频繁的，利用 sync.Pool 复用 pp 对象能够极大地提升性能，减少内存占用，同时降低 GC 压力。

3.2、Gin框架的应用(context)

在Gin框架中，Context 对象代表了处理一个HTTP请求的上下文。每个请求都需要一个Context，请求处理完毕，Context的生命周期也就结束了。

高频的创建于销毁：在高并发下，每秒会创建和销毁大量Context对象。
固定生命周期：Context的生命周期始于请求到来，止于请求处理完毕，非常短暂。

3.2.1、定义对象池

在gin.Engine结构体的定义中，你可以看到pool字段就是一个sync.Pool

type Engine struct {
    // ... 其他字段
    pool sync.Pool // context 对象池
}
如下：

3.2.2、初始化对象池

在创建Gin引擎实例的时，会初始化sync.Pool，并指定New函数。
当池子中无对象可用的时，会调用此函数创建新的Context。

func New() *Engine {
    // ...
    engine.pool.New = func() any {
        return engine.allocateContext(engine.maxParams)
    }
    return engine
}

func (engine *Engine) allocateContext(maxParams uint16) *Context {
    // 分配并初始化一个Context
    v := make(Params, 0, maxParams)
    return &Context{engine: engine, params: &v, skippedNodes: &skippedNodes}
}

3.2.3、从池中获取Context

当HTTP请求到达时，Gin会从sync.Pool中获取一个Context对象。

func (engine *Engine) ServeHTTP(w http.ResponseWriter, req *http.Request) {
    // 从对象池中获取一个 context[citation:7]
    c := engine.pool.Get().(*Context)
    c.writermem.reset(w)
    c.Request = req
    c.reset()
    // ... 处理 http 请求
    engine.handleHTTPRequest(c)
    // 把 context 放回对象池[citation:7]
    engine.pool.Put(c)
}

3.2.4、处理请求后放回池中

请求处理完毕后，Gin会将Contex重置并放回sync.Pool中，以供后面复用。

func (engine *Engine) ServeHTTP(w http.ResponseWriter, req *http.Request) {
    c := engine.pool.Get().(*Context)
    // ... 处理 http 请求
    engine.handleHTTPRequest(c)
    // 请求处理完成后，将 Context 放回池中[citation:7]
    engine.pool.Put(c)
}

切记，重点是要重置的，如调用c.reset()。确保放回的是干净的上下文。

四、我在项目中的实战

4.1、为何最初选择sync.Pool

因以后其他博客还会提及，所以这里就简洁的说一下：
我的目的：

设计了一个支持多存储驱动的图片上传模块，重点解决了并发性能、资源管理和动态切换的问题

为了解决所谓的高并发，复用实例的问题，我就自然的想到去使用sync.Pool，但问题来了！

对象复用：避免频繁创建和销毁对象
并发安全：多个用户可同时使用不同驱动

为此，我还美滋滋的，描绘了一个草图：

// 多驱动对象池管理器
type MultiDriverPool struct {
    pools   map[string]ObjectPool
    mu      sync.RWMutex
    current string // 当前默认驱动
}
// 对象池接口
type ObjectPool interface {
    Get() (Driver, error)
    Put(Driver) error
    Close()
    Size() int
    Available() int
}

4.2、又为何选择放弃sync.pool

4.2.1、存储驱动通常是无状态的

比如： 七牛云驱动使用相同的A***essKey和SecretKey，每个实例都执行相同的操作，没有必要维护多个实例。实际上，一个驱动实例就可以处理所有请求，而且通常驱动本身是线程安全的（或者可以通过在方法内部分配资源来做到线程安全）

**换句话说就是：**认为每个驱动实例需要频繁创建和销毁，但实际上驱动实例是可以复用的，而且创建成本不高，并且“存储驱动是无状态的” ！

所以我最终的设计模式是：单例+多驱动模式。

五、总结

适合 sync.Pool 的场景：

创建成本高 对象初始化有显著开销
生命周期短 使用后很快就不再需要
使用频率高 大量并发创建销毁
可安全重置 能完全清理之前的状态

不适合：

// 1、存储驱动 - 创建成本低，生命周期长
var driverPool = sync.Pool{
    New: func() interface{} { return &QiniuDriver{} },
}

// 2、 数据库连接 - 需要连接池，不是对象池
var dbPool = sync.Pool{
    New: func() interface{} { return sql.Open(...) },
}

// 3、配置对象 - 长期存在，不需要频繁创建
var configPool = sync.Pool{
    New: func() interface{} { return loadConfig() },
}

在结尾处，我在声明一下：
sync.Pool 的核心作用，不是资源管理。
而是通过保存和复用临时对象，减少内存分配，降低GC压力！

六、sync.Pool的底层剖析

6.1 底层结构体

// [Go 内存模型]: https://go.dev/ref/mem
type Pool struct {
	noCopy noCopy

	local     unsafe.Pointer // 每个 P 的本地固定大小池，实际类型是 [P]poolLocal
	localSize uintptr        // 本地数组的大小

	victim     unsafe.Pointer // 上一个周期的本地池
	victimSize uintptr        // victim 数组的大小

	// New 可以选择性地指定一个函数，用于在 Get 否则会返回 nil 时生成一个值。
	// 不能在与 Get 调用并发的情况下修改此函数。
	New func() any
}

6.2 重点

在Pool的底层，核心有两点：分别是local与victim

6.2.1 local unsafe.Pointer

local 是一个按 P（GOMAXPROCS）分片的本地对象池。
每个 P 都有自己的 poolLocal，无需锁，极快。
Get / Put 操作优先访问本地池，不需要加锁。

6.2.2 victim

Go 认为 Pool 内的对象是可丢弃的，所以每次 GC 会清空 pool.local。
为了避免冲击（比如刚清空就马上又需要大量对象），Go 引入了：上一 GC 周期的 pool.local 备份。
避免 GC 后对象全部被清空导致性能抖动。
victim的大致流程如下：

七、性能测试

7.1 测试主函数


type User struct {
	ID       int64     `json:"id"`
	Username string    `json:"username"`
	Email    string    `json:"email"`
	Profile  [512]byte `json:"profile_data"`
}

// 创建 User Pool
var userPool = sync.Pool{New: func() interface{} {
	atomic.AddUint64(&poolMisses, 1)
	return new(User)
}}

// 创建 Buffer Pool
var bufPool = sync.Pool{New: func() interface{} {
	return new(bytes.Buffer)
}}

// 调用Get的次数
var totalGets uint64

// 必须创建新对象的次数
var poolMisses uint64

// 获取一个 User
func getUser() *User {
	atomic.AddUint64(&totalGets, 1)
	return userPool.Get().(*User)
}

// 放回 User
func putUser(u *User) {
	// 1、清空数据
	u.ID = 0
	u.Username = ""
	u.Email = ""
	for i := range u.Profile {
		u.Profile[i] = 0
	}
	// 2、放回
	userPool.Put(u)
}

// 处理 User
func processUser(data []byte) *User {
	u := getUser()
	_ = json.Unmarshal(data, u)
	return u
}

// 处理 HTTP 请求
func handleProcess(w http.ResponseWriter, r *http.Request) {
	// 1、获取 Buffer
	var b bytes.Buffer

	// 2、获取 User
	_, _ = io.Copy(&b, r.Body)

	// 3、处理 User
	u := processUser(b.Bytes())
	defer putUser(u)
	// ...
}

func handleMetrics(w http.ResponseWriter, _ *http.Request) {
	hits := atomic.LoadUint64(&totalGets) - atomic.LoadUint64(&poolMisses)
	_, _ = w.Write([]byte("sync_pool_gets " + strconv.FormatUint(atomic.LoadUint64(&totalGets), 10) + "\n"))
	_, _ = w.Write([]byte("sync_pool_misses " + strconv.FormatUint(atomic.LoadUint64(&poolMisses), 10) + "\n"))
	_, _ = w.Write([]byte("sync_pool_hits " + strconv.FormatUint(hits, 10) + "\n"))
}

func main() {
	http.HandleFunc("/process", handleProcess)
	http.HandleFunc("/metrics", handleMetrics)
	_ = http.ListenAndServe(":8080", nil)
}

7.2 对象的复用率


// TestHTTPConcurrent
// 测试结果：totalGets=500, poolMisses=28
func TestHTTPConcurrent(t *testing.T) {
	data := []byte(`{"id":10,"username":"concurrent","email":"c@example.***"}`)
	req := httptest.NewRequest(http.MethodPost, "/process", bytes.NewBuffer(data))

	n := 500
	var wg sync.WaitGroup
	wg.Add(n)
	for i := 0; i < n; i++ {
		go func() {
			defer wg.Done()
			w := httptest.NewRecorder()
			handleProcess(w, req)
			if w.Code != http.StatusOK {
				t.Errorf("bad status")
			}
		}()
	}
	wg.Wait()

	t.Logf("totalGets=%d, poolMisses=%d", totalGets, poolMisses)
}

=== RUN TestHTTPConcurrent
pool_test.go:65: totalGets=500（调用get的总次数）,poolMisses=4（新new的次数）
— PASS: TestHTTPConcurrent (0.00s)
PASS
但若大家自己测，由于处于不同环境，结果应该会有些许波动。

7.3 对象复用性能测试

采用 基准测试 ：用来测性能的测试，包括耗时、内存分配、GC 压力等

// b.N 是测试循环次数（Go 自动调整）
// b.ReportAllocs()：显示内存分配次数
// b.ResetTimer()：重置计时器（忽略前面初始化的耗时）

// ------------------------
// Benchmark - 无对象池
// ------------------------
func BenchmarkWithoutPool(b *testing.B) {
	data := []byte(`{"id":123,"username":"user123","email":"user123@example.***"}`)
	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		u := &User{}
		_ = json.Unmarshal(data, u)
	}
}

// ------------------------
// Benchmark - 对象池
// ------------------------
func BenchmarkWithPool(b *testing.B) {
	data := []byte(`{"id":123,"username":"user123","email":"user123@example.***"}`)
	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		u := getUser()
		_ = json.Unmarshal(data, u)
		putUser(u)
	}
}

测试项	ns/op（每次操作耗时）	B/op（分配内存字节数）	allocs/op（分配次数）
WithoutPool	721 ns	816 B	7 allocs
WithPool	664 ns	240 B	6 allocs

可以从，B/op（分配内存字节数），近4倍的差距，看出性能的差距。
当然大家自测时，应该会出现偏差，要以withoutPool与withPool的差距作为对比标准。

八、自测

sync.Pool 的主要作用是什么？为什么它能减少 GC 压力？
sync.Pool.New 是在什么情况下被调用的？
为什么从 pool 取出的对象必须重置（reset）？
为什么在你的代码中，putUser() 必须把结构体所有字段清空？(原：str = “” 设为空)
为什么结构体字段清空了却依旧要 Reset()？
sync.Pool 为什么不是普通的缓存？它有什么生命周期特性？

借鉴：
1、Go 语言高性能编程 - sync.pool
2、深度解密 Go 语言之 sync.Pool

【MCP探索实践】Google GenAI Toolbox：Google开源的企业级AI数据库中间件、5分钟搞定LLM-SQL安全互联

Wed, 26 Nov 2025 17:22:02 +0800

系列篇章💥

No.	文章
1	【MCP探索实践】Cherry Studio+MCP实战：3步让AI自动抓网页/读文件/调API
2	【MCP探索实践】FastAPI + MCP：2025年最火的后端与AI集成方案
3	【MCP探索实践】GitHub MCP Server：为开发者打造的高效自动化工具
4	【MCP探索实践】MoLing：零依赖跨平台办公自动化神器，3分钟搞定文件+浏览器双核操作
5	【MCP探索实践】3分钟搭建AI服务器！FastMCP让开发效率飙升10倍
6	【MCP探索实践】MindsDB：借助 MCP 协议，让 AI 大模型秒变 SQL 专家
7	【MCP探索实践】Web Search MCP Server：无需 API 密钥的免费网络搜索服务
8	【MCP探索实践】百度地图 MCP Server：告别繁琐集成、让地图服务接入更简单
9	【MCP探索实践】MCP生态下的LangChain适配器：AI开发的“加速引擎”，多工具集成一步到位
10	【MCP探索实践】OpenMemory MCP：如何用MCP协议解锁AI工具的跨平台记忆共享
11	【MCP探索实践】Playwright MCP：微软打造的AI自动化利器，一键搞定浏览器操作
12	【MCP探索实践】ROS MCP Server：自然语言控制机器人，从此告别复杂指令！
13	【MCP探索实践】蚂蚁AntV开源的可视化图表MCP Server Chart：高效数据可视化的利器
14	【MCP探索实践】Firecrawl MCP Server：为LLM客户端赋能的开源Web爬虫服务器，数据采集效率提升10倍
15	【MCP探索实践】MiniMax MCP Server：多模态生成服务器，让AI同时玩转视频/语音/图像生成
16	【MCP探索实践】Bright Data MCP：实时、安全、智能，网络数据抓取的三剑客来袭
17	【MCP探索实践】Chrome MCP Server：基于Chrome扩展的AI浏览器自动化神器——技术解析与实践指南
18	【MCP探索实践】Redis官方MCP Server：用自然语言驱动Redis的AI原生存储引擎
19	【MCP探索实践】MCP MongoDB Server：让LLM与MongoDB无缝交互
20	【MCP探索实践】Windows-MCP：开源 AI Agent 一键打通 Windows 全接口
21	【MCP探索实践】mcp-installer：一键部署MCP服务器的高效工具
22	【MCP探索实践】MCP-Shield：守护MCP服务器安全的利器
23	【MCP探索实践】Google GenAI Toolbox：Google开源的企业级AI数据库中间件、5分钟搞定LLM-SQL安全互联

前言

随着生成式 AI 进入生产环境，开发者急需一种“低代码、高安全、可观测”的方式来把 LLM 与关系型数据库打通。Google 2024 年开源的 genai-toolbox（MCP Toolbox for Databases）正是为此而生，它通过统一的服务端代理 + 多语言 SDK，把 SQL 查询包装成 LLM 可调用的工具，10 行代码即可上线，极大降低 RAG、智能报表、Agent 等场景的开发门槛。

一、项目概述

genai-toolbox 是一款面向企业级场景的 MCP（Model-Context-Protocol）服务器开源工具箱，用 Go 语言实现，内置连接池、身份鉴权、OpenTelemetry 链路追踪，可把 PostgreSQL、AlloyDB 等数据库表/视图快速映射成 LLM 可调用的 Function Calling 工具，并支持 Python、Node.js、Go、Java 等多语言 SDK 集成。

二、技术原理

（一）、整体架构

Server 端：解析 YAML 配置，建立数据库连接池；暴露 RESTful API：/loadToolset、/invokeTool 等；内嵌 auth 中间件（OAuth2/JWT）与 OpenTelemetry Collector，实现零侵入可观测。

Client SDK：封装 HTTP 调用，提供异步 loadToolset()；将工具元数据（name、description、JSONSchema）转成 LangChain、LlamaIndex、Genkit 等框架的 Tool 对象。

MCP 协议：通过统一的 JSON-RPC 风格协议，让 LLM 在对话中以 Function Calling 方式调用 SQL，Server 端负责参数校验、SQL 预编译、结果序列化。

（二）、性能与安全

连接池复用 + Prepared Statement 防注入；

支持 IAM 集成、SSL/TLS 加密、行级权限控制；

提供开箱即用的 Prometheus Metrics（latency、qps、error rate）。

三、主要功能

（一）、核心能力

零代码 SQL 转换工具：只需在 tools.yaml 文件中声明 SQL 语句及其参数，系统便能自动生成相应工具，显著降低了开发成本和技术门槛，让开发者无需编写大量代码即可实现功能。

多数据源支持：全面支持多种数据库，包括 PostgreSQL、AlloyDB、Cloud SQL 以及处于实验阶段的 MySQL，能够满足不同企业多样化的数据库使用需求，为企业的数据管理提供了更多的选择和灵活性。

工具集高效管理：工具集按照业务模块进行分组，支持版本化发布与灰度更新。这使得在工具的更新和维护过程中，可以更加精准地控制范围，降低风险，确保系统的稳定性和可靠性。

全生态 SDK 覆盖：提供了丰富的 SDK 支持，涵盖 Python、Node.js、Go、Java 等多种编程语言，以及 LangChain、LlamaIndex、Genkit 等主流框架。无论开发者使用何种技术栈，都能方便地集成该工具，实现与系统的无缝对接。

端到端观测能力：具备完整的端到端观测体系，集成了 OpenTelemetry Trace、Prometheus Metrics 以及结构化日志。通过这些观测手段，开发者可以实时监控系统的运行状态，及时发现并解决潜在问题，保障系统的稳定运行。

（二）、高级特性

向量 SQL 功能：内置 text_embedding() 调用，支持向量相似度检索（借助 pgvector 扩展）。这一特性使得系统能够处理复杂的语义搜索和分析任务，为企业提供更智能、高效的数据分析服务。

事务级工具支持：在一次对话内，多个工具可以共享同一连接事务。这种设计确保了数据操作的一致性和完整性，避免了因事务处理不当而导致的数据错误，提高了系统的可靠性和数据安全性。

动态参数校验机制：采用 JSONSchema 与 SQL 类型映射的方式，实现动态参数校验。在运行时，系统能够自动检查参数的合法性，减少因参数错误而导致的运行时错误，提高了系统的稳定性和健壮性。

热加载功能：当配置发生修改时，无需重启系统，配置修改可在秒级内生效。这一特性大大提高了系统的灵活性和可维护性，使得开发者能够快速响应业务需求的变化，及时调整系统配置。

四、应用场景

（一）、RAG 知识库

在企业知识管理领域，可将企业知识库表精准映射为 search_docs 工具。当用户提出问题时，大语言模型（LLM）能够依据问题实时召回排名前 K 的相关段落，为用户提供准确且高效的知识检索服务，极大地提升了企业内部知识的利用效率。

（二）、NL2SQL 数据助手

对于运营人员和分析师而言，无需再花费大量时间学习和编写复杂的 SQL 语句。他们可以直接使用自然语言来查询订单信息、库存状况以及用户行为数据等。通过该工具，自然语言能够被准确转化为 SQL 查询，为数据分析和决策提供了极大的便利。

（三）、智能客服 Agent

智能客服场景中，结合订单表、物流表以及知识库表等多源数据，智能客服 Agent 能够实现诸如“查订单→改地址→退差价”等多步决策流程。在与客户的交互过程中，它可以根据客户需求，灵活调用不同的数据表，提供一站式的优质服务。

（四）、低代码 BI

在商业智能领域，前端用户可以通过简单的拖拽操作来生成查询条件。而后端则会调用 Toolbox 工具，将查询结果以 JSON 格式返回，直接用于图表的渲染。这种低代码的方式大大降低了 BI 开发的门槛，使得业务人员也能够轻松实现数据可视化。

（五）、AIOps

在运维管理方面，SRE（站点可靠性工程师）只需在 Slack Bot 中输入如“最近 10 分钟错误率最高的服务”这样的自然语言指令，Toolbox 就能实时查询 Prometheus 落地表，并迅速返回查询结论，帮助运维人员及时发现和解决系统问题，保障系统的稳定运行。

五、快速使用

（一）、环境准备

OS：Linux/macOS/Windows WSL2；
Docker ≥ 20.10 或直接下载二进制；
PostgreSQL 12+（或 AlloyDB）已运行，示例数据库 toolbox_db 已创建。

（二）、5 分钟上手

下载二进制

export VERSION=0.2.0
curl -O https://storage.googleapis.***/genai-toolbox/v${VERSION}/linux/amd64/toolbox
chmod +x toolbox

创建 tools.yaml

sources:
  my-pg:
    kind: postgres
    host: 127.0.0.1
    port: 5432
    database: toolbox_db
    user: postgres
    password: postgres

tools:
  search_user:
    kind: postgres-sql
    source: my-pg
    description: 根据姓名模糊查询用户
    parameters:
      - name: name
        type: string
    statement: SELECT id, name, email FROM users WHERE name ILIKE '%' || $1 || '%';

启动服务

./toolbox --tools_file tools.yaml --port 5000

Python 客户端调用

pip install toolbox-core
from toolbox_core import ToolboxClient
import asyncio

async def main():
    async with ToolboxClient("http://127.0.0.1:5000") as client:
        tools = await client.load_toolset("default")
        result = await tools["search_user"].invoke({"name": "alice"})
        print(result)

asyncio.run(main())

LangChain 集成（可选）

pip install toolbox-langchain
from toolbox_langchain import ToolboxClient
client = ToolboxClient("http://127.0.0.1:5000")
tools = client.load_toolset()
agent = initialize_agent(tools, llm, agent="zero-shot-react-description", verbose=True)
agent.run("帮我找出所有名字包含 alice 的用户")

（三）、Docker 一键部署

docker run -d --name toolbox \
  -p 5000:5000 \
  -v $(pwd)/tools.yaml:/tools.yaml \
  ghcr.io/googleapis/genai-toolbox:v0.2.0 \
  --tools_file /tools.yaml

（四）、Kuber***es 生产级部署示例

apiVersion: apps/v1
kind: Deployment
metadata:
  name: toolbox
spec:
  replicas: 3
  selector:
    matchLabels: { app: toolbox }
  template:
    metadata:
      labels: { app: toolbox }
    spec:
      containers:
      - name: toolbox
        image: ghcr.io/googleapis/genai-toolbox:v0.2.0
        args: ["--tools_file=/config/tools.yaml"]
        ports: [{ containerPort: 5000 }]
        volumeMounts:
        - name: config
          mountPath: /config
      volumes:
      - name: config
        configMap:
          name: toolbox-config

配合 HorizontalPodAutoscaler 可根据 QPS 自动扩缩容。

（五）、常见踩坑与排查

连接拒绝：确认 PostgreSQL 监听 0.0.0.0 且防火墙放行 5432；
工具未找到：检查 toolset 名称是否匹配，或执行 ./toolbox validate --tools_file tools.yaml 做预检；
高并发超时：在 YAML 中调大 max_connections 并开启连接池 pool_size: 20。

结语

genai-toolbox 把“让 LLM 安全、高效地访问数据库”这一复杂命题封装成一条 YAML + 十行代码，使开发者得以专注业务逻辑而非底层连接、鉴权与可观测。随着 Google 社区持续迭代（路线图已规划支持 BigQuery、Spanner、Cloud SQL Auth Proxy），它有望成为 GenAI 时代数据库中间件的事实标准。现在就动手试试吧！

项目地址

GitHub 源码：https://github.***/googleapis/genai-toolbox
官方文档：https://cloud.google.***/alloydb/docs/genai-toolbox
Codelabs 实战：https://codelabs.developers.google.***/genai-toolbox-for-alloydb

🎯🔖更多专栏系列文章：AI大模型提示工程完全指南、AI大模型探索之路（零基础入门）、AI大模型预训练微调进阶、AI大模型开源精选实践、AI大模型RAG应用探索实践🔥🔥🔥 其他专栏可以查看博客主页📑

😎 作者介绍：资深程序老猿，从业10年+、互联网系统架构师，目前专注于AIGC的探索（CSDN博客之星|AIGC领域优质创作者）
📖专属社群：欢迎关注【小兵的AI视界】公众号或扫描下方👇二维码，回复‘入群’ 即刻上车，获取邀请链接。
💘领取三大专属福利：1️⃣免费赠送AI+编程📚500本，2️⃣AI技术教程副业资料1套，3️⃣DeepSeek资料教程1套🔥（限前500人）
如果文章内容对您有所触动，别忘了点赞、⭐关注，收藏！加入我们，一起携手同行AI的探索之旅，开启智能时代的大门！

π0源码(openpi)剖析——从π0模型架构的实现：如何基于PaLI-Gemma和扩散策略去噪生成动作，到基于C/S架构下的模型训练与部署

Wed, 26 Nov 2025 17:21:59 +0800

前言

ChatGPT出来后的两年多，也是我疯狂写博的两年多(年初deepseek更引爆了下)，比如从创业起步时的15年到后来22年之间每年2-6篇的，干到了23年30篇、24年65篇、25年前两月18篇，成了我在大模型和具身的原始技术积累

如今一转眼已到25年3月初，时光走得太快，近期和团队接了好几个大客户订单，使得3月起不得不全力加速落地，自己也得每天抠paper、搞代码

so，为何在明明如此之忙一天当两天用的情况下，还要继续努力更新博客呢？

原因在于

一方面，我确实喜欢分享，因为写博的这10多年下来确实可以帮到很多、很多人，不然本博客也不会有如今如此巨大的访问量与影响力
更何况有些文章是之前既定计划中的，在本文之前，上一篇关于π0的文章是π0_fast《π0开源了且推出自回归版π0-FAST——打造机器人动作专用的高效Tokenizer：比扩散π0的训练速度快5倍但效果相当》，文中提到，会解读π0的源码

至于什么是π0 详见此文《π0——用于通用机器人控制的VLA模型：一套框架控制7种机械臂(基于PaliGemma和流匹配的3B模型)》
二方面，我司「七月在线」在做一系列工厂落地场景的过程中，我们也希望团结到可以和我们一块做的朋友，而若想团结，便需要借助博客顺带分享我们每个季度在重点做的业务场景

比如过去一周，我把lerobot、reflect vlm、π0的仿真环境都在我自己本地电脑上跑了下

过程中，GitHub copilot这种AI编程工具在环境的安装上帮了我很大的忙——各种环境只要几句命令，直接帮我装好，真心不错)

如此硬着头皮冥思苦想、摸索了好几天，随后使得我自己知道怎么带队完成『太多工厂希望实现的一个生产线任务』了，3月初先仿真训练，2-3个月内部署到真机

当然了，也不单纯只是「这几天的想」就能想出来的，这几天之前

有把过去一年当三年用的具身技术积累

有一年多来，和同事们如姚博士，以及朋友们许多的讨论

有去年十几个工厂对我们的支持与信任

我们正在不断壮大队伍

有我司内部同事，亦有我联合带的北理、中南等985的具身研究生，及一块合作开发的朋友，很快会把多个生产线任务并行开发起来

且无论哪个项目，都是不断长期迭代的，故过程中少不了科研层面的突破，欢迎更多伙伴加入我们(全职、兼职、实习皆可，有意者，敬请私我)，和我们一块开发

话休絮烦，本文便按照如下图所示的源码结构，重点解读一下π的整个源码「 π0及π0-FAST的GitHub地址：github.***/Physical-Intelligence/openpi」

π0的源码结构非常清晰、可读性高，不愧是成熟的商业化公司，是我司七月的学习榜样之一
另，我在解读时，除了尽可能像解读iDP3那样，比如特意在分析代码文件之前，贴一下对应的代码结构截图——避免只是堆砌代码，我还会尽可能把模块之间、模块内部的函数之间彼此的联系及互相调用的关系都阐述出来

如此，不但从宏观上做到一目了然(注意，本文按照上图π0的代码结构，先解读src模块下的model-对应下文第一部分、policy-对应下文第二部分、training-对应下文第三部分，第四部分则解读图中src上面的packages/openpi-client，以及scripts)，更从微观上做到抽丝剥茧，看到彼此的联系与调用关系
我身边的很多朋友目前都在做π0的微调及二次开发，相信本文无论对我身边的朋友，还是对更多人的学习与工作，都会起到比较大的提升
PS，有兴趣或也在对π0做微调的，欢迎私我一两句自我简介(比如在哪个公司做什么，或在哪个高校研几什么专业)，邀请进：『七月具身：π0复现微调交流群』

第一部分 π0模型架构的实现：src下models的全面分析与解读

接下来，我们来看核心src下的各个模块，首先是其中的src/openpi/models

1.1 models/model.py：核心基础模型的定义

这是模型框架的核心文件，定义了基础的抽象类和数据结构：

`BaseModelConfig`: 所有模型配置的抽象基类
`BaseModel`: 所有模型实现的抽象基类
`Observation`: 保存模型输入的数据类
`Actions`: 定义动作数据格式
提供了通用功能如`preprocess_observation`和`restore_params`

1.1.1 基础组件和关键常量

首先是模型类型枚举，定义了两种支持的模型类型：

`PI0`：标准PI0模型
`PI0_FAST`：自回归版PI0模型

class ModelType(enum.Enum):
    """Supported model types."""

    PI0 = "pi0"
    PI0_FAST = "pi0_fast"

接下来是图像输入配置，定义了模型期望的图像输入的键名。这表明模型设计为同时接收三个视角的图像：

一个基础视图（机器人环境的全局视图）
左手腕视图（来自左手腕摄像头）
右手腕视图（来自右手腕摄像头）

# The model always expects these images
IMAGE_KEYS = (
    "base_0_rgb",
    "left_wrist_0_rgb",
    "right_wrist_0_rgb",
)

再其次，是图像分辨率设置——定义了模型处理图像的标准分辨率为224×224像素

# This may need change if we release a small model.
IMAGE_RESOLUTION = (224, 224)

1.1.2 `Observation` 类与Actions类型的详解

`Observation` 类是 OpenPI 框架中的一个核心数据结构，用于存储和管理模型的输入数据

首先，它包含了机器人感知系统收集的所有必要信息：

图像数据 (`images`)

class Observation(Generic[ArrayT]):
    """Holds observations, i.e., inputs to the model.

    See `Observation.from_dict` to see the expected dictionary form. This is the format
    that should be produced by the data transforms.
    """

    # Images, in [-1, 1] float32.
    images: dict[str, at.Float[ArrayT, "*b h w c"]]

类型：`dict[str, at.Float[ArrayT, "*b h w c"]]
用途：存储多个摄像头视角的图像数据
格式：浮点数数组，范围在 [-1, 1] 之间
维度：`*b` 表示任意批量维度，`h` 和 `w` 是图像高度和宽度，`c` 是颜色通道数

图像掩码 (`image_masks`)
```
    # Image masks, with same keys as images.
    image_masks: dict[str, at.Bool[ArrayT, "*b"]]
```
类型：`dict[str, at.Bool[ArrayT, "*b"]]`
用途：标记对应的图像是否有效
格式：布尔值数组
维度：与图像批量维度相同
机器人状态 (`state`)
```
    # Low-dimensional robot state.
    state: at.Float[ArrayT, "*b s"]
```
类型：`at.Float[ArrayT, "*b s"]`
用途：存储低维度的机器人状态向量
维度：`*b` 表示批量维度，`s` 表示状态向量维度

语言提示相关字段
`tokenized_prompt`：已经tokenized的语言提示

    # Tokenized prompt.
    tokenized_prompt: at.Int[ArrayT, "*b l"] | None = None

`tokenized_prompt_mask`：语言提示的掩码

    # Tokenized prompt mask.
    tokenized_prompt_mask: at.Bool[ArrayT, "*b l"] | None = None

当然了，两者都是可选字段（可以为 `None`）

PI0-FAST 模型特有字段
`token_ar_mask`：自回归模型的标记掩码

    # Token auto-regressive mask (for FAST autoregressive model).
    token_ar_mask: at.Int[ArrayT, "*b l"] | None = None

`token_loss_mask`：损失计算的标记掩码

    # Token loss mask (for FAST autoregressive model).
    token_loss_mask: at.Bool[ArrayT, "*b l"] | None = None

接下来，定义了`from_dict` 方法，用于从非结构化的字典数据创建 `Observation` 对象：

数据验证：确保 `tokenized_prompt` 和 `tokenized_prompt_mask` 要么同时存在，要么同时不存在

    def from_dict(cls, data: at.PyTree[ArrayT]) -> "Observation[ArrayT]":
        """This method defines the mapping between unstructured data (i.e., nested dict) to the structured Observation format."""
        # Ensure that tokenized_prompt and tokenized_prompt_mask are provided together.
        if ("tokenized_prompt" in data) != ("tokenized_prompt_mask" in data):
            raise ValueError("tokenized_prompt and tokenized_prompt_mask must be provided together.")

图像格式转换：如果输入图像是 `uint8` 格式（0-255 范围），自动转换为 `float32` 格式（范围 [-1, 1]）

        # If images are uint8, convert them to [-1, 1] float32.
        for key in data["image"]:
            if data["image"][key].dtype == np.uint8:
                data["image"][key] = data["image"][key].astype(np.float32) / 255.0 * 2.0 - 1.0

结构化数据创建：从字典数据创建结构化的 `Observation` 对象

        return cls(
            images=data["image"],
            image_masks=data["image_mask"],
            state=data["state"],
            tokenized_prompt=data.get("tokenized_prompt"),
            tokenized_prompt_mask=data.get("tokenized_prompt_mask"),
            token_ar_mask=data.get("token_ar_mask"),
            token_loss_mask=data.get("token_loss_mask"),
        )

再接下来，又定义了`to_dict` 方法，将 `Observation` 对象转换回非结构化的字典格式：

使用 `dataclasses.asdict()` 将数据类转换为字典

    def to_dict(self) -> at.PyTree[ArrayT]:
        """Convert the Observation to a nested dict."""
        result = dataclasses.asdict(self)

重命名字段以符合原始数据格式约定（`images` → `image`，`image_masks` → `image_mask`）

        result["image"] = result.pop("images")
        result["image_mask"] = result.pop("image_masks")
        return result

最后，在类外定义了 `Actions` 类型，用于表示模型的输出动作：

# Defines the format of the actions. This field is included as "actions" inside the dictionary
# produced by the data transforms.
Actions = at.Float[ArrayT, "*b ah ad"]

类型：`at.Float[ArrayT, "*b ah ad"]`
维度：`*b` 表示批量维度，`ah` 表示动作时间步长，`ad` 表示每个动作的维度

一朋友在我组建的『七月具身：π0复现微调交流群』问了个比较细节的问题，即
我想采集自己的数据来微调这个openpi，然后在采自己的数据时，我的action到底应该采什么（如果采当前帧末端位姿的话，和state有什么区别，只是差个fk而已，不是冗余了么）

真正送到模型训练的时候，action又是什么，有大佬可以解决一下吗

根据OpenPI的代码结构，state和action在robotics任务中具有不同的含义：

State (状态)，代表机器人当前的状态信息，包括：

机器人当前的配置，比如关节角度、末端执行器位置等

末端执行器(end-effector)的位置和方向

可能还包括物体的状态、环境信息等

如果只采集末端位姿，确实与状态信息存在冗余，只是差一步FK(正向运动学)计算。实际上，有效的Action (动作)代表机器人应该执行的下一步控制命令(告诉机器人如何移动)——通常是从当前状态到下一个目标状态的转换，可能是：

关节控制
表示目标关节角度，或关节角度的增量变化(delta)
说白了，state描述：我在哪里，action描述：我要去哪里

相对位移/速度
末端位置到目标位置(target position)的增量变化，和方向

控制信号
直接发送给执行器的命令，或力矩

1.1.3 preprocess_observation

1.1.4 BaseModelConfig(abc.ABC)

1.1.5 class BaseModel(nnx.Module, abc.ABC)

1.1.6 restore_params

// 待更

1.2 models/pi0.py的实现

Pi0是一个多模态扩散模型：继承自`BaseModel`，使用SigLIP处理视觉输入、使用Gemma处理语言输入，实现了基于扩散的动作生成系统，且包含`***pute_loss`和`sample_actions`方法的实现

总之，Pi0结合了多模态输入(图像和文本)来生成机器人动作序列。下面是对代码的详细解析：

1.2.1 make_attn_mask：注意力掩码生成函数

这个函数生成transformer中使用的注意力掩码，控制 token 之间的注意力流动方式

def make_attn_mask(input_mask, mask_ar):
    """
    从big_vision项目改编的注意力掩码生成函数
    
    Token可以关注那些累积mask_ar小于等于自己的有效输入token。
    这样`mask_ar` bool[?B, N]可用于设置几种类型的注意力，例如：
    
      [[1 1 1 1 1 1]]: 纯因果注意力。
    
      [[0 0 0 1 1 1]]: 前缀语言模型注意力。前3个token之间可以互相关注，
                      后3个token有因果注意力。第一个条目也可以是1，不改变行为。
    
      [[1 0 1 0 1 0 0 1 0 0]]: 4个块之间的因果注意力。一个块的token可以
                              关注所有之前的块和同一块内的所有token。
    
    参数:
      input_mask: bool[B, N] 如果是输入的一部分则为true，如果是填充则为false
      mask_ar: bool[?B, N] 如果前面的token不能依赖于它则为true，
               如果它共享与前一个token相同的注意力掩码则为false
    """

    # 将mask_ar广播到与input_mask相同的形状
    mask_ar = jnp.broadcast_to(mask_ar, input_mask.shape)  

    # 计算mask_ar在序列维度上的累积和
    cumsum = jnp.cumsum(mask_ar, axis=1)  

    # 创建注意力掩码：当目标位置的累积值<=查询位置的累积值时，允许注意力流动
    attn_mask = cumsum[:, None, :] <= cumsum[:, :, None]  

    # 创建有效掩码：只有有效的输入位置之间才能有注意力
    valid_mask = input_mask[:, None, :] * input_mask[:, :, None]  

    # 结合注意力掩码和有效掩码
    return jnp.logical_and(attn_mask, valid_mask)

它支持多种注意力模式：

纯因果注意力（每个 token 只能关注自己和之前的 token）
前缀语言模型注意力（允许前缀内部自由注意，后缀部分使用因果注意力）
块状因果注意力（在块内自由注意，块之间是因果的）

1.2.2 posemb_sincos：位置编码函数

使用正弦余弦函数实现位置编码

def posemb_sincos(
    pos: at.Real[at.Array, Any], embedding_dim: int, min_period: float, max_period: float
) -> at.Float[at.Array, f"b {embedding_dim}"]:
    """计算标量位置的正弦余弦位置嵌入向量"""
    if embedding_dim % 2 != 0:      # 检查嵌入维度是否为偶数
        raise ValueError(f"embedding_dim ({embedding_dim}) must be divisible by 2")

    fraction = jnp.linspace(0.0, 1.0, embedding_dim // 2)  # 创建均匀分布的分数值
    period = min_period * (max_period / min_period) ** fraction  # 计算周期值，对数空间中均匀分布
    sinusoid_input = jnp.einsum(
        "i,j->ij",
        pos,
        1.0 / period * 2 * jnp.pi,                      # 计算角频率
        precision=jax.lax.Precision.HIGHEST,            # 使用最高精度进行计算
    )

    # 连接sin和cos值，形成完整的位置编码
    return jnp.concatenate([jnp.sin(sinusoid_input), jnp.cos(sinusoid_input)], axis=-1)

1.2.3 class Pi0Config：含inputs_spec、get_freeze_filter

Pi0Config这个类中，定义了

动作专家底层结构gemma_300m

inputs_spec：π0模型本身接收的输入数据格式

get_freeze_filter(决定对VLM和action expect的哪部分微调，还是都微调)

1.2.3.1 模型配置参数的定义

首先，这个类定义了模型的配置参数，比如PaLI-Gemma 变体：`gemma_2b，尤其值得注意的是在本π0的官方实现中，动作专家的底层结构用的300M大小的gemma模型变体

class Pi0Config(_model.BaseModelConfig):
    dtype: str = "bfloat16"  # 设置数据类型为bfloat16
    paligemma_variant: _gemma.Variant = "gemma_2b"          # 设置PaLI-Gemma变体为2B参数版本
    action_expert_variant: _gemma.Variant = "gemma_300m"    # 设置动作专家为gemma的300M变体版本

    # 设置模型特定的默认值
    action_dim: int = 32          # 设置动作维度为32
    action_horizon: int = 50      # 设置动作序列长度为50步
    max_token_len: int = 48       # 设置最大token长度为48

1.2.3.2 inputs_spec：定义了π0模型本身接收的输入数据格式

其次，通过inputs_spec函数定义了π0模型本身接收的输入数据格式，函数采用关键字参数 `batch_size`（默认为1），返回一个包含观察规格和动作规格的元组

def inputs_spec(self, *, batch_size: int = 1) -> Tuple[Type[_model.Observation], Type[_model.Actions]]

其支持多种输入，比如
视觉输入(三个不同视角的RGB图像)、语言输入(分词后的文本prompt)、状态输入(当前机器人状态)
输出上
则是一个时序动作序列(包含50个连续的动作向量，每个动作向量有32个维度，可能对应关节角度或其他控制信号)

具体而言该函数进行如下4个操作
一、创建图像规格

        image_spec = jax.ShapeDtypeStruct([batch_size, *_model.IMAGE_RESOLUTION, 3], jnp.float32)

其中的

`[batch_size, *_model.IMAGE_RESOLUTION, 3]` 定义了图像张量的形状：比如
批次大小
图像分辨率（从 `_model.IMAGE_RESOLUTION` 获取，可能是如 [224, 224] 这样的值）
3 个颜色通道 (RGB)
`jnp.float32` 指定了数据类型为 32 位浮点数

二、创建图像掩码规格

        image_mask_spec = jax.ShapeDtypeStruct([batch_size], jnp.bool_)

其定义了图像掩码规格，每个批次中的每个图像都有一个布尔值，这个掩码用于指示哪些图像是有效的（`True`）或无效的（`False`）

三、创建观察规格：包含视觉输入、机器人状态、指令输入
`at.disable_typechecking()` 临时禁用类型检查，可能是因为这里创建的是类型规格而不是实际的数据，且观察规格包含多个组件：

多视角图像
base_0_rgb: 机器人底座/身体视角的RGB图像
left_wrist_0_rgb: 左手腕视角的RGB图像
right_wrist_0_rgb: 右手腕视角的RGB图像

        with at.disable_typechecking():
            observation_spec = _model.Observation(
                images={
                    "base_0_rgb": image_spec,
                    "left_wrist_0_rgb": image_spec,
                    "right_wrist_0_rgb": image_spec,
                },

图像掩码
对应每个视角图像的有效性掩码
机器人状态：
形状为 `[batch_size, self.action_dim]` 的浮点数张量，其中的`self.action_dim` 默认为32，表示状态向量的维度
```
                state=jax.ShapeDtypeStruct([batch_size, self.action_dim], jnp.float32),
```
分词后的文本prompt
形状为 `[batch_size, self.max_token_len]` 的整数张量
`self.max_token_len` 默认为48，表示最大token数量
数据类型为 `jnp.int32`，表示token ID

提示掩码
与分词提示相同形状的布尔张量，用于指示哪些位置有有效的token

                state=jax.ShapeDtypeStruct([batch_size, self.action_dim], jnp.float32),
                tokenized_prompt=jax.ShapeDtypeStruct([batch_size, self.max_token_len], jnp.int32),
                tokenized_prompt_mask=jax.ShapeDtypeStruct([batch_size, self.max_token_len], bool),
            )

四、创建动作规格

        action_spec = jax.ShapeDtypeStruct([batch_size, self.action_horizon, self.action_dim], jnp.float32)

其定义了动作数据的形状和类型：

`batch_size`: 批次大小
`self.action_horizon`: 动作序列长度，默认为50
`self.action_dim`: 每个动作的维度，默认为32
`jnp.float32` 指定了数据类型为32位浮点数

然后返回

        return observation_spec, action_spec

1.2.3.3 get_freeze_filter：参数冻结器，包含谁则相当于谁被冻结/过滤

此外，该配置类还实现了get_freeze_filter这个函数，作用是如果选择LoRA微调(冻结原始预训练模型的参数，只更新新添加的低秩适应层参数)，则需要对模型中的某些参数做冻结

三种可能的情况：

只对 PaLI-Gemma 使用 LoRA
意味着只冻结 Gemma 原始参数，然后排除动作专家原始参数，微调Gemma原始参数之外的少量LoRA部分

注意
首先，不微调π0，也有指令跟随能力
其次，对π0的语言模型部分使用“指令跟随数据集”做lora微调，不是说让其丧失指令跟随能力，而是「lora微调对指令跟随能力的加强」作用相对没很大
「关于什么是LoRA，详见此文《LLM高效参数微调方法：从Prefix Tuning、Prompt Tuning、P-Tuning V1/V2到LoRA、QLoRA(含对模型量化的解释)》的第4部分」
毕竟lora微调的本质是原始参数冻结，而是微调「两个可以近似原矩阵的两个小矩阵」参数
且为免歧义，再强调一句：lora微调也是有效的，有时甚至可以逼近全参微调
只对动作专家使用 LoRA
意味着只冻结动作专家参数，微调动作专家原始参数之外的少量LoRA部分
对两者都使用 LoRA
意味着冻结两者的基础参数，微调两者原始参数之外的少量LoRA部分

如此，可以选择性地微调模型的特定部分(语言部分或动作预测部分）

具体而言，该get_freeze_filter分为4大阶段

第一阶段，定义函数本身、初始化变量，并创建参数过滤器

首先，定义函数

    def get_freeze_filter(self) -> nnx.filterlib.Filter:
        """返回基于模型配置的冻结过滤器"""

其次，初始化变量

        filters = []      # 初始化过滤器列表
        has_lora = False  # 初始化LoRA标志

接着，创建参数过滤器

        # 匹配所有LLM参数的正则表达式，用于选择 Gemma 语言模型的参数
        gemma_params_filter = nnx_utils.PathRegex(".*llm.*")  

        # 匹配动作专家参数的正则表达式
        action_expert_params_filter = nnx_utils.PathRegex(".*llm.*_1.*")

第二阶段，分情况添加LoRA权重

即要么只对语言模型使用LoRA(意味着不对动作专家使用LoRA)，要么只对动作专家使用LoRA

即，接下来是对PaLI-Gemma变体的处理
如果只对PaLI-Gemma使用LoRA，则
一方面，所有Gemma的原始参数将被冻结/过滤掉，因为LoRA就是只微调原始参数之外的少量LoRA部分

        # 如果只针对PaLI-Gemma使用LoRA
        if "lora" in self.paligemma_variant:
            # 过滤器列表添加Gemma的原始参数
            filters.append(
                gemma_params_filter,
            )

且二方面，代表动作专家的原始参数不被冻结/过滤，故过滤器列表不添加动作专家expert原始参数，意味着动作专家可能被全参微调而非被LoRA微调

            if "lora" not in self.action_expert_variant:
                # 因为只冻结Gemma参数，故过滤器列表不添加动作专家expert的原始参数
                filters.append(
                    nnx.Not(action_expert_params_filter),
                )
            has_lora = True

再下来是对动作专家变体的处理，如果对action_expert_variant使用LoRA，则过滤器列表添加动作专家expert的原始参数，而微调动作专家原始参数之外的少量LoRA部分

        elif "lora" in self.action_expert_variant:
            # 如果动作专家使用LoRA，则过滤器列表添加动作专家expert的原始参数
            filters.append(
                action_expert_params_filter,
            )
            has_lora = True

第三阶段，针对需要LoRA微调的少量参数处理，以及如果没有需要LoRA微调时的处理

如果有需要被LoRA微调的部分，则过滤器列表里不添加原始参数之外的LoRA相关参数(代表着不被过滤)

        if has_lora:
            # If any lora is used, exclude all lora params.
            filters.append(
                nnx.Not(nnx_utils.PathRegex(".*lora.*")),
            )

如果没有被冻结/过滤的参数，则什么都不需要处理——即默认微调所有参数
```
        if not filters:
            return nnx.Nothing
```

第四阶段，返回所有需要被冻结/被过滤的参数，这毕竟是get_freeze_filter函数本身定义所追求的目标

        return nnx.All(*filters)

值得注意的是，也是我之前看到这里思考过的一个问题，即在训练 π0 的动作预测能力时

默认会同时调整 VLM 和动作专家的参数

如果需要只调整动作专家的参数，可以通过修改 `get_freeze_filter` 方法来冻结 VLM 的参数

1.2.4 class Pi0：含特征嵌入(embed_prefix/embed_suffix)、损失函数(训练去噪的准确性)、推理(去噪生成动作)

核心模型类，继承自 `_model.BaseModel`，实现了：

多模态输入处理
处理多视角图像（基础视角、左手腕视角、右手腕视角）
处理文本提示（如指令）
处理机器人当前状态
扩散过程
训练时：将干净动作添加噪声，让模型学习去噪
推理时：从纯噪声开始，逐步降噪生成动作序列
注意力机制
使用精心设计的注意力掩码控制信息流动
前缀（图像和文本）内部使用全注意力
后缀（状态和动作）使用特殊的注意力模式

1.2.4.1 初始化方法 `init`

class Pi0(_model.BaseModel):
    def __init__(self, config: Pi0Config, rngs: nnx.Rngs):
        # 初始化基类
        super().__init__(config.action_dim, config.action_horizon, config.max_token_len)
        
        # 获取PaLI-Gemma和动作专家配置
        paligemma_config = _gemma.get_config(config.paligemma_variant)
        action_expert_config = _gemma.get_config(config.action_expert_variant)

其组合了多个核心组件：

一个是PaLI-Gemma 模型：结合了 Gemma 语言模型和 SigLIP 视觉模型

先是对语言模型的初始化

        # 创建并初始化语言模型
        # TODO: 用NNX重写Gemma，目前使用桥接
        llm = nnx_bridge.ToNNX(
            _gemma.Module(
                configs=[paligemma_config, action_expert_config],  # 配置两个Gemma模型
                embed_dtype=config.dtype,          # 设置嵌入数据类型
            )
        )
        llm.lazy_init(rngs=rngs, method="init")    # 延迟初始化LLM

然后是对视觉模型的初始化

        # 创建并初始化图像模型
        img = nnx_bridge.ToNNX(
            _siglip.Module(
                num_classes=paligemma_config.width,  # 设置图像特征维度与语言模型宽度相匹配
                variant="So400m/14",  # 使用400M参数SigLIP模型
                pool_type="none",  # 不使用池化，保留所有图像token
                scan=True,  # 启用扫描优化
                dtype_mm=config.dtype,  # 设置矩阵乘法数据类型
            )
        )

        # 使用假观察中的图像初始化图像模型
        img.lazy_init(next(iter(config.fake_obs().images.values())), train=False, rngs=rngs)

最后，把语言模型和视觉模型组合成PaLI-Gemma多模态模型

        
        # 组合LLM和图像模型为PaLI-Gemma多模态模型
        self.PaliGemma = nnx.Dict(llm=llm, img=img)

另一个是线性投影层：用于

状态投影

        # 状态投影层：将机器人状态投影到模型维度
        self.state_proj = nnx.Linear(config.action_dim, action_expert_config.width, rngs=rngs)

动作投影

        # 动作输入投影层：将动作投影到模型维度
        self.action_in_proj = nnx.Linear(config.action_dim, action_expert_config.width, rngs=rngs)

时间-动作混合等

        # 动作-时间MLP输入层：将连接的动作和时间特征投影到模型维度
        self.action_time_mlp_in = nnx.Linear(2 * action_expert_config.width, action_expert_config.width, rngs=rngs)

        # 动作-时间MLP输出层
        self.action_time_mlp_out = nnx.Linear(action_expert_config.width, action_expert_config.width, rngs=rngs)

        # 动作输出投影层：将模型输出投影回动作维度
        self.action_out_proj = nnx.Linear(action_expert_config.width, config.action_dim, rngs=rngs)

1.2.4.2 特征嵌入方法：embed_prefix(图像和文本输入)、embed_suffix(状态和动作信息)

`embed_prefix`：处理图像和文本输入(图像通过SigLip模型编码，文本通过Gemma LLM编码)，创建前缀 token，皆为双向注意力，用ar_mask = false表示
`embed_suffix`：处理机器人状态信息、噪声化的动作信息(状态和噪声动作经过线性投影和MLP处理)，创建后缀 token
其中
状态为单个token，和第一个动作token均设置为单向注意力，用ar_mask = true表示
其余动作tokens之间设置为双向注意力，用ar_mask = false表示

首先，对于前者embed_prefix

    def embed_prefix(
        self, obs: _model.Observation
    ) -> Tuple[at.Float[at.Array, Any], at.Bool[at.Array, Any], at.Bool[at.Array, Any]]:
        """嵌入前缀部分（图像和文本）"""
        input_mask = []           # 初始化输入掩码列表
        ar_mask = []              # 初始化自回归掩码列表
        tokens = []               # 初始化token列表

其工作流程为

图像处理：说白了，就是把图像token化
使用SigLip视觉模型处理每个图像，生成图像tokens序列

        # 嵌入图像
        for name in obs.images:
            # 通过图像模型获取图像token
            image_tokens, _ = self.PaliGemma.img(obs.images[name], train=False)

            tokens.append(image_tokens)      # 添加图像token

图像掩码扩展
将图像掩码扩展到与图像tokens相同的序列长度，使用einops.repeat进行形状变换，这些掩码会指示哪些图像是有效的，而哪些是填充的

            # 重复图像掩码以匹配token维度
            input_mask.append(
                einops.repeat(
                    obs.image_masks[name],
                    "b -> b s",               # 调整形状：批次维度保持不变，添加序列维度
                    s=image_tokens.shape[1],  # 序列长度等于图像token数
                )
            )

自回归掩码设置
设置图像tokens之间的注意力为双向(False表示双向注意力)，原因在于图像内容通常是非时序性的数据
```
            # 图像token之间互相关注（非自回归）
            ar_mask += [False] * image_tokens.shape[1]
```

文本处理
使用LLM模型对文本输入tokenized_inputs进行嵌入

        # 添加语言（即分词后的输入）
        if obs.tokenized_prompt is not None:
            # 通过语言模型嵌入分词后的提示
            tokenized_inputs = self.PaliGemma.llm(obs.tokenized_prompt, method="embed")
            tokens.append(tokenized_inputs)                  # 添加文本token
            input_mask.append(obs.tokenized_prompt_mask)     # 添加提示掩码

且同样设置为双向注意力，相当于语言token可以关注图像token，图像token反过来亦可关注语言token，最终实现多模态融合

            # 图像和语言输入之间完全关注（非自回归）
            ar_mask += [False] * tokenized_inputs.shape[1]

最后，连接所有token和掩码，其中包含了
多模态信息的融合表示tokens——图像token和语言token
以及指示哪些token是有效信息的input_mask
和如何在这些token之间进行注意力计算规则的ar_mask——相当于控制信息流动的方向

        # 连接所有token和掩码
        tokens = jnp.concatenate(tokens, axis=1)    # 在序列维度上连接token
        input_mask = jnp.concatenate(input_mask, axis=1)  # 在序列维度上连接输入掩码
        ar_mask = jnp.array(ar_mask)                # 转换自回归掩码为数组
        
        return tokens, input_mask, ar_mask          # 返回token、输入掩码和自回归掩码

顺便，再回顾下此图

其次，对于后者embed_suffix

定义如下，其参数包括obs(一般包含图像和机器人状态)、noisy_actions、timestep

    def embed_suffix(
        self, obs: _model.Observation, noisy_actions: _model.Actions, timestep: at.Float[at.Array, Any]
    ) -> Tuple[at.Float[at.Array, Any], at.Bool[at.Array, Any], at.Bool[at.Array, Any]]:
        """嵌入后缀部分（状态和动作）"""
        input_mask = []           # 初始化输入掩码列表
        ar_mask = []              # 初始化自回归掩码列表
        tokens = []               # 初始化token列表

其工作流程为

状态处理
将状态信息投影到embedding空间

        # 添加单个状态token
        state_token = self.state_proj(obs.state)[:, None, :]  # 投影状态并添加序列维度
        tokens.append(state_token)                            # 添加状态token

        # 添加状态掩码（全为1），表示这个状态token是有效的
        input_mask.append(jnp.ones((obs.state.shape[0], 1), dtype=jnp.bool_))

并设置为单向注意力(True)，表明图像和语言输入不能关注状态信息，因为image/language do not attend to state or actions

        # 图像/语言输入不关注状态或动作（自回归）
        ar_mask += [True]

时间步嵌入，使用正弦-余弦位置编码生成时间步嵌入

       # 使用正弦余弦位置编码嵌入时间步，敏感度范围为[0, 1]
        time_emb = posemb_sincos(timestep, self.action_in_proj.out_features, min_period=4e-3, max_period=4.0)

动作和时间信息融合，比如通过action_time_tokens连接：「带噪声的动作」和「时间token」

        # 混合时间步 + 动作信息，使用MLP
        action_tokens = self.action_in_proj(noisy_actions)  # 投影带噪声的动作

        # 重复时间嵌入以匹配动作序列长度
        time_tokens = einops.repeat(time_emb, "b emb -> b s emb", s=self.action_horizon)

        # 连接动作和时间token
        action_time_tokens = jnp.concatenate([action_tokens, time_tokens], axis=-1)

MLP处理
使用两层MLP和swish激活函数对「动作和时间的组合表示」进行非线性变换，以进一步融合：(噪声)动作和时间信息

        # 通过MLP处理
        action_time_tokens = self.action_time_mlp_in(action_time_tokens)   # 输入层
        action_time_tokens = nnx.swish(action_time_tokens)                 # Swish激活函数
        action_time_tokens = self.action_time_mlp_out(action_time_tokens)  # 输出层

注意力掩码设置
第一个动作token设置为单向注意力「上面说过了的，单向注意力，用ar_mask = true表示」，其余动作tokens之间设置为双向注意力

        # 添加动作时间token
        tokens.append(action_time_tokens)

        # 添加掩码（全为1），表示所有动作token都是有效的
        input_mask.append(jnp.ones(action_time_tokens.shape[:2], dtype=jnp.bool_))  

        # 图像/语言/状态输入不关注动作token（动作第一个是自回归的——单向，其余不是——双向）
        ar_mask += [True] + ([False] * (self.action_horizon - 1))

最后连接所有token和掩码

        # 连接所有token和掩码
        tokens = jnp.concatenate(tokens, axis=1)          # 在序列维度上连接token
        input_mask = jnp.concatenate(input_mask, axis=1)  # 在序列维度上连接输入掩码
        ar_mask = jnp.array(ar_mask)        # 转换自回归掩码为数组
        
        return tokens, input_mask, ar_mask  # 返回token、输入掩码和自回归掩码

1.2.4.3 损失函数***pute_loss：训练模型去噪的准确率

总的来讲

训练的时候，对其中的「原始动作action」数据加噪，最后去预测所添加的真实噪声，预测噪声的结果为，然后计算预测噪声与真实噪声之间的均方误差
也就是说，训练时的本质其实是为了让模型具备生成真正想要动作的能力，以确保在推理时，能得到真正想要动作的能力

那可能有同学疑问了，既然通过对原始动作加噪，然后预测噪声，最后噪声动作减掉预测噪声 便是所预测的原始动作，那为何不对比实际的原始动作，与所预测的原始动作是否一致呢
其实我之前在此文《图像生成发展起源：从VAE、扩散模型DDPM、DDIM到DETR、ViT、Swin transformer》中的「2.1.1 从扩散模型概念的提出到DDPM(含U-***网络的简介)、DDIM」已经讲了，原因在于
1 对噪声的预测，比对动作的预测更容易，一者预测噪声收敛更稳定，二者噪声通常是标准化的，比如高斯噪声的均值为0 方差为1，使得模型预测噪声时不需要适应不同尺度的输出
2 -prediction 和 -prediction其实理论上也是等价的，毕竟 + =
如此，便可以在推理的时候，针对一个随机生成的纯噪声，基于observation(包含图像和机器人状态)，逐步去噪生成机器人的动作序列

具体而言，***pute_loss实现了扩散模型的训练损失计算

对输入观察进行预处理，其中
preprocess_rng用于观察预处理(比如图像增强等)
noise_rng用于生成噪声
time_rng用于从beta分布采样时间步

    def ***pute_loss(
        self, rng: at.KeyArrayLike, observation: _model.Observation, actions: _model.Actions, *, train: bool = False
    ) -> at.Float[at.Array, Any]:
        """计算扩散模型的损失函数"""
        # 分割随机数生成器为三部分，用于不同的随机操作
        preprocess_rng, noise_rng, time_rng = jax.random.split(rng, 3)

生成随机噪声并采样时间点 t

        # 获取动作的批次形状
        batch_shape = actions.shape[:-2]

        # 生成与动作相同形状的高斯噪声
        noise = jax.random.normal(noise_rng, actions.shape)

        # 从Beta分布采样时间点，范围为[0.001, 1]，Beta(1.5, 1)偏向较低的值
        time = jax.random.beta(time_rng, 1.5, 1, batch_shape) * 0.999 + 0.001

        # 扩展时间维度以匹配动作形状
        time_expanded = time[..., None, None]

创建带噪动作序列 x_t，相当于x_t是噪声化的动作，随着时间从0到1，原始动作逐渐添加真实噪声，变为纯噪声
而代表所加的真实噪声，便是咱们所要预测噪声的ground truth
故所添加的噪声 即 = 加满噪声的动作 - 原始动作
```
        # 创建带噪声的动作：t * noise + (1-t) * actions
        x_t = time_expanded * noise + (1 - time_expanded) * actions

        # 计算真实噪声减去动作的差异，这是模型需要预测的目标
        u_t = noise - actions
```

嵌入前缀和后缀

        # 一次性前向传递前缀+后缀
        # 嵌入前缀（图像和文本）
        prefix_tokens, prefix_mask, prefix_ar_mask = self.embed_prefix(observation)

        # 嵌入后缀（状态和带噪声的动作）
        suffix_tokens, suffix_mask, suffix_ar_mask = self.embed_suffix(observation, x_t, time)

构建注意力掩码和位置编码
根据下图

可得

        # 连接掩码：通过链接前缀和后缀的掩码，从而创建完整的输入掩码
        input_mask = jnp.concatenate([prefix_mask, suffix_mask], axis=1)
        ar_mask = jnp.concatenate([prefix_ar_mask, suffix_ar_mask], axis=0)

        # 创建注意力掩码make_attn_mask，从而控制不同token之间的可见性
        attn_mask = make_attn_mask(input_mask, ar_mask)

        # 计算位置编码
        positions = jnp.cumsum(input_mask, axis=1) - 1

模型前向传播，即调用PaliGemma进行推理，处理前缀和后缀token
当然了，输出中我们只关注与后缀相关的部分，因为其中包含了我们想要的动作预测的部分
```
        # 通过PaLI-Gemma模型处理token
        _, suffix_out = self.PaliGemma.llm(
            [prefix_tokens, suffix_tokens], mask=attn_mask, positions=positions
        )
```

预测噪声

        # 将模型输出投影回动作空间
        v_t = self.action_out_proj(suffix_out[:, -self.action_horizon :])

计算预测噪声与实际噪声间的均方误差

        # 返回预测噪声和真实噪声之间的均方误差
        return jnp.mean(jnp.square(v_t - u_t), axis=-1)

注解 LeRobotDataset：训练数据集的来源(即训练数据集长什么样)

不知道有没有同学会疑问这段代码里面的数据集是从哪来的，比如原始动作action 从哪来的，我暂且不管有没有疑惑，假设有人有此疑惑，故我来解释下数据集的来源途径

π0主要使用两种数据集：

FakeDataset - 生成随机数据用于测试
LeRobotDataset - 真实的机器人操作数据

LeRobotDataset 是一个专为机器人学习设计的数据集格式，来自`lerobot.***mon.datasets.lerobot_dataset`模块。这个数据集包含了训练π0模型所需的观察数据和动作数据，其包含

Aloha数据集，侧重双臂协同的精确操作，适合特定任务的模仿学习，比如这个是打开笔帽的任务

Libero数据集，注重多样化任务和泛化能力，适合语言引导的通用机器人控制

LeRobotDataset 数据通常包含以下几个关键部分：
观察数据 (Observation)
图像数据：来自不同摄像头的图像
"observation.images.cam_high"
"observation.images.cam_low"
"observation.images.cam_left_wrist"
"observation.images.cam_right_wrist"
状态数据：机器人的关节角度等状态信息
"observation.state"
动作数据 (Actions)
动作序列：每个时间步的机器人动作指令
"action"
时间戳信息：通过`delta_timestamps`定义的时间间隔
任务信息
任务描述：可用于生成提示(prompt)
元数据：包括帧率(fps)等信息
数据集示例
ALOHA数据集
physical-intelligence/aloha_pen_uncap_diverse
{
    "observation": {
        "images": {
            "cam_high": np.ndarray(shape=(3, 224, 224), dtype=np.uint8),
            "cam_left_wrist": np.ndarray(shape=(3, 224, 224), dtype=np.uint8),
            "cam_right_wrist": np.ndarray(shape=(3, 224, 224), dtype=np.uint8)
        },
        "state": np.ndarray(shape=(14,), dtype=np.float32)
    },
    "action": np.ndarray(shape=(14,), dtype=np.float32),
    "prompt": "uncap the pen"
}
其中，14维机器人状态向量的含义
[
    # 左臂关节角度 (6维)
    left_shoulder_pitch,
    left_shoulder_roll,
    left_shoulder_yaw,
    left_elbow_pitch,
    left_elbow_roll,
    left_wrist_pitch,

    # 左手爪状态 (1维)
    left_gripper,

    # 右臂关节角度 (6维)
    right_shoulder_pitch,
    right_shoulder_roll,
    right_shoulder_yaw,
    right_elbow_pitch,
    right_elbow_roll,
    right_wrist_pitch,

    # 右手爪状态 (1维)
    right_gripper
]
一个LeRobotDataset的样本可能看起来像这样
比如Libero数据集：physical-intelligence/libero
{
    "observation": {
        "images": {
            # 高视角RGB图像，224x224x3
            "cam_high": np.ndarray(shape=(224, 224, 3), dtype=np.uint8),
            # 低视角RGB图像
            "cam_low": np.ndarray(shape=(224, 224, 3), dtype=np.uint8),
            # 左手腕视角RGB图像
            "cam_left_wrist": np.ndarray(shape=(224, 224, 3), dtype=np.uint8),
            # 右手腕视角RGB图像
            "cam_right_wrist": np.ndarray(shape=(224, 224, 3), dtype=np.uint8)
        },

        # 机器人状态向量，包含关节角度等信息
        "state": np.ndarray(shape=(14,), dtype=np.float32),  
    },

    # 动作序列，50个时间步，每步14维动作向量
    "actions": np.ndarray(shape=(50, 14), dtype=np.float32),

    # 任务描述文本
    "prompt": "fold the towel"
}
再比如
{
    "observation": {
        "images": {
            "cam_high": <224x224x3 RGB image of robot workspace from above>,
            "cam_left_wrist": <224x224x3 RGB image from left gripper>,
            "cam_right_wrist": <224x224x3 RGB image from right gripper>
        },
        "state": [0.1, -0.5, 0.3, ...],  # 14维机器人关节状态
    },
    "actions": [
        [0.1, -0.2, 0.3, ...],  # t=0时刻的动作
        [0.15, -0.25, 0.35, ...],  # t=1时刻的动作
        ...  # 共50个时间步
    ],
    "prompt": "pick up the blue cube and place it in the red bowl"
}

真实数据来自`lerobot_dataset`模块，通过以下代码加载——下文「2.2.2 create_dataset：创建适合训练的数据集」还会详解：

dataset_meta = lerobot_dataset.LeRobotDatasetMetadata(repo_id, local_files_only=data_config.local_files_only)
dataset = lerobot_dataset.LeRobotDataset(
    data_config.repo_id,
    delta_timestamps={
        key: [t / dataset_meta.fps for t in range(model_config.action_horizon)]
        for key in data_config.action_sequence_keys
    },
    local_files_only=data_config.local_files_only,
)

这里的`repo_id`指向一个特定的数据仓库，是Hugging Face上的数据集或其他存储位置。数据集通过配置文件中的参数指定，例如我们在`config.py`中看到的配置——下文「2.1 配置系统 (config.py)」还会详解：

    # Inference Aloha configs.
    #
    TrainConfig(
        name="pi0_aloha",
        model=pi0.Pi0Config(),
        data=LeRobotAlohaDataConfig(
            assets=AssetsConfig(asset_id="trossen"),
        ),
    ),

以下是对数据流程总结

从LeRobot数据集加载原始数据，包含观察(observation)和动作(action)
通过数据转换管道处理数据，包括重打包和归一化
在训练期间，向原始动作添加噪声
模型学习预测添加的噪声，而不是直接预测原始动作
在推理时，模型从纯噪声开始，通过迭代去噪过程生成动作序列

这种基于扩散的方法允许π0从噪声中逐步精炼动作，最终生成平滑且符合任务要求的机器人动作序列

1.2.4.4 推理函数 `sample_actions`：基于扩散模型逆向采样(即去噪)，生成机器人动作序列

sample_actions函数是Pi0模型的核心推理方法，实现了基于扩散模型的逆向采样过程——说白了就是去噪，它从纯噪声开始，通过多步骤逐渐"去噪"，最终生成符合条件分布的机器人动作序列

函数的核心是一个基于while循环的迭代过程，每一步都使用训练好的神经网络预测从当前噪声化动作到目标动作的方向——从噪声到目标的方向代表速度场，毕竟咱们去噪的方向得对不然就去歪了

总之，这个函数将观察数据（图像和可选的文本提示）转换为具体的动作轨迹，是模型部署时的主要接口，简言之，其包含以下流程

首先从纯噪声开始 (t=1)
通过重复迭代降噪步骤，逐步将噪声转化为有意义的动作序列
使用KV缓存优化推理速度
实现了一个迭代降噪过程
最终返回完全降噪后的动作序列 x_0

具体而言，包含如下步骤

第一，初始化

首先，函数对输入观察数据进行预处理，包括标准化图像大小等操作

def sample_actions(
    self,
    rng: at.KeyArrayLike,               # 随机数生成器
    observation: _model.Observation,    # 观察输入，包含图像和文本等
    *,
    num_steps: int = 10,                # 扩散过程的步数，默认为10步
) -> _model.Actions:                    # 返回生成的动作序列

    # 对观察数据进行预处理，不进行训练时的数据增强
    observation = _model.preprocess_observation(None, observation, train=False)

然后设置时间步长`dt`为负值（因为是从t=1向t=0方向演化），生成初始随机噪声作为起点，且时间上约定："t=1是噪声，t=0是目标分布"，这是扩散文献中常见的约定，不过与Pi0论文相反

    # 注意：这里使用扩散模型文献中更常见的约定，t=1是噪声，t=0是目标分布
    # 这与pi0论文相反
    dt = -1.0 / num_steps                       # 计算时间步长，从1到0
    batch_size = observation.state.shape[0]     # 获取批次大小

    # 生成初始噪声，形状为[批次大小, 动作序列长度, 动作维度]
    noise = jax.random.normal(rng, (batch_size, self.action_horizon, self.action_dim))

第二，Key-Value缓存初始化(预计算并存储前缀表示，减少冗余计算)

处理观察数据，得到前缀表示和相关掩码

    # 首先通过前缀的前向传递填充KV缓存
    # 获取前缀的token表示和掩码
    prefix_tokens, prefix_mask, prefix_ar_mask = self.embed_prefix(observation)

    # 创建前缀的注意力掩码
    prefix_attn_mask = make_attn_mask(prefix_mask, prefix_ar_mask)

    # 计算位置编码
    positions = jnp.cumsum(prefix_mask, axis=1) - 1

然后使用PaliGemma语言模型进行一次前向传递，生成Key-Value缓存（`kv_cache`）——这是一个性能优化：因为前缀部分在整个采样过程中保持不变，预先计算并缓存它们的表示可以避免重复计算

    # 进行前向传递，获取KV缓存
    _, kv_cache = self.PaliGemma.llm([prefix_tokens, None], mask=prefix_attn_mask, positions=positions)

第三，通过step函数构建注意力掩码系统并让PaliGemma做推理

核心迭代通过 `jax.lax.while_loop` 实现

根据源码

可知，该class Pi0(_model.BaseModel)类的最后两行是

    # 使用while循环进行迭代采样，从t=1（噪声）开始
    x_0, _ = jax.lax.while_loop(cond, step, (noise, 1.0))

    # 返回最终的去噪结果（生成的动作序列）
    return x_0

具体而言，包含 `step` 函数和 `cond` 函数，其中，`step` 函数是每次迭代的核心

首先，step函数通过 `embed_suffix` 处理当前状态，包括状态信息嵌入、噪声化动作、时间步编码

    def step(carry):
        """定义单步去噪函数"""
        x_t, time = carry  # carry数组包含当前状态和时间

        # 将时间广播到批次维度，并嵌入后缀（状态和动作）
        suffix_tokens, suffix_mask, suffix_ar_mask = self.embed_suffix(
            observation, x_t, jnp.broadcast_to(time, batch_size)
        )

其次，构建复杂的注意力掩码系统，处理前缀-后缀之间的注意力关系——这个复杂的掩码系统允许后缀token（包括状态和动作）有选择地关注前缀token（图像和文本），实现了条件生成，具体而言，其构建了三层注意力掩码：

后缀内部注意力掩码，控制后缀token（状态和动作）之间的注意力关系

        # 创建后缀内部的注意力掩码，形状为(批次, 后缀长度, 后缀长度)
        suffix_attn_mask = make_attn_mask(suffix_mask, suffix_ar_mask)

前缀-后缀注意力掩码，控制后缀token如何关注前缀token（图像和文本输入）

        # 创建后缀对前缀的注意力掩码，形状为(批次, 后缀长度, 前缀长度)
        prefix_attn_mask = einops.repeat(prefix_mask, "b p -> b s p", s=suffix_tokens.shape[1])

完整注意力掩码，将前两个掩码组合，形成完整的注意力控制机制

        # 组合掩码，形状为(批次, 后缀长度, 前缀长度+后缀长度)
        # 控制后缀token（生成查询）如何关注完整序列（生成键和值）
        full_attn_mask = jnp.concatenate([prefix_attn_mask, suffix_attn_mask], axis=-1)

当然了，过程中还做了形状检查，确保张量维度正确

        # 验证掩码形状正确
        assert full_attn_mask.shape == (
            batch_size,
            suffix_tokens.shape[1],
            prefix_tokens.shape[1] + suffix_tokens.shape[1],
        )

接着，计算位置编码，为后缀token计算其在完整序列中的位置，这对于Transformer模型理解序列顺序很重要

        # 计算后缀token的位置编码
        positions = jnp.sum(prefix_mask, axis=-1)[:, None] + jnp.cumsum(suffix_mask, axis=-1) - 1

之后，模型推理，使用PaliGemma语言模型进行推理，利用缓存的前缀信息（`kv_cache`）提高效率

        # 使用KV缓存进行高效的前向传递
        (prefix_out, suffix_out), _ = self.PaliGemma.llm(
            [None, suffix_tokens], mask=full_attn_mask, positions=positions, kv_cache=kv_cache
        )

        # 且确保前缀输出为None（因为使用了KV缓存）
        assert prefix_out is None

第四，step函数中做最后的速度预测与动作更新(去噪)

在每一步中，模型预测速度场 `v_t`（从噪声到目标的方向），并通过类欧拉法更新动作表示——使用简单而有效的欧拉方法进行轨迹采样

本质就是对去噪，而便是预测的噪声，是时间步长——如上面说过的「时间步长`dt`为负值（因为是从t=1向t=0方向演化），生成初始随机噪声作为起点，且时间上约定："t=1是噪声，t=0是目标分布"」

具体而言

一方面，提取模型输出并预测速度场`v_t`——相当于本质是通过PaliGemma模型预测去噪方向 `v_t`

        # 预测噪声
        v_t = self.action_out_proj(suffix_out[:, -self.action_horizon :])

二方面，使用欧拉法更新动作状态和时间步

        # 使用欧拉方法更新状态和时间
        return x_t + dt * v_t, time + dt

至于cond函数确定何时停止迭代，通过检查时间是否接近零(当然，要考虑浮点精读可能存在的误差)

    def cond(carry):
        """定义循环终止条件"""
        x_t, time = carry

        # 考虑浮点误差，当时间接近0时停止
        return time >= -dt / 2

1.3 语言模型实现：models/gemma.py

src/openpi/models/gemma.py实现了Gemma语言模型的核心组件，定义了RMSNorm、Embedder、Attention、FeedForward等模块，且提供了不同规模Gemma模型的配置（300M, 2B等）

// 待更

1.4 视觉模型实现：models/siglip.py

`siglip.py`: 实现了视觉编码器，基于Vision Transformer (ViT)，定义了位置编码、注意力池化等组件，支持不同大小的模型变体

// 待更

1.5 tokenizer.py: 提供文本tokenization功能

这段代码实现了两个相关但功能不同的tokenizer类：`PaligemmaTokenizer` 和 `FASTTokenizer`

1.5.1 PaligemmaTokenizer 类：专门处理文本prompt

`PaligemmaTokenizer` 是一个相对简单的Tokenizer，专门处理文本prompt

第一方面，在初始化阶段

`__init__` 方法接收一个 `max_len` 参数（默认为 48）来设定token序列的最大长度

    # 初始化方法，设置最大token长度，默认为48
    def __init__(self, max_len: int = 48):  
        # 存储最大token长度
        self._max_len = max_len

接着，它调用 `download.maybe_download` 函数从 Google Cloud Storage 获取预训练的 PaliGemma 分词模型
```
        # 下载PaliGemma分词器模型
        path = download.maybe_download("gs://big_vision/paligemma_tokenizer.model", gs={"token": "anon"})  
```
这个下载机制设计得很智能：如果本地缓存中已存在该模型，则直接使用，避免重复下载；否则，会创建一个锁文件确保并发安全，并从 `gs://big_vision/paligemma_tokenizer.model` 下载模型文件。参数 `gs={"token": "anon"}` 表示使用匿名方式访问 GCS 存储桶

下载完成后，代码以二进制读取模式打开文件，并使用 SentencePiece 处理器加载模型

        # 以二进制读取模式打开下载的模型文件
        with path.open("rb") as f:  
            # 初始化SentencePiece处理器
            self._tokenizer = sentencepiece.SentencePieceProcessor(model_proto=f.read())

第二方面，`tokenize` 方法是处理文本输入的核心，它执行以下步骤：

文本清理：首先通过 `strip()` 去除首尾空白，然后将下划线替换为空格，并将换行符也替换为空格，确保输入文本格式一致

    # 定义分词方法，输入为提示文本，返回tokens和mask
    def tokenize(self, prompt: str) -> tuple[np.ndarray, np.ndarray]:  
        # 清理文本：移除首尾空格，将下划线和换行符替换为空格
        cleaned_text = prompt.strip().replace("_", " ").replace("\n", " ")

Tokenizer：将清理后的文本送入 SentencePiece 编码器，设置 `add_bos=True` 添加句子开始token
```
        # 单独将"\n"作为"答案开始"的token
        # 对清理后的文本编码，添加开始标记，并附加换行符的编码
        tokens = self._tokenizer.encode(cleaned_text, add_bos=True) + self._tokenizer.encode("\n")  
```
特别的是，它还单独编码了一个换行符 `\n` 并将其附加到token序列末尾，作为"答案开始"的特殊token。这种设计允许模型明确区分提示和生成内容的边界

长度处理：根据实际编码后的token序列长度

        # 获取token列表长度
        tokens_len = len(tokens)

代码采取两种策略：
如果token数少于 `max_len`，则用 `False` 值填充 `tokens` 序列，同时创建一个掩码 `mask`，其中实际token位置为 `True`(如此，填充位置自然为 `False`)

        # 如果token长度小于最大长度
        if tokens_len < self._max_len:  
            # 创建填充列表，用False填充
            padding = [False] * (self._max_len - tokens_len)  

            # 创建mask列表，真实token位置为True(如此，填充位置自然为False)
            mask = [True] * tokens_len + padding  

            # 对token列表进行填充
            tokens = tokens + padding

如果token数超过 `max_len`，则发出警告并截断序列，掩码全部设为 `True`（因为所有保留的位置都是有效token）

        # 如果token长度大于或等于最大长度
        else:  
            # 如果token长度大于最大长度
            if len(tokens) > self._max_len:  

                # 记录警告日志
                logging.warning(  
                    # 警告token长度超出最大长度，将进行截断
                    f"Token length ({len(tokens)}) exceeds max length ({self._max_len}), truncating. "  
                    # 建议如果频繁发生，增加模型配置中的最大token长度
                    "Consider increasing the `max_token_len` in your model config if this happens frequently."  
                )

            # 截断token列表，只保留前max_len个
            tokens = tokens[: self._max_len]  
            # 创建全True的mask列表，长度为max_len
            mask = [True] * self._max_len

返回结果：最后，方法将token序列和掩码转换为 NumPy 数组并返回，便于后续的模型处理

        # 将token列表和mask列表转换为numpy数组并返回
        return np.asarray(tokens), np.asarray(mask)

1.5.2 FASTTokenizer 类

`FASTTokenizer` 是一个更复杂的Tokenizer，可同时处理文本和动作数据，详见此文《π0开源了且推出自回归版π0-FAST——打造高效Tokenizer：比扩散π0的训练速度快5倍但效果相当(含π0-FAST源码剖析)》

首先是初始化过程

同样下载 PaliGemma Tokenizer模型

# 定义FAST分词器类
class FASTTokenizer:  
    # 初始化方法，设置最大长度和FAST分词器路径
    def __init__(self, max_len: int = 256, fast_tokenizer_path: str = "physical-intelligence/fast"):  

        # 存储最大token长度
        self._max_len = max_len  

        # 下载PaliGemma分词器模型
        path = download.maybe_download("gs://big_vision/paligemma_tokenizer.model", gs={"token": "anon"})  

        # 以二进制读取模式打开模型文件
        with path.open("rb") as f:  
            self._paligemma_tokenizer = sentencepiece.SentencePieceProcessor(model_proto=f.read()

加载专门的 FAST Tokenizer——用于处理动作序列

        # 实例化FAST分词器
        # 从预训练路径加载FAST处理器
        self._fast_tokenizer = AutoProcessor.from_pretrained(fast_tokenizer_path, trust_remote_code=True)

设置 `_fast_skip_tokens = 128` 以跳过 PaliGemma 词汇表末尾的特殊token

        # 跳过PaliGemma词表中的最后128个token，因为它们是特殊token
        self._fast_skip_tokens = 128

其次，是Tokenizer流程

接收文本提示、状态数组和可选的动作数组

    # 定义分词方法
    def tokenize(  
        # 输入：提示文本、状态数组和可选的动作数组
        self, prompt: str, state: np.ndarray, actions: np.ndarray | None  
     # 返回四个numpy数组：tokens、token_mask、ar_mask和loss_mask
    ) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: 

        # 清理文本：转小写，移除首尾空格，将下划线替换为空格
        cleaned_text = prompt.lower().strip().replace("_", " ")

将状态值离散化为 256 个区间（范围 [-1, 1]）

        # 约定：状态被离散化为256个离散区间（假设归一化后的范围：[-1, 1]）
        # 将状态数组离散化为0-255的整数
        discretized_state = np.digitize(state, bins=np.linspace(-1, 1, 256 + 1)[:-1]) - 1

创建格式化前缀prefix，包含文本提示和状态信息

        # 约定：前缀包括提示和状态的字符串表示，后跟';'
        # 将离散化状态转换为空格分隔的字符串
        state_str = " ".join(map(str, discretized_state))  

        # 构建前缀文本，包含任务和状态信息
        prefix = f"Task: {cleaned_text}, State: {state_str};\n"  

        # 使用PaliGemma分词器编码前缀，添加开始token
        prefix_tokens = self._paligemma_tokenizer.encode(prefix, add_bos=True)

如果提供了动作：
使用 FAST Tokenizer对动作进行Tokenizer

        # 如果提供了动作
        if actions is not None:  
            # 使用FAST分词器对动作进行分词，并映射到PaliGemma词表的最后部分

            # 将动作转换为token
            action_tokens = self._fast_tokenizer(actions[None])[0]

通过 `_act_tokens_to_paligemma_tokens` 将这些动作token映射到 PaliGemma 词汇表中

            # 将FAST token转换为PaliGemma token
            action_tokens_in_pg = self._act_tokens_to_paligemma_tokens(action_tokens)

创建包含 "Action:" 的后缀，后跟编码的动作和结束符 "|"

            # 约定：后缀包含'Action:'，然后是FAST token，最后是'|'
            # 构建后缀token
            postfix_tokens = (  
                # 编码"Action: "文本
                self._paligemma_tokenizer.encode("Action: ")  
                 # 添加转换后的动作token
                + action_tokens_in_pg.tolist() 
                # 添加结束分隔符'|'的编码
                + self._paligemma_tokenizer.encode("|")  
            )

        # 如果没有提供动作
        else:  
            # 后缀token为空列表
            postfix_tokens = []

创建三种掩码：
token_mask：指示哪些位置有实际token
ar_mask：自回归掩码（前缀为 0，后缀为 1），控制注意力流
loss_mask：控制哪些标记在训练期间贡献损失（只在后缀上计算损失）

        # 创建输出token序列和掩码
        # AR掩码在前缀上为0（双向注意力），在后缀上为1（对所有先前token的因果注意力）

        # 合并前缀和后缀token
        tokens = prefix_tokens + postfix_tokens  

        # 创建token掩码，全为True
        token_mask = [True] * len(tokens)  

        # 创建自回归掩码，前缀部分为0，后缀部分为1
        ar_mask = [0] * len(prefix_tokens) + [1] * len(postfix_tokens)  

        # 创建损失掩码，仅在后缀部分计算损失
        loss_mask = [False] * len(prefix_tokens) + [True] * len(postfix_tokens)

处理所有token序列和掩码的填充或截断

别忘了，上文所说的

1.2.4.2 特征嵌入方法：embed_prefix(图像和文本输入)、embed_suffix(状态和动作信息)

`embed_prefix`：处理图像和文本输入(图像通过SigLip模型编码，文本通过Gemma LLM编码)，创建前缀 token，皆为双向注意力，用ar_mask = false表示

`embed_suffix`：处理机器人状态信息、噪声化的动作信息(状态和噪声动作经过线性投影和MLP处理)，创建后缀 token
其中
状态为单个token，和第一个动作token均设置为单向注意力，用ar_mask = true表示
其余动作tokens之间设置为双向注意力，用ar_mask = false表示

再其次，是动作提取功能

从token序列中提取动作

    # 定义从token中提取动作的方法
    def extract_actions(self, tokens: np.ndarray, action_horizon: int, action_dim: int) -> np.ndarray:  
        # 解码预测的输出token —— 将token列表解码为文本
        decoded_tokens = self._paligemma_tokenizer.decode(tokens.tolist())

定位 "Action:" 后和 "|" 前的部分

        # 从FAST模型输出中提取动作：如果解码文本中不包含"Action: "
        if "Action: " not in decoded_tokens:  
            # 返回全零动作数组
            return np.zeros((action_horizon, action_dim), dtype=np.float32)

重新映射token以恢复原始动作空间

        # 从解码的token中提取动作
        raw_action_tokens = np.array(  
            # 提取"Action: "和"|"之间的内容，并编码为token
            self._paligemma_tokenizer.encode(decoded_tokens.split("Action: ")[1].split("|")[0].strip())
        )

        # 将原始action token转换为PaliGemma token格式
        action_tokens = self._act_tokens_to_paligemma_tokens(raw_action_tokens)  

        # 使用FAST分词器将token解码为动作向量
        return self._fast_tokenizer.decode(  
            [action_tokens.tolist()], time_horizon=action_horizon, action_dim=action_dim
        )[0]

最后是token映射函数

    # 定义将FAST token转换为PaliGemma token的方法
    def _act_tokens_to_paligemma_tokens(self, tokens: np.ndarray | list[int]) -> np.ndarray:  

         # 如果输入是列表
        if isinstance(tokens, list): 
            # 转换为numpy数组
            tokens = np.array(tokens)  

        # 将FAST token映射到PaliGemma词表的对应位置
        return self._paligemma_tokenizer.vocab_size() - 1 - self._fast_skip_tokens - tokens

`_act_tokens_to_paligemma_tokens` 方法实现了 FAST 动作token到 PaliGemma 词汇空间的双向映射
计算公式：`vocab_size - 1 - skip_tokens - token_id`
这种巧妙的映射让两个不同的Tokenizer系统能够协同工作

1.6 `lora.py` ：实现了LoRA (Low-Rank Adaptation)微调方法

如之前所述，「关于什么是LoRA，详见此文《LLM高效参数微调方法：从Prefix Tuning、Prompt Tuning、P-Tuning V1/V2到LoRA、QLoRA(含对模型量化的解释)》的第4部分」

1.6.1 Einsum类中的setup

`setup` 方法，负责初始化模块所需的所有参数

首先，方法通过调用 `self.param` 创建了一个名为 "w" 的参数，这是模块的主要权重矩阵
接下来，代码使用海象运算符（`:=`）检查是否提供了 `lora_config`。如果存在配置，则进入 LoRA 参数的初始化流程

LoRA 的核心思想是将权重更新分解为两个低秩矩阵 A 和 B 的乘积。为此，代码首先创建了原始形状的可变副本 `shape_a` 和 `shape_b`，使用 `list()` 将可能是元组的 `self.shape` 转换为可修改的列表
随后，`shape_a` 的第二个指定轴（由 `config.axes[1]` 索引）被替换为 `config.rank`
而 `shape_b` 的第一个指定轴（由 `config.axes[0]` 索引）也被替换为相同的 `config.rank`
说白了，就是A矩阵是降维矩阵，故第二个指定轴是rank
b是升维矩阵，故b的第一个指定轴是rank
最后，代码使用 `config.init_fn` 初始化函数(通常是一个小标准差的正态分布)和修改后的形状，创建了两个 LoRA 参数：`self.w_a` 和 `self.w_b`。这些参数分别对应于 LoRA 的 A 和 B 矩阵，它们将在前向传播过程中用于计算 LoRA 更新

1.6.2 Einsum类中的call

`__call__` 方法实现了支持 LoRA (Low-Rank Adaptation) 技术的前向传播逻辑

首先，方法获取并存储输入张量 `x` 的数据类型 (`dtype`)
接下来，方法使用 `jnp.einsum` 函数计算标准的 Einstein 求和乘积，将输入 `x` 与权重矩阵 `self.w` 相乘。注意权重矩阵会被显式转换为与输入相同的数据类型，这是通过 `self.w.astype(dtype)` 实现的
此操作产生的 `result` 变量表示不带 LoRA 修正的基础输出
如果模块配置了 LoRA（通过 `self.lora_config` 存在），代码会进入 LoRA 计算分支。使用海象运算符 (`:=`) 既检查了 `lora_config` 的存在性，又将其赋值给局部变量 `config` 以便后续使用 LoRA 计算过程首先调用 `self._make_lora_eqns` 方法，将原始 einsum 方程转换为两个新方程 `eqn_a` 和 `eqn_b`，分别用于与 LoRA 矩阵 A 和 B 的乘法运算

然后，代码执行这两个 einsum 运算：第一个将输入 `x` 与矩阵 A (`self.w_a`) 相乘，结果存储在 `lora` 变量中；第二个将 `lora` 与矩阵 B (`self.w_b`) 相乘，更新 `lora` 变量
同样，为保持数值一致性，LoRA 参数也会被转换为与输入相同的数据类型

最后，将 LoRA 计算结果乘以配置中指定的缩放值 (`config.scaling_value`)——缩放因子通常设置为 `alpha/rank` 或对于 RS-LoRA 为 `alpha/sqrt(rank)`，并将其添加到基础输出中，形成最终结果

1.6.3 Einsum类中的_make_lora_eqns

_make_lora_eqns负责将标准的 Einstein 求和表达式转换为两个新的表达式，以支持 LoRA 的低秩分解计算。其工作原理基于巧妙的字符串处理，将一个矩阵乘法操作分解为两个连续的矩阵乘法

方法首先执行两项重要的验证
第一个验证，它检查输入的方程 `eqn` 中是否已经包含字符 "L"（默认的 LoRA 标签）
如果存在，方法会抛出 `ValueError` 异常，因为 "L" 被保留用作 LoRA 的特殊维度标识符
```
    def _make_lora_eqns(self, eqn: str) -> tuple[str, str]:
        if "L" in eqn:
            raise ValueError(f"L already in eqn: {eqn}")
```
第二个验证，方法使用正则表达式 `re.match("(.*),(.*)->(.*)", eqn)` 解析输入的 einsum 方程。此正则表达式期望方程遵循标准格式 "lhs,rhs->out"，其中包含三个捕获组：左侧(输入)、右侧(权重)和输出
如果方程格式不符合此模式，方法会抛出另一个 `ValueError`
```
        if not (m := re.match("(.*),(.*)->(.*)", eqn)):
            raise ValueError(f"Unsupported einsum eqn: {eqn}")
```
成功匹配后，方法通过调用 `m.groups()` 提取这三个组件，并将它们分别存储在 `lhs`、`rhs` 和 `out` 变量中
```
        lhs, rhs, out = m.groups()
```
例如，对于方程 "bd,dh->bh"，这些变量将分别包含 "bd"、"dh" 和 "bh"
接下来是方法的核心部分
首先，根据 `self.lora_config.axes` 指定的索引，从 `rhs` 字符串中提取两个关键轴标签 `a_label` 和 `b_label`
```
        assert self.lora_config is not None
        a_label, b_label = (rhs[x] for x in self.lora_config.axes)
        label = self.lora_config.label
```
例如，如果 `rhs` 是 "dh" 且 `axes` 为 (-2, -1)——代表最后两个轴，则
`a_label` 为 "d"
`b_label` 为 "h"

其次，进行两步字符串替换，创建两个新的 einsum 方程
第一步，它将 `rhs` 和 `out` 中的 `b_label` 替换为 LoRA 标签(存储在 `label` 变量中，默认为 "L")。这产生了 `a_rhs` 和 `a_out`，用于构造第一个方程 `eqn_a`

例如，对于前面 "lhs,rhs->out所对应的例子"bd,dh->bh"，`a_rhs`-dh 会变成 "dL"，`a_out`-bh 会变成 "bL"
```
        a_rhs = rhs.replace(b_label, label)
        a_out = out.replace(b_label, label)
```
生成的 `eqn_a` 为 "bd,dL->bL"，表示将输入bd 与 LoRA 矩阵 A dL相乘，得到此第一步的结果bL
```
        eqn_a = f"{lhs},{a_rhs}->{a_out}"
```
第二步，方法创建 `b_rhs`，通过将 `rhs` 中的 `a_label` 替换为 LoRA 标签
使用前面 "lhs,rhs->out所对应的例子"bd,dh->bh"
`b_rhs-dh` 将变为 "Lh"
```
        b_rhs = rhs.replace(a_label, label)
```
然后构造第二个方程 `eqn_b`，形式为 "bL/bL,Lh->bh"
为何这里的输入是bL/bL呢，因为其表示的就是将第一步的结果bL/bL与 LoRA 矩阵 B Lh 相乘
```
        eqn_b = f"{a_out},{b_rhs}->{out}"
```
最后，方法返回这两个新创建的 einsum 方程作为元组
```
        return eqn_a, eqn_b
```
这些方程将被用于在前向传播过程中计算 LoRA 的低秩更新

总的来说，上面的整个过程还是比较绕的，为方便大家一目了然的快速理解，我特意花了10分钟画了个图示——而我一个人多花10分钟，可以让数千人、数万人在理解上少花10分钟，这价值非常大，会更清晰

1.6.4 FeedForward类中的setup、call、_dot

1.7 `vit.py`: Vision Transformer实现

// 待更

第二部分策略适配接口：src下policy的全面分析与解读

src/openpi/policies目录包含以下文件：

BasePolicy (policy.py)
├── Policy
│ ├── BaseModel
│ └── transforms.py
├── AlohaPolicy (aloha_policy.py)
├── DroidPolicy (droid_policy.py)
└── LiberoPolicy (libero_policy.py)

此外，每个特定机器人都有自己的策略文件，如

aloha_policy.py
droid_policy.py
libero_policy.py

这些文件定义了特定于机器人的输入和输出转换函数，处理数据格式、规范化和特定的转换需求

比如每种机器人（ALOHA、DROID、LIBERO）的策略文件定义了特定的输入/输出转换类
这些转换类作为 `transforms` 参数传递给 `Policy` 构造函数，例如，`AlohaInputs` 处理 ALOHA 机器人特有的状态和图像格式，`AlohaOutputs` 处理对应的输出转换

2.1 policy.py：实现了Policy类和 PolicyRecorder类

2.1.1 Policy 类

policy.py 定义了基本的 `Policy` 类和 `PolicyRecorder` 类，它们继承自`openpi_client.base_policy.BasePolicy`

首先，做一系列初始化

class Policy(BasePolicy):  # 定义Policy类，继承自BasePolicy
    def __init__(
        self,
        model: _model.BaseModel,  # 模型参数，必须是BaseModel的实例
        *,  # 之后的所有参数必须使用关键字传递
        rng: at.KeyArrayLike | None = None,  # 随机数生成器，可选

        # 输入转换函数序列，默认为空
        transforms: Sequence[_transforms.DataTransformFn] = (),  

        # 输出转换函数序列，默认为空
        output_transforms: Sequence[_transforms.DataTransformFn] = (),  

        # 传递给sample_actions的额外参数，可选
        sample_kwargs: dict[str, Any] | None = None,  

        metadata: dict[str, Any] | None = None,  # 元数据字典，可选
    ):

        # 使用JIT编译model的sample_actions方法提高性能
        self._sample_actions = nnx_utils.module_jit(model.sample_actions)  

        # 组合所有输入转换函数为一个函数
        self._input_transform = _transforms.***pose(transforms)  

        # 组合所有输出转换函数为一个函数
        self._output_transform = _transforms.***pose(output_transforms)  
        self._rng = rng or jax.random.key(0)       # 设置随机数生成器，如果未提供则创建一个新的
        self._sample_kwargs = sample_kwargs or {}  # 存储采样参数，如果未提供则使用空字典
        self._metadata = metadata or {}            # 存储元数据，如果未提供则使用空字典

其次，对于infer 方法——在策略内部流程上

复制输入观察数据

    def infer(self, obs: dict) -> dict:  # type: ignore[misc]  # 推理方法，接收观察字典，返回动作字典
        # 复制输入，因为转换可能会修改输入
        inputs = jax.tree.map(lambda x: x, obs)  # 使用JAX树映射创建输入的深拷贝

应用输入转换
Policy.infer` 方法首先应用输入转换：self._input_transform，将客户端提供的观察转换为模型所需的格式
```
        inputs = self._input_transform(inputs)  # 应用输入转换函数处理输入数据
```

将数据转换为批处理格式并转为 JAX 数组

        # 将输入转换为批处理格式并转为jax数组
        inputs = jax.tree.map(lambda x: jnp.asarray(x)[np.newaxis, ...], inputs)  # 添加批次维度并转为JAX数组

生成新的随机数键

        self._rng, sample_rng = jax.random.split(self._rng)  # 分割随机数键以保持随机性

模型推理
调用模型的 `sample_actions` 方法「该方法的实现，详见上文的1.2.4.4 推理函数 `sample_actions`：基于扩散模型逆向采样，生成机器人动作序列」进行推理，即获取动作预测

        outputs = {
            "state": inputs["state"],  # 保留状态信息
            "actions": self._sample_actions(sample_rng, _model.Observation.from_dict(inputs), **self._sample_kwargs),  # 使用模型生成动作
        }

解除批处理并转换为 NumPy 数组

        # 移除批次维度并转换为NumPy数组
        outputs = jax.tree.map(lambda x: np.asarray(x[0, ...]), outputs)  # 取第一个样本并转为NumPy数组

输出转换
最后应用输出转换 (`self._output_transform`)，将模型输出转换为客户端期望的格式
```
        return self._output_transform(outputs)  # 应用输出转换并返回结果
```

2.1.2 `PolicyRecorder`

PolicyRecorder是一个装饰器类，它包装了一个基础策略，并在执行策略的同时将所有的输入和输出保存到磁盘，用于记录策略的行为

对于初始化函数：`policy`，涉及被包装的基础策略、record_dir`：保存记录的目录路径

对于infer 方法

调用被包装策略的 `infer` 方法获取结果
将输入和输出数据组织为字典
使用 Flax 的 `flatten_dict` 函数将嵌套字典展平
构建输出文件路径
将数据保存为 NumPy 数组文件
返回策略结果

// 待更

2.2 policy_config.py

policy_config.py 定义了 `PolicyConfig` 类和 `create_trained_policy` 函数
`create_trained_policy` 函数用于从训练好的检查点创建策略实例，加载模型参数、归一化统计数据，并配置转换函数

相当于客户端代码会实例化一个 `Policy` 对象，通常是通过 `create_trained_policy` 函数，客户端通过调用 `policy.infer(obs)` 方法获取策略输出

2.2.1 PolicyConfig 数据类

`PolicyConfig` 是一个使用 `@dataclasses.dataclass` 装饰的数据类，用于存储创建策略所需的所有配置信息：

 # 定义策略配置类
class PolicyConfig:     
    model: _model.BaseModel      # 模型实例，必须是BaseModel类型
    norm_stats: dict[str, transforms.NormStats]        # 归一化统计信息，键是特征名称，值是归一化统计数据

    input_layers: Sequence[transforms.DataTransformFn]      # 输入数据转换函数序列
    output_layers: Sequence[transforms.DataTransformFn]     # 输出数据转换函数序列

    model_type: _model.ModelType = _model.ModelType.PI0     # 模型类型，默认为PI0
    default_prompt: str | None = None                  # 默认提示文本，可选
    sample_kwargs: dict[str, Any] | None = None        # 采样参数字典，可选

这个类主要是作为配置容器，将所有策略创建时需要的参数组织在一起

2.2.2 create_trained_policy 函数

`create_trained_policy` 函数是从训练好的检查点创建可用策略的工厂函数

def create_trained_policy(
    train_config: _config.TrainConfig,       # 训练配置对象，包含训练时的所有参数设置
    checkpoint_dir: pathlib.Path | str,      # 检查点目录路径，可以是Path对象或字符串
    *,  # 强制后续参数使用关键字传递
    repack_transforms: transforms.Group | None = None,  # 可选的重新打包转换组
    sample_kwargs: dict[str, Any] | None = None,        # 采样参数，可选
    default_prompt: str | None = None,                  # 默认提示文本，可选
    norm_stats: dict[str, transforms.NormStats] | None = None,  # 归一化统计信息，可选
) -> _policy.Policy:                         # 返回类型是Policy对象

函数的核心流程是：

处理输入参数，确保 `repack_transforms` 不为空
且检查并可能下载检查点目录

    repack_transforms = repack_transforms or transforms.Group()      # 确保repack_transforms不为空，如果未提供则创建空Group
    checkpoint_dir = download.maybe_download(str(checkpoint_dir))    # 检查并可能下载检查点目录

使用 `train_config` 加载模型参数

    logging.info("Loading model...")  # 记录日志，表示正在加载模型

    # 加载模型参数并创建模型实例，使用bfloat16数据类型
    model = train_config.model.load(_model.restore_params(checkpoint_dir / "params", dtype=jnp.bfloat16))

创建数据配置

    data_config = train_config.data.create(train_config.assets_dirs, train_config.model)  # 创建数据配置
    if norm_stats is None:  # 如果未提供归一化统计信息
        # 我们从检查点而非配置资源目录加载归一化统计信息，以确保策略使用与原始训练过程相同的归一化统计信息

如果未提供 `norm_stats`，从检查点加载归一化统计信息

        if data_config.asset_id is None:  # 如果数据配置中没有asset_id
            raise ValueError("Asset id is required to load norm stats.")  # 抛出异常，需要asset_id来加载归一化统计信息
        norm_stats = _checkpoints.load_norm_stats(checkpoint_dir / "assets", data_config.asset_id)  # 从检查点加载归一化统计信息

构建并返回 `Policy` 实例，将所有转换函数组织为有序的处理流程：

    return _policy.Policy(  # 创建并返回Policy实例
        model,  # 传入模型

输入处理：重新打包转换 → 注入默认提示 → 数据转换 → 归一化 → 模型特定转换

        transforms=[  # 输入转换函数序列
            *repack_transforms.inputs,          # 展开重打包转换的输入部分
            transforms.InjectDefaultPrompt(default_prompt),  # 注入默认提示
            *data_config.data_transforms.inputs,   # 展开数据转换的输入部分
            transforms.Normalize(norm_stats, use_quantiles=data_config.use_quantile_norm),      # 添加归一化转换
            *data_config.model_transforms.inputs,  # 展开模型特定转换的输入部分
        ],

输出处理：模型特定转换 → 反归一化 → 数据转换 → 重新打包转换

        output_transforms=[  # 输出转换函数序列
            *data_config.model_transforms.outputs,     # 展开模型特定转换的输出部分
            transforms.Unnormalize(norm_stats, use_quantiles=data_config.use_quantile_norm),          # 添加反归一化转换
            *data_config.data_transforms.outputs,      # 展开数据转换的输出部分
            *repack_transforms.outputs,           # 展开重打包转换的输出部分
        ],
        sample_kwargs=sample_kwargs,              # 设置采样参数
        metadata=train_config.policy_metadata,    # 设置策略元数据
    )

`create_trained_policy` 函数是框架中连接训练过的模型与实际部署使用的关键桥梁，它通过组合各种转换函数，创建出可直接用于推理的 `Policy` 实例

2.3 policies/aloha_policy.py

这段代码实现了一个用于 Aloha 策略的输入输出处理和数据转换的模块

2.3.1 make_aloha_example：输入示例——状态向量、图像数据、文本prompt

首先，`make_aloha_example` 函数创建了一个随机的输入示例，包括一个14维的状态向量和四个摄像头的图像数据（高、低、左腕、右腕视角），以及一个文本提示信息

# 定义一个函数，创建Aloha策略的随机输入示例
def make_aloha_example() -> dict:  
    # 返回一个字典，包含状态、图像和提示信息
    return {  
        # 创建一个14维的状态向量，所有值为1
        "state": np.ones((14,)),  

        # 创建一个包含四个摄像头图像的字典
        "images": {  
            # 高位摄像头图像
            "cam_high": np.random.randint(256, size=(3, 224, 224), dtype=np.uint8),  

            # 低位摄像头图像
            "cam_low": np.random.randint(256, size=(3, 224, 224), dtype=np.uint8), 

            # 左手腕摄像头图像 
            "cam_left_wrist": np.random.randint(256, size=(3, 224, 224), dtype=np.uint8),  

            # 右手腕摄像头图像
            "cam_right_wrist": np.random.randint(256, size=(3, 224, 224), dtype=np.uint8),  
        },
        "prompt": "do something", 
    }

这些数据将用于测试和验证 Aloha 策略的输入处理

可能有的同学对上面的4个摄像头有疑问，简单，详见此文《一文通透动作分块算法ACT：斯坦福ALOHA团队推出的动作序列预测算法(Action Chunking with Transformers)》的「1.2 硬件套装：ALOHA——低成本的开源硬件系统，用于手动远程操作」

如下图所示

左侧为前、顶部和两个手腕摄像机的视角(这4个相机的视角分别用从当前往后的蓝线、从顶向下的绿线、从左往右的红线、从右往左的红线表示)，以及ALOHA双手工作空间的示意图

具体而言，总计4个Logitech C922x网络摄像头，每个流输出480×640 RGB图像
其中两个网络摄像头安装在跟随机器人手腕上，以提供夹具的近距离视角(allowing for a close-up view of the grippers)
剩下的两个相机分别安装在桌面的前方(front camera)和桌子上方的顶部位置(top camera)，遥控操作和数据记录均以50Hz频率进行

2.3.2 AlohaInputs：定义Aloha 策略的输入数据结构

接下来，`AlohaInputs` 类定义了 Aloha 策略的输入数据结构

class AlohaInputs(transforms.DataTransformFn):  # 定义AlohaInputs类，继承自transforms.DataTransformFn
    """Inputs for the Aloha policy.
    # 预期输入格式
    # 图像字典，键是名称，值是形状为[channel, height, width]的图像
    - images: dict[name, img]

     # 状态向量，长度为14
    - state: [14] 

    # 动作矩阵，形状为[action_horizon, 14]
    - actions: [action_horizon, 14]  
    """

    # 模型的动作维度，将用于填充状态和动作
    action_dim: int  # 动作维度

    # 如果为True，将关节和夹持器值从标准Aloha空间转换为pi内部运行时使用的空间
    # pi内部运行时使用的空间用于训练基础模型
    # 是否适配pi内部运行时，默认为True
    adapt_to_pi: bool = True  

    # 预期的摄像头名称，所有输入摄像头必须在此集合中。缺失的摄像头将用黑色图像替代
    # 缺失的摄像头将用黑色图像替代，对应的`image_mask`将设置为False
    # 预期的摄像头名称集合
    EXPECTED_CAMERAS: ClassVar[tuple[str, ...]] = ("cam_high", "cam_low", "cam_left_wrist", "cam_right_wrist")

这个类使用 `dataclasses.dataclass` 装饰器来简化类的定义，并确保实例是不可变的（`frozen=True`）
类中定义了输入数据的预期格式，包括图像、状态和动作数据

__call__方法，实现了对Aloha策略输入数据的标准化处理。该方法将原始输入数据转换为模型可接受的格式，包括多项关键处理步骤，比如进行必要的解码和填充操作，并检查图像数据是否包含预期的摄像头视角

首先，方法通过调用`_decode_aloha`函数对输入数据进行初步解码，根据`adapt_to_pi`参数决定是否将数据适配到π内部运行时环境
```
    # 定义__call__方法，处理输入数据
    def __call__(self, data: dict) -> dict:  

        # 解码Aloha数据，根据adapt_to_pi参数进行适配
        data = _decode_aloha(data, adapt_to_pi=self.adapt_to_pi)  
```
这一步主要处理状态向量以及将图像格式从`[channel, height, width]`转换为`[height, width, channel]`

接着，方法将14维的状态向量使用零填充扩展到模型所需的动作维度(`action_dim`)

        # 获取状态数据，将其从14维填充到模型的动作维度
        # 使用transforms.pad_to_dim函数填充状态数据
        state = transforms.pad_to_dim(data["state"], self.action_dim)

随后，进行输入图像的验证：检查输入图像的键集合是否超出了预期的摄像头列表范围，若发现未知摄像头视角则抛出`ValueError`

        # 获取输入图像数据
        in_images = data["images"]  

        # 检查输入图像是否包含所有预期的摄像头
        if set(in_images) - set(self.EXPECTED_CAMERAS):  
            # 如果缺少预期的摄像头，抛出异常
            raise ValueError(f"Expected images to contain {self.EXPECTED_CAMERAS}, got {tuple(in_images)}")

在构建输出字典时，方法首先假定"cam_high"（高视角摄像头）图像必定存在

        # 假设基础图像总是存在，获取高位摄像头图像
        base_image = in_images["cam_high"]

并将其作为基础图像（`base_0_rgb`）

        # 创建图像字典
        images = {  
            # 基础图像
            "base_0_rgb": base_image,  
        }

同时创建了相应的图像掩码字典，标记该图像为有效

        # 创建图像掩码字典
        image_masks = {  
            # 基础图像掩码为True
            "base_0_rgb": np.True_,  
        }

对于其他摄像头视角（左腕和右腕），方法使用映射关系字典进行处理：

        # 添加额外的图像
        # 额外图像名称映射
        extra_image_names = {  
            # 左手腕图像
            "left_wrist_0_rgb": "cam_left_wrist",  

            # 右手腕图像
            "right_wrist_0_rgb": "cam_right_wrist",  
        }

如果相应的源图像存在，则将其添加到输出图像字典并标记为有效；

        # 遍历额外图像名称映射
        for dest, source in extra_image_names.items():  

            # 如果输入图像中包含该图像
            if source in in_images:  
                # 添加到图像字典
                images[dest] = in_images[source]  

                # 设置图像掩码为True
                image_masks[dest] = np.True_

若不存在，则创建一个与基础图像相同大小的全零图像（黑图），并标记为无效

            # 如果输入图像中不包含该图像
            else:  
                # 用黑色图像替代
                images[dest] = np.zeros_like(base_image)  

                # 设置图像掩码为False
                image_masks[dest] = np.False_

这种处理方式确保了模型在缺失某些视角图像时仍能正常工作

        # 创建输入字典
        inputs = {  
            "image": images,              # 图像数据
            "image_mask": image_masks,    # 图像掩码
            "state": state,               # 状态数据
        }

方法还会处理训练时特有的数据，如动作序列
若输入数据包含"actions"字段，则将其转换为NumPy数组，应用`_encode_actions_inv`进行编码转换，并使用零填充扩展到模型动作维度

        # 动作数据仅在训练期间可用
        # 如果输入数据中包含动作数据
        if "actions" in data:  
             # 将动作数据转换为NumPy数组
            actions = np.asarray(data["actions"]) 

            # 编码动作数据，根据adapt_to_pi参数进行适配
            actions = _encode_actions_inv(actions, adapt_to_pi=self.adapt_to_pi)  

            # 填充动作数据到模型的动作维度
            inputs["actions"] = transforms.pad_to_dim(actions, self.action_dim)

最后，如果输入包含"prompt"文本提示，也会将其添加到输出字典中，然后返回处理后的输入数据

        # 如果输入数据中包含提示信息
        if "prompt" in data:  
            # 添加提示信息到输入字典
            inputs["prompt"] = data["prompt"]  

        # 返回处理后的输入数据
        return inputs

整体而言，这个方法实现了从多样化的原始输入到标准化模型输入的转换流程，处理了数据格式转换、缺失数据补充、维度调整等核心问题，确保了Aloha策略模型能够接收一致的输入格式，从而实现稳定的推理和训练

2.3.3 AlohaOutputs：定义Aloha 策略的输出数据结构

`AlohaOutputs` 类定义了 Aloha 策略的输出数据结构，同样使用 `dataclasses.dataclass` 装饰器

# 定义AlohaOutputs类，继承自transforms.DataTransformFn
class AlohaOutputs(transforms.DataTransformFn):  

    # 如果为True，将关节和夹持器值从标准Aloha空间转换为pi内部运行时使用的空间
    # pi内部运行时使用的空间用于训练基础模型
    adapt_to_pi: bool = True  # 是否适配pi内部运行时，默认为True

`__call__` 方法处理输出数据，仅返回前14个维度的动作数据，并进行必要的编码转换

    # 定义__call__方法，处理输出数据
    def __call__(self, data: dict) -> dict:  
        # 仅返回前14维的动作数据，即将动作数据转换为NumPy数组，并取前14维
        actions = np.asarray(data["actions"][:, :14])  

        # 编码动作数据并返回字典
        return {"actions": _encode_actions(actions, adapt_to_pi=self.adapt_to_pi)}

2.3.4 多个辅助函数：数据的标准化、反标准化、关节角度翻转

此外，代码中还包含多个辅助函数，用于数据的标准化、反标准化、关节角度翻转、夹持器位置的线性和角度转换等

这些函数确保了数据在不同控制系统之间的兼容性和一致性

// 待更

第三部分模型训练的配置：src下training模块的全面分析与解读

training模块是 OpenPI 项目中负责训练相关功能的核心部分，该目录下包含了以下主要文件：

checkpoints.py - 检查点管理
config.py - 配置系统
data_loader.py - 数据加载器
data_loader_test.py - 数据加载器测试
optimizer.py - 优化器实现
sharding.py - 模型分片工具
utils.py - 通用工具函数
weight_loaders.py - 模型权重加载器

3.1 配置系统 (config.py)

定义了训练过程的各种配置类型，包括：

`TrainConfig`：顶级训练配置，包含模型、数据、优化器等所有训练参数
`DataConfigFactory`：抽象工厂类，用于创建特定环境的数据配置
`AssetsConfig`：管理资产（如归一化统计数据）的位置
预定义了多种常用配置（如 ALOHA、DROID、LIBERO 等环境的配置）
通过 `get_config` 函数根据名称检索预定义配置

在配置流程上

- 训练脚本通过 `_config.cli()` 或 `_config.get_config()` 获取配置
- 配置系统加载预定义的训练参数，确定训练环境和模型参数
- 数据配置通过工厂模式创建，根据不同环境（ALOHA、DROID 等）提供不同的预处理流程

3.1.1 基础配置类AssetsConfig、DataConfig

一个是AssetsConfig

class AssetsConfig:
    """用于确定数据pipeline所需资产(如归一化统计信息)的位置"""
    assets_dir: str | None = None      # 资产目录
    asset_id: str | None = None        # 资产ID

一个是DataConfig

@dataclasses.dataclass(frozen=True)
class DataConfig:
    repo_id: str | None = None            # 数据集仓库ID
    asset_id: str | None = None           # 资产ID
    norm_stats: dict[str, _transforms.NormStats] | None = None  # 归一化统计信息
    repack_transforms: _transforms.Group  # 数据重打包转换
    data_transforms: _transforms.Group    # 数据预处理转换
    model_transforms: _transforms.Group   # 模型特定转换

3.1.2 数据集配置：包含ALOHA、Libero两套数据集——LeRobotLiberoDataConfig

涉及两个配置

一个是LeRobotAlohaDataConfig

@dataclasses.dataclass(frozen=True)
class LeRobotAlohaDataConfig(DataConfigFactory):
    """ALOHA数据集配置"""
    use_delta_joint_actions: bool = True      # 是否使用关节角度增量
    default_prompt: str | None = None         # 默认提示语
    adapt_to_pi: bool = True                  # 是否适配到π内部运行时

一个是LeRobotLiberoDataConfig

@dataclasses.dataclass(frozen=True)
class LeRobotLiberoDataConfig(DataConfigFactory):
    """Libero数据集配置"""

对于后者的结构，详见下图

`LeRobotLiberoDataConfig` 是一个用于机器人控制系统的数据配置类，它负责定义整个数据管道中不同阶段的数据转换操作。这个类通过 `@dataclasses.dataclass(frozen=True)` 装饰器声明为不可变数据类，确保配置一旦创建就不能被修改，增强了数据处理的稳定性
该类重写了基类 `DataConfigFactory` 的 `create` 方法，该方法是整个配置系统的核心，负责构建完整的数据配置
```
    def create(self, assets_dirs: pathlib.Path, model_config: _model.BaseModelConfig) -> DataConfig:
        # 重写父类方法，创建数据配置。参数包括资产目录路径和模型配置，返回DataConfig对象
        # ..
```
方法接收两个关键参数：存放数据资产的目录路径和模型配置对象，然后返回一个完整的 `DataConfig` 对象
在方法内部，首先定义了 `repack_transform`，这是一个仅在训练阶段应用的转换器，用于将数据集中的键名映射到推理环境期望的键名
例如，将 `"observation/image"` 映射到 `"image"`。这种转换确保了训练数据和推理环境之间的一致性，是适配不同数据源的关键步骤

接下来，`data_transforms` 配置了同时应用于训练和推理阶段的转换操作
它使用 `libero_policy.LiberoInputs` 处理输入数据，`libero_policy.LiberoOutputs` 处理输出数据

        # 数据转换应用于来自数据集的数据和推理过程中的数据
        # 下面，定义了进入模型的数据转换（"inputs"）和从模型输出的数据转换（"outputs"）（后者仅在推理时使用）
        # 这些转换在`libero_policy.py`中定义
        # 一旦创建了自己的转换，你可以用自己的替换下面的转换
        data_transforms = _transforms.Group(
             # 定义输入转换，使用LiberoInputs处理器
            inputs=[libero_policy.LiberoInputs(action_dim=model_config.action_dim, model_type=model_config.model_type)], 

            # 定义输出转换，使用LiberoOutputs处理器
            outputs=[libero_policy.LiberoOutputs()],  
        )

这些转换器负责将原始数据调整为模型能够处理的格式

特别值得注意的是关于动作表示的转换：该配置支持将绝对动作（如具体的关节角度）转换为相对动作（相对于初始状态的变化量）
通过 `delta_action_mask` 创建一个布尔掩码，指定哪些动作维度需要进行转换（这里是前6个维度对应机器人关节，保留最后一个维度对应夹爪不变）
```
        # 创建动作掩码，指定哪些维度需要转换为相对动作（前6个关节），哪些保持绝对值（夹爪）
        # 创建布尔掩码，前6个维度为True，最后一个维度为False
        delta_action_mask = _transforms.make_bool_mask(6, -1)  
```
这对于训练基于相对动作的模型（如Pi0模型）非常重要
最后，`model_transforms` 处理模型特有的转换操作，比如提示文本的token化和图像尺寸调整
```
        # 使用模型配置创建模型转换——处理提示文本的token化和其他模型特定的转换
        model_transforms = ModelTransformFactory()(model_config)  
```
这些转换由 `ModelTransformFactory` 根据模型类型动态创建，支持不同类型的模型（Pi0或Pi0_FAST）

整个方法通过 `dataclasses.replace` 将这些转换器与基础配置（通过 `create_base_config` 创建）合并，生成最终的数据配置对象

        return dataclasses.replace(
            self.create_base_config(assets_dirs),         # 创建基础配置
            repack_transforms=repack_transform,           # 设置重新打包转换
            data_transforms=data_transforms,              # 设置数据转换
            model_transforms=model_transforms,            # 设置模型转换
        )

3.1.3 训练配置TrainConfig：模型、数据、优化器等训练参数的设置

class TrainConfig:
    name: str                              # 配置名称
    project_name: str = "openpi"           # 项目名称
    exp_name: str                          # 实验名称
    model: _model.BaseModelConfig          # 模型配置
    batch_size: int = 32                   # 批次大小
    num_train_steps: int = 30_000          # 训练步数
    lr_schedule: _optimizer.LRScheduleConfig      # 学习率调度
    optimizer: _optimizer.OptimizerConfig         # 优化器配置

3.1.4 预定义配置：基于ALOHA/Libero数据集微调π0——比如完成aloha_sim_transfer_cube_human

文件最后定义了多个具体的训练配置：

比如ALOHA的

TrainConfig(
    name="pi0_aloha_pen_uncap",      # 配置名称，反映模型和数据集
    model=pi0.Pi0Config(),           # 使用pi0模型配置
    data=LeRobotAlohaDataConfig(     # 使用LeRobotAloha数据集配置

        # 数据集仓库ID
        repo_id="physical-intelligence/aloha_pen_uncap_diverse",  

        # 资产配置
        assets=AssetsConfig(  
            # 资产目录
            assets_dir="s3://openpi-assets/checkpoints/pi0_base/assets",  
            # 资产ID
            asset_id="trossen",          
        ),
        # 默认提示语
        default_prompt="uncap the pen",  

        # 数据重打包转换
        repack_transforms=_transforms.Group(      
            inputs=[
                # 重打包转换
                _transforms.RepackTransform(      
                    {
                        "images": {
                            # 高视角摄像头图像
                            "cam_high": "observation.images.cam_high",  

                            # 左手腕摄像头图像
                            "cam_left_wrist": "observation.images.cam_left_wrist",

                            # 右手腕摄像头图像  
                            "cam_right_wrist": "observation.images.cam_right_wrist",                  
                        },

                        # 机器人状态
                        "state": "observation.state",  

                        # 动作
                        "actions": "action",           
                    }
                )
            ]
        ),

        base_config=DataConfig(
            # 是否只使用本地数据集，False表示允许从Hugging Face下载
            local_files_only=False,  
        ),
    ),

    # 加载预训练权重
    weight_loader=weight_loaders.CheckpointWeightLoader("s3://openpi-assets/checkpoints/pi0_base/params"),  

    # 训练步数为20,000步
    num_train_steps=20_000,  
),

当然，这里面还涉及到ALOHA中一个仿真环境中的操作任务

# 这个配置用于演示如何在简单的模拟环境中进行训练
TrainConfig(
    name="pi0_aloha_sim",          # 配置名称
    model=pi0.Pi0Config(),         # 使用pi0模型配置
    data=LeRobotAlohaDataConfig(   # 使用LeRobotAloha数据集配置

         # 数据集仓库ID
        repo_id="lerobot/aloha_sim_transfer_cube_human", 
        default_prompt="Transfer cube",      # 默认提示语
        use_delta_joint_actions=False,       # 是否使用关节角度增量
    ),
    weight_loader=weight_loaders.CheckpointWeightLoader("s3://openpi-assets/checkpoints/pi0_base/params"),        # 加载预训练权重
    num_train_steps=20_000,                  # 训练步数为20,000步
),

再比如Libero的

TrainConfig(
    # 更改名称以反映你的模型和数据集
    name="pi0_libero",
    
    # 在这里定义模型配置 - 这个例子中我们使用pi0作为模型架构并执行完整微调
    # 在后面的例子中我们会展示如何修改配置来执行低内存(LORA)微调
    # 以及如何使用pi0-FAST作为替代架构
    model=pi0.Pi0Config(),
    
    # 在这里定义要训练的数据集。这个例子中我们使用Libero数据集
    # 对于你自己的数据集，你可以更改repo_id指向你的数据集
    # 同时修改DataConfig以使用你为数据集创建的新配置
    data=LeRobotLiberoDataConfig(
        # 指定数据集的Hugging Face仓库ID
        repo_id="physical-intelligence/libero",
        
        # 基础配置设置
        base_config=DataConfig(
            # 是否只使用本地数据集，False表示允许从Hugging Face下载
            local_files_only=False,  
            
            # 这个标志决定是否从LeRobot数据集的task字段加载提示(即任务指令)
            # 如果设为True，提示将会出现在输入字典的prompt字段中
            # 推荐设置为True
            prompt_from_task=True,
        ),
    ),
    
    # 在这里定义要加载哪个预训练检查点来初始化模型
    # 这应该与你上面选择的模型配置匹配 - 即在这种情况下我们使用pi0基础模型
    weight_loader=weight_loaders.CheckpointWeightLoader(
        "s3://openpi-assets/checkpoints/pi0_base/params"
    ),
    
    # 在下面你可以定义其他超参数，如学习率、训练步数等
    # 查看TrainConfig类以获取完整的可用超参数列表
    num_train_steps=30_000,  # 设置训练步数为30,000步
),

3.2 数据加载系统 data_loader.py

定义了数据集和数据加载器的接口（`Dataset` 和 `DataLoader`）

实现了数据转换管道，将原始数据转换为模型可用的格式
支持各种数据源：真实数据集（通过 LeRobot 数据集接口）、模拟数据（使用 `FakeDataset`）
提供数据归一化和转换功能

在数据加载流程上

TrainConfig
└── data (DataConfigFactory)
├── create() → DataConfig
│ ├── repo_id: 数据集 ID
│ ├── norm_stats: 归一化统计数据
│ ├── repack_transforms: 数据重包装转换
│ ├── data_transforms: 特定于环境的转换
│ └── model_transforms: 特定于模型的转换
└── _load_norm_stats() → 归一化统计数据

create_data_loader(config)
├── data_config = config.data.create()
├── dataset = create_dataset(data_config, config.model)
├── dataset = transform_dataset(dataset, data_config)
└── return DataLoaderImpl(data_config, TorchDataLoader(...))

3.2.1 FakeDataset类

3.2.2 create_dataset：创建适合训练的数据集

`create_dataset` 函数是一个关键的数据准备工具，负责根据配置参数创建适合模型训练的数据集。这个函数通过处理不同数据源和应用必要的转换，为模型提供标准化的训练数据。

首先，函数检查 `data_config.repo_id` 的值，这个参数指定了数据仓库的标识符

def create_dataset(data_config: _config.DataConfig, model_config: _model.BaseModelConfig) -> Dataset:
    """创建用于训练的数据集"""
    # 从数据配置中获取仓库ID
    repo_id = data_config.repo_id

如果 `repo_id` 为 `None`，函数会抛出 `ValueError` 异常，明确指出无法创建数据集。这是一种防御性编程的体现，确保基本的配置参数存在

    # 如果仓库ID为空，抛出错误
    if repo_id is None:
        raise ValueError("Repo ID is not set. Cannot create dataset.")

如果 `repo_id` 的值为 "fake"，函数则创建并返回一个 `FakeDataset` 实例，其样本数设为 1024。这种虚拟数据集在测试模型架构、调试训练流程或者进行性能基准测试时非常有用，无需加载真实数据即可快速验证系统功能

    # 如果是fake数据集，返回包含1024个样本的假数据集
    if repo_id == "fake":
        return FakeDataset(model_config, num_samples=1024)

对于其他情况（即使用真实数据），函数首先创建 `LeRobotDatasetMetadata` 对象来获取数据集的元信息

    # 创建数据集元数据对象，包含数据集的基本信息（如fps等）
    dataset_meta = lerobot_dataset.LeRobotDatasetMetadata(
        repo_id, 
        local_files_only=data_config.local_files_only
    )

然后初始化 `LeRobotDataset` 实例

    # 创建LeRobot数据集实例
    dataset = lerobot_dataset.LeRobotDataset(
        data_config.repo_id,
        # 创建时间戳字典，用于采样动作序列
        delta_timestamps={
            # 对每个动作序列键，根据模型的动作视界长度和数据集的fps生成时间戳列表
            key: [t / dataset_meta.fps for t in range(model_config.action_horizon)]
            for key in data_config.action_sequence_keys
        },
        # 是否只使用本地文件
        local_files_only=data_config.local_files_only,
    )

特别值得注意的是，函数会根据模型的 `action_horizon`（动作预测的时间步长）和数据集的帧率（fps）计算 `delta_timestamps`，这些时间戳用于在时序数据中定位动作序列。这种计算确保了动作序列的时间间隔与模型预期一致，无论原始数据的采样率如何

最后，如果 `data_config.prompt_from_task` 设置为 `True`，函数会将原始数据集包装在 `TransformedDataset` 中，并应用 `PromptFromLeRobotTask` 转换

    # 如果配置指定从任务中提取提示信息
    if data_config.prompt_from_task:
        # 创建转换后的数据集，应用PromptFromLeRobotTask转换，将任务描述转换为提示
        dataset = TransformedDataset(
            dataset, 
            [_transforms.PromptFromLeRobotTask(dataset_meta.tasks)]
        )

这个转换可能将任务描述转换为自然语言提示，增强模型对任务上下文的理解能力
然后返回处理好的数据集

    # 返回处理后的数据集
    return dataset

3.2.3 transform_dataset：对数据集应用转换，比如数据清洗等(创建TransformedDataset实例)

`transform_dataset` 函数是数据预处理管道中的关键组件，负责对原始数据集应用一系列转换操作，以满足模型训练的需求。该函数接收一个原始数据集、数据配置对象以及一个可选的控制标志，并返回经过转换的新数据集

首先，函数会处理数据归一化统计信息（normalization statistics）。对于实际数据集（非"fake"数据集），如果没有显式跳过归一化统计（`skip_norm_stats=False`），函数会检查数据配置中是否包含必要的归一化统计数据。如果这些统计数据缺失，函数会抛出一个明确的错误信息，提示用户需要运行特定脚本来计算这些统计数据。这种检查机制确保了数据归一化步骤能够正确执行，避免了训练过程中可能出现的数值问题

核心转换逻辑通过创建一个 `TransformedDataset` 实例来实现，该实例封装了原始数据集和一系列转换函数。这些转换函数按照特定顺序应用：

首先是数据重新打包转换（`repack_transforms`），可能用于调整数据的基本结构
接着是一般数据转换（`data_transforms`），处理数据清洗、增强等操作
然后应用归一化转换（`Normalize`），使用前面获取的统计数据
最后是模型特定的转换（`model_transforms`），针对特定模型架构的数据格式要求

3.2.4 create_data_loader：创建用于训练的数据加载器

`create_data_loader` 函数是整个数据处理流水线的核心组件，它协调多个模块共同工作，创建一个用于模型训练的数据加载器

整个函数的工作流程可以分为三个主要阶段：

第一阶段：数据集准备
函数首先通过调用 `data_config.create()` 方法创建数据配置对象，该对象包含了所有数据处理相关的配置信息

随后，通过 `create_dataset` 函数创建原始数据集，这可能是一个真实的机器人数据集或者是一个用于测试的假数据集（当 `repo_id` 为 "fake" 时）

然后，调用 `transform_dataset` 函数应用一系列数据转换，包括数据重新打包、数据清洗、归一化和模型特定转换。这些转换确保了原始数据被正确处理为模型所需的格式
第二阶段：PyTorch 数据加载器创建
接下来，函数实例化一个 `TorchDataLoader` 对象，这是对 PyTorch 数据加载器的封装。这个过程涉及多个关键参数设置：计算各进程的本地批量大小（通过全局批量大小除以进程数）
配置数据分片策略（sharding）用于分布式训练
设置是否打乱数据、工作进程数和随机种子等

`TorchDataLoader` 的设计支持无限迭代数据（当 `num_batches` 为 `None` 时）或限定批次数的迭代，这对于训练和评估场景都很适用。其内部使用 JAX 的分片机制确保数据在分布式环境中正确分布
第三阶段：接口适配器实现
最后，函数通过定义嵌套类 `DataLoaderImpl` 来适配 `DataLoader` 协议接口。这个类封装了前面创建的 `TorchDataLoader` 实例，并提供了两个关键方法：
1. `data_config()` 返回数据配置信息，便于训练代码访问数据处理的元信息

2. `__iter__()` 生成器方法对数据批次进行最后的格式转换：
将字典格式的观察数据转换为结构化的 `Observation` 对象（通过 `Observation.from_dict`）提取动作数据
以元组形式 `(observation, actions)` 返回每个批次

这种设计实现了关注点分离，使数据加载、转换和格式适配各自独立，同时又协同工作，为模型训练提供了一个干净的数据流接口。函数还处理了多进程环境、数据分片和内存效率等复杂问题，这些都是大规模机器学习训练中的关键挑战

3.3 优化器系统 (optimizer.py)

定义了多种学习率调度策略：

`CosineDecaySchedule`：余弦衰减学习率
`RsqrtDecaySchedule`：反平方根衰减学习率

实现了常用优化器配置：

`AdamW`：带有权重衰减的 Adam 优化器
`SGD`：随机梯度下降优化器

通过 `create_optimizer` 函数统一创建优化器实例

3.4 检查点系统 (checkpoints.py)

负责模型状态的保存和恢复，比如管理训练状态的序列化，包括：

模型参数
优化器状态
EMA 参数（如果使用）

且使用 Orbax 库实现高效的检查点存储

模型初始化流程	训练步骤流程	与 models 模块的交互	检查点管理流程
init_train_state(config, rng, mesh) ├── 创建模型：model = config.model.create(rng) ├── 加载权重：partial_params = config.weight_loader.load(params) ├── 设置冻结参数：params = state_map(params, config.freeze_filter, ...) ├── 创建优化器：tx = create_optimizer(config.optimizer, config.lr_schedule) └── 返回 TrainState	train_step(config, rng, state, batch) ├── 计算梯度：loss, grads = value_and_grad(model.***pute_loss)() ├── 更新参数：updates, new_opt_state = state.tx.update(grads, state.opt_state, params) ├── 应用更新：new_params = optax.apply_updates(params, updates) ├── 更新 EMA 参数（如果配置） └── 返回 new_state, info	- 训练系统加载模型定义 (`BaseModel`) - 处理模型参数的保存和加载 - 调用模型的 `**pute_loss` 方法计算损失——详见上文的「1.2.4.3 损失函数 `**pute_loss`」	save_state(checkpoint_manager, state, data_loader, step) ├── _split_params(state) → 分离训练状态和推理参数 ├── 保存归一化统计数据到 assets 目录 └── checkpoint_manager.save() → 保存检查点 restore_state(checkpoint_manager, state, data_loader) ├── checkpoint_manager.restore() → 恢复检查点 └── _merge_params() → 合并恢复的参数

// 待更

3.5 模型分片系统(sharding.py)：含FSDP的实现

实现分布式训练时的模型参数分片

提供 `fsdp_sharding` 函数用于全参数数据并行(FSDP)的实现
基于 JAX 的分片机制，优化大规模模型的训练性能
通过 `activation_sharding_constraint` 处理激活值的分片

3.6 权重加载系统 (weight_loaders.py)

定义了 `WeightLoader` 协议，用于加载预训练权重，且实现了多种加载策略：

`NoOpWeightLoader`：不加载权重（用于从头训练）
`CheckpointWeightLoader`：从检查点加载完整权重
`PaliGemmaWeightLoader`：从官方 PaliGemma 检查点加载权重

另，还支持权重合并功能，可以部分加载权重（如 LoRA 微调）

3.7 辅助工具(utils.py)

定义了 `TrainState` 数据类，封装了训练过程的状态

提供日志记录和调试功能
实现了 PyTree 转换和可视化功能

// 待更

第四部分模型的训练与部署：基于客户端-服务器C/S架构——openpi-Client/Scripts

packages/openpi-client，是一个独立的客户端库openpi-client 库，主要负责：

提供与策略服务器通信的接口：使用 WebSocketClientPolicy 连接服务器
处理观察数据(图像、状态等)的发送，和动作数据的接收
管理客户端运行时环境
被各种机器人平台(如 ALOHA、DROID)使用来与服务器交互

scripts这个模块提供了服务器端的各种工具和脚本，主要包括：

策略服务相关——serve_policy.py：启动策略服务器，处理来自客户端的请求
训练相关——train.py: 模型训练的入口点
数据处理——***pute_norm_stats.py: 计算数据归一化统计信息
部署相关：提供 Docker 相关的配置和安装脚本

总的来说，这是一个典型的分布式系统设计：packages/openpi-client 提供轻量级的客户端接口，而 scripts/ 则提供服务器端的功能实现，两者通过 WebSocket 协议进行通信，形成了一个完整的策略部署和执行系统

所谓客户端-服务器架构——Client-server model，也称C/S架构、主从zòng式架构，是一种将客户端与服务器分割开来的分布式架构。每一个客户端软件的实例都可以向一个服务器或应用程序服务器发出请求。有很多不同类型的服务器，例如文件服务器、游戏服务器等

客户端的特征：

主动的角色（主）

发送请求

等待直到收到响应

服务端的特征：

被动的角色（从）

等待来自客户端的请求

处理请求并传回结果

4.1 packages/openpi-client：帮真机或Sim与策略服务器进行通信和交互

该模块的目录结构如下

这个客户端包的设计非常模块化，具有良好的扩展性，主要用于：

连接到 OpenPI 服务器
处理观察数据和动作序列
管理机器人或仿真环境的运行
提供事件监控和记录功能

它的设计允许在不同的机器人平台上灵活部署，支持实时控制和异步通信，是 OpenPI 项目中连接模型服务器和实际机器人执行系统的重要桥梁

4.1.1 核心接口层

`BasePolicy`: 定义策略接口
`Environment`: 定义环境接口
`Agent`: 定义代理接口

4.1.2 通信层WebsocketClientPolicy

`WebsocketClientPolicy`: 实现与服务器的 WebSocket 通信
`msgpack_numpy`: 处理数据序列化

4.1.3 数据处理层

`ActionChunkBroker`: 处理动作序列的分块和缓存
`image_tools`: 提供图像处理和优化功能

4.1.4 运行时系统层

`Runtime`: 核心运行时系统
`Subscriber`: 事件订阅系统
`agents`: 具体代理实现

4.1.5 工具支持

图像处理工具
数据类型转换
网络通信优化

4.2 scripts(策略服务器)：包含数据处理、模型训练、模型推理的多个脚本

根据下图

可知，scripts 目录包含多个 Python 脚本，这些脚本用于数据处理、模型训练和服务部署等任务，每个脚本通常对应一个特定的功能或任务

__init__.py
***pute_norm_stats.py: 计算数据的归一化统计信息
serve_policy.py：启动策略服务，提供模型推理接口
总之，serve_policy.py 是 openpi 中的策略推理服务端脚本，作用为：启动一个 WebSocket 服务器，加载预训练策略模型，等待外部请求（如来自 main.py 的控制程序），然后执行动作推理并返回结果
说白了，将一个 Pi0 策略模型部署为网络服务（WebSocket API），供机器人主控进程远程调用
train_test.py: 训练和测试模型
train.py: 训练模型

4.2.1 init.py

4.2.2 ***pute_norm_stats.py：计算数据的归一化统计信息

4.2.3(上) serve_policy.py：启动策略服务，用于模型推理——且支持定义特定任务的文本指令prompt

在这个代码片段中，首先导入了一些必要的模块和库，包括 `policy`、`policy_config`、`websocket_policy_server` 和 `config`，这些模块来自 `openpi` 项目

from openpi.policies import policy as _policy       # 导入 openpi.policies.policy 模块并重命名为 _policy
from openpi.policies import policy_config as _policy_config  # 导入 openpi.policies.policy_config 模块并重命名为 _policy_config
from openpi.serving import websocket_policy_server  # 导入 openpi.serving.websocket_policy_server 模块
from openpi.training import config as _config       # 导入 openpi.training.config 模块并重命名为 _config

接下来定义了一个枚举类 `EnvMode`，它表示支持的环境类型，包括 `ALOHA`、`ALOHA_SIM`、`DROID` 和 `LIBERO`

class EnvMode(enum.Enum):
    """支持的环境。"""
    ALOHA = "aloha"              # ALOHA 环境
    ALOHA_SIM = "aloha_sim"      # ALOHA 模拟环境
    DROID = "droid"              # DROID 环境
    LIBERO = "libero"            # LIBERO 环境

然后定义了几个数据类
`Checkpoint` 类用于从训练好的检查点加载策略，包含两个字段：`config`（训练配置名称）和 `dir`（检查点目录）
`Default` 类表示使用默认策略
`Args` 类定义了脚本的参数，包括环境类型、默认prompt、端口、是否记录策略行为以及如何加载策略

@dataclasses.dataclass
class Args:
    """Arguments for the serve_policy script."""

    # Environment to serve the policy for. This is only used when serving default policies.
    env: EnvMode = EnvMode.ALOHA_SIM

    # If provided, will be used in case the "prompt" key is not present in the data, or if the model doesn't have a default
    # prompt.
    default_prompt: str | None = None

    # Port to serve the policy on.
    port: int = 8000
    # Record the policy's behavior for debugging.
    record: bool = False

    # Specifies how to load the policy. If not provided, the default policy for the environment will be used.
    policy: Checkpoint | Default = dataclasses.field(default_factory=Default)

相当于如果你想定义你的特定任务指令prompt，则可以修改上面代码中的default_prompt

接下来定义了一个字典 `DEFAULT_CHECKPOINT`，它为每个环境类型指定了默认的检查点配置

# 每个环境应使用的默认检查点
DEFAULT_CHECKPOINT: dict[EnvMode, Checkpoint] = {
    EnvMode.ALOHA: Checkpoint(
        config="pi0_aloha",
        dir="s3://openpi-assets/checkpoints/pi0_base",
    ),
    EnvMode.ALOHA_SIM: Checkpoint(
        config="pi0_aloha_sim",
        dir="s3://openpi-assets/checkpoints/pi0_aloha_sim",
    ),
    EnvMode.DROID: Checkpoint(
        config="pi0_fast_droid",
        dir="s3://openpi-assets/checkpoints/pi0_fast_droid",
    ),
    EnvMode.LIBERO: Checkpoint(
        config="pi0_fast_libero",
        dir="s3://openpi-assets/checkpoints/pi0_fast_libero",
    ),
}

加载策略模型
`create_default_policy` 函数根据环境类型创建默认策略，如果环境类型不支持，则抛出异常

def create_default_policy(env: EnvMode, *, default_prompt: str | None = None) -> _policy.Policy:
    """为给定环境创建默认策略 """
    if checkpoint := DEFAULT_CHECKPOINT.get(env):              # 获取环境对应的默认检查点
        return _policy_config.create_trained_policy(
            _config.get_config(checkpoint.config), checkpoint.dir, default_prompt=default_prompt
        )  # 创建训练好的策略
    raise ValueError(f"Unsupported environment mode: {env}")   # 如果环境不支持，抛出异常

`create_policy` 函数根据传入的参数创建策略，如果参数中指定了检查点，则从检查点加载策略，否则使用默认策略

def create_policy(args: Args) -> _policy.Policy:
    """根据给定的参数创建策略 """
    match args.policy:          # 匹配策略类型
        case Checkpoint():      # 如果是 Checkpoint 类型
            return _policy_config.create_trained_policy(
                _config.get_config(args.policy.config), args.policy.dir, default_prompt=args.default_prompt
            )      # 创建训练好的策略
        case Default():          # 如果是 Default 类型
            return create_default_policy(args.env, default_prompt=args.default_prompt)      # 创建默认策略

启动推理服务
`main` 函数是脚本的入口点，它首先调用 `create_policy` 函数创建策略，然后记录策略的元数据

def main(args: Args) -> None:
    policy = create_policy(args)           # 创建策略
    policy_metadata = policy.metadata      # 获取策略的元数据

如果参数中指定了记录策略行为，则使用 `PolicyRecorder` 包装策略

    # 记录策略的行为
    if args.record:
        # 使用 PolicyRecorder 记录策略行为
        policy = _policy.PolicyRecorder(policy, "policy_records")

接着获取主机名和本地 IP 地址

    hostname = socket.gethostname()              # 获取主机名
    local_ip = socket.gethostbyname(hostname)    # 获取本地 IP 地址
    logging.info("Creating server (host: %s, ip: %s)", hostname, local_ip)  # 记录服务器创建信息

并创建一个 WebSocket 服务器来提供策略服务，最后调用 `serve_forever` 方法启动服务器

    server = websocket_policy_server.WebsocketPolicyServer(
        policy=policy,
        host="0.0.0.0",
        port=args.port,
        metadata=policy_metadata,
    )  # 创建 WebSocket 策略服务器
    server.serve_forever()      # 启动服务器，永远运行

在脚本的最后，使用 `logging` 模块配置日志记录，并调用 `main` 函数启动脚本，参数通过 `tyro.cli` 解析

更多还可以看下姚博士所写的这篇文章：openpi π₀ 项目部署运行逻辑（三）——策略推理服务器 serve_policy.py

4.2.3(下) 人类下达的任务指令prompt是如何在整个代码库中流转的

有一朋友在我建的「七月具身：π0复现微调交流群」里提问，为何不论设置怎样的指令prompt，机器人都执行同一套动作「后来，在他们使用多任务数据集训练后，π0可以实现prompt跟随，之前不能的原因是因为评估时机器人使用了和训练时的不同预备位姿」

对此，我特意梳理了下自定义的文本指令prompt在整个π0官方库中的数据流转——花了我一两个小时的时间，^_^

4.2.3.1 分别启动WebSocket服务器、WebSocket客户端并互联

第一阶段，设定prompt，随后分别启动WebSocket服务器、WebSocket客户端并互联

在上面介绍的这里设定prompt

class Args:
    """Arguments for the serve_policy script."""

    # Environment to serve the policy for. This is only used when serving default policies.
    env: EnvMode = EnvMode.ALOHA_SIM

    # If provided, will be used in case the "prompt" key is not present in the data, or if the model doesn't have a default
    # prompt.
    default_prompt: str | None = None

首先，启动策略服务器scripts/serve_policy.py，在这个策略服务器的代码文件中，main函数中

第一，执行scripts/serve_policy.py中的create_policy
```
def main(args: Args) -> None:
    policy = create_policy(args)
    policy_metadata = policy.metadata
```
而create_policy中，要么调用create_trained_policy，要么调用create_default_policy

比如，如果最终选择的是ALOHA的策略，则examples/aloha_real/main.py中的main函数会调用AlohaRealEnvironment类

而AlohaRealEnvironment被定义在examples/aloha_real/env.py中的，基于AlohaRealEnvironment的定义，可以看出来其通过其中的『AlohaRealEnvironment-__init__』函数设定环境的初始化「注意，这个AlohaRealEnvironment类中还定义了get_observation，下文会介绍」
第二，再执行scripts/serve_policy.py中的
```
    policy_metadata = policy.metadata
```
策略对象的metadata属性会包含default_prompt，且其在policy_metadata = policy.metadata时被提取出来
第三，再在scripts/serve_policy.py中，唤起并初始化WebSocket服务器而上面这个WebsocketPolicyServer，被定义在src/openpi/serving/websocket_policy_server.py 于此，上面(scripts/serve_policy.py中的)的policy_metadata传递给它(openpi/serving中的WebsocketPolicyServer)，存储在服务器中的self._metadata
第四，通过scripts/serve_policy.py中serve_forever的启动WebSocket服务器上面那个serve_forever被定义在src/openpi/serving/websocket_policy_server.py中

其次，启动WebSocket客户端：WebsocketClientPolicy

packages/openpi-client/src/openpi_client/websocket_client_policy.py中的WebsocketClientPolicy被初始化时，调用_wait_for_server 连接WebSocket服务端
服务端WebsocketPolicyServer的_handler方法在接受连接后，立即发送self._metadata——await websocket.send(packer.pack(self._metadata)) 给客户端客户端_wait_for_server的接收到这个元数据之后，便存储在_server_metadata中

4.2.3.2 客户端发送推理请求、服务端处理推理请求

第二阶段，客户端发送推理请求、服务端处理推理请求

推理请求：客户端向服务端发送全部数据
一方面，环境(examples/aloha_real/env.py)通过get_observation获取观察数据二方面，客户端的infer(obs)被调用，它将包含状态和图像的obs字典打包发给服务器
服务器处理推理请求
首先，服务器的_handler接收到obs字典，然后调用action = self._policy.infer(obs) 其次，策略执行推理
策略内部处理 (policies 下的具体策略文件)——策略的 `infer` 方法被调用以获取prompt
由于传入的 `obs` 字典没有 `"prompt"` 键，策略会查找并使用它在步骤 1 中存储的 `self._default_prompt`，类似prompt_to_use = obs.get("prompt", self._default_prompt)`。这里 `prompt_to_use` 会被赋值为自定义的指令字符串

4.2.3.3 模型获得全部输入数据，生成动作序列

第三阶段，模型获得全部输入数据，生成动作序列

获取到的prompt被传递给分词器Tokennizer，其将文本指令转换为token ID序列
这些token ID序列与图像数据、状态数据一起被输入到π0中
π0处理这些输入，生成预测的动作序列

// 待更

4.2.4 train_test.py：训练和测试模型

4.2.5 train.py：训练模型——损失函数计算、梯度下降、参数更新

这段代码是一个基于JAX的分布式训练脚本，集成了模型初始化、训练循环、日志记录、实验跟踪和检查点管理等功能。以下是对代码的模块化解读：

一开始先后涉及日志初始化 (`init_logging`)、Weights & Biases 初始化 (`init_wandb`)、权重加载与验证 (`_load_weights_and_validate`)

之后是训练状态初始化 (`init_train_state`)

创建优化器（`tx`）和模型实例
合并预训练参数（若有）到模型状态
参数类型转换（如冻结参数转`bfloat16`）
定义分布式分片策略（`fsdp_sharding`）
返回值：包含模型参数、优化器状态、EMA参数的`TrainState`对象及分片信息

再之后，是单步训练`train_step`

前向计算：模型计算损失(启用训练模式)，loss_fn中调用的损失函数来自——1.2.4.3 损失函数***pute_loss：训练模型去噪的准确率(含训练数据集的来源介绍)

def train_step(
    config: _config.TrainConfig,
    rng: at.KeyArrayLike,
    state: training_utils.TrainState,
    batch: tuple[_model.Observation, _model.Actions],
) -> tuple[training_utils.TrainState, dict[str, at.Array]]:
    """执行单个训练步骤"""
    # 合并模型定义和参数
    model = nnx.merge(state.model_def, state.params)
    model.train()  # 设置模型为训练模式

    @at.typecheck
    def loss_fn(
        model: _model.BaseModel, rng: at.KeyArrayLike, observation: _model.Observation, actions: _model.Actions
    ):
        """损失函数"""
        # 计算每个数据项的损失
        chunked_loss = model.***pute_loss(rng, observation, actions, train=True)
        return jnp.mean(chunked_loss)  # 返回平均损失

随机数生成

    # 根据当前步数折叠随机数种子，确保每步使用不同随机数
    train_rng = jax.random.fold_in(rng, state.step)

    # 解包批次数据
    observation, actions = batch

梯度计算：通过`nnx.value_and_grad`获取梯度，仅更新可训练参数

    # 过滤出可训练参数
    diff_state = nnx.DiffState(0, config.trainable_filter)

    # 计算损失和梯度
    loss, grads = nnx.value_and_grad(loss_fn, argnums=diff_state)(model, train_rng, observation, actions)

参数更新：应用优化器更新，合并新参数到模型

    # 过滤出可训练参数
    params = state.params.filter(config.trainable_filter)

    # 使用优化器更新参数
    updates, new_opt_state = state.tx.update(grads, state.opt_state, params)
    new_params = optax.apply_updates(params, updates)

    # 更新模型参数并返回新的完整状态
    nnx.update(model, new_params)
    new_params = nnx.state(model)

EMA维护：指数平滑更新关键参数

    # 创建新的训练状态，更新步数、参数和优化器状态
    new_state = dataclasses.replace(state, step=state.step + 1, params=new_params, opt_state=new_opt_state)
    if state.ema_decay is not None:
        # 如果使用EMA，更新EMA参数
        new_state = dataclasses.replace(
            new_state,
            ema_params=jax.tree.map(
                lambda old, new: state.ema_decay * old + (1 - state.ema_decay) * new, state.ema_params, new_params
            ),
        )

    # 过滤出核心参数（不包括偏置、缩放等）
    kernel_params = nnx.state(
        model,
        nnx.All(
            nnx.Param,  # 必须是参数
            nnx.Not(nnx_utils.PathRegex(".*/(bias|scale|pos_embedding|input_embedding)")),  # 排除特定名称
            lambda _, x: x.value.ndim > 1,  # 必须是多维的
        ),
    )

指标收集：损失、梯度范数、参数范数（过滤非核参数）

    # 收集训练信息
    info = {
        "loss": loss,  # 损失值
        "grad_norm": optax.global_norm(grads),              # 梯度范数
        "param_norm": optax.global_norm(kernel_params),     # 参数范数
    }
    return new_state, info

最后是主函数`main`

环境初始化：日志、JAX配置、随机种子、设备分片
数据准备：分布式数据加载器，分片策略（数据并行）
状态恢复：检查点管理器处理恢复逻辑。
训练循环：
JIT编译的分布式训练步骤（`ptrain_step`）
定期日志记录（控制台 + W&B）
检查点保存（间隔保存 + 最终保存）
清理：等待异步保存操作完成

// 待更

4.2.6 scripts/docker

好的，下面是对 `openpi-main/scripts/docker` 目录的详细分析。这个目录通包含与 Docker 相关的脚本和配置文件，用于构建和管理 Docker 容器，具体而言，包含以下文件和子目录：

主要文件和功能如下所示

docker/***pose.yml
docker/install_docker_ubuntu22.sh
docker/install_nvidia_container_toolkit.sh
docker/serve_policy.Dockerfile

// 待更

第五部分 examples ：各种机器人平台及策略客户端的示例实现

根据π0对应examples模块的结构

其涉及以下模块

aloha_real/：真实机器人ALOHA的示例
aloha_sim/：ALOHA模拟器的示例
droid/：DROID机器人的示例
libero/：LIBERO基准测试的示例
simple_client/：简单客户端的示例
ur5/：UR5机器人的示例
inference.ipynb：推理示例的Jupyter Notebook
policy_records.ipynb：策略记录示例的Jupyter Notebook

5.1 aloha_real

`aloha_real` 模块是OpenPI项目中用于控制真实ALOHA双臂机器人的完整实现。它提供了从OpenPI策略模型到真实机器人硬件的完整控制链路

5.1.1 核心架构

主控制流程 (main.py)
作为系统入口点，协调各个组件
其关键组件包括
`WebsocketClientPolicy: 通过WebSocket连接到OpenPI策略服务器
ActionChunkBroker: 处理动作序列，支持25步动作预测
Runtime: 运行时环境，以50Hz频率执行控制循环
PolicyAgent: 策略代理，桥接策略和环境
环境接口 (env.py 和 real_env.py)
`AlohaRealEnvironment` (高级封装)：提供标准化的环境接口、处理图像预处理和尺寸调整 (224x224)、将图像格式从 HWC 转换为 CHW

`RealEnv` (底层硬件接口)
双臂控制: 左右两个Interbotix vx300s机械臂
动作空间 (14维)：
```
  [left_arm_qpos(6), left_gripper(1), right_arm_qpos(6), right_gripper(1)]
```
观察空间:
`qpos`: 关节位置 (14维)
`qvel`: 关节速度 (14维)
`images`: 4个摄像头视角
- `cam_high`: 俯视视角
- `cam_low`: 平视视角
- `cam_left_wrist`: 左手腕视角
- `cam_right_wrist`: 右手腕视角
硬件常量定义 (constants.py)
关节名称: `["waist", "shoulder", "elbow", "forearm_roll", "wrist_angle", "wrist_rotate"]`
夹爪位置限制: 开合状态的物理限位
标准化函数: 将夹爪位置映射到[0,1]区间
默认复位姿态: `[0, -0.96, 1.16, 0, -0.3, 0]`
数据转换工具
convert_aloha_data_to_lerobot.py
将ALOHA原生数据格式转换为LeRobot标准格式
支持训练数据的预处理和标准化

robot_utils.py
包含机器人设置和数据记录工具
Recorder: 记录关节状态数据
ImageRecorder: 记录摄像头图像数据

5.1.2 系统工作流程与部署方式

初始化阶段
启动ROS节点 → 初始化双臂机器人 → 连接摄像头 → 建立WebSocket连接
运行时循环(50Hz)
获取观察(图像+状态) → 发送到策略服务器 → 接收动作序列 → 执行动作 → 更新状态
动作执行
策略预测25步动作序列
`ActionChunkBroker`管理动作缓冲和执行
每步动作包含14维关节目标位置

至于部署方式有以下两种

一个是Docker部署，则直接安装 Docker 并运行

export SERVER_ARGS="--env ALOHA --default_prompt='take the toast out of the toaster'"
docker ***pose -f examples/aloha_real/***pose.yml up --build

一个是本地部署，其需要启动3个终端

[终端1] 机器人客户端 ←→ WebSocket ←→ [终端3] 策略服务器/serve_policy.py
    ↓                                      ↑
[终端2] ROS硬件层                      OpenPI模型推理

机器人控制客户端，相当于WebSocket客户端
该客户端从机器人硬件获取观察数据(图像 + 状态)，然后通过WebSocket发送观察数据到策略服务器
之后，接收策略服务器返回的动作指令，将动作指令发送给机器人执行
————————————————
具体而言，初始化虚拟环境并运行机器人控制主程序
```
# Create virtual environment
uv venv --python 3.10 examples/aloha_real/.venv
source examples/aloha_real/.venv/bin/activate
uv pip sync examples/aloha_real/requirements.txt
uv pip install -e packages/openpi-client
 
# Run the robot
python -m examples.aloha_real.main
```
以上代码分别对应
1 创建Python 3.10虚拟环境
使用 uv 工具创建一个 Python 虚拟环境，路径为 examples/aloha_real/.venv
uv 是一个替代 venv + pip 的高性能依赖管理工具
2 激活虚拟环境
进入刚才创建的虚拟环境，使之后的 Python 执行与 pip 安装都仅作用于该环境
3 安装所需依赖
安装 requirements.txt 中精确指定的依赖版本
sync 是比 install -r 更稳定的方式，确保包版本锁定且无冗余
4 安装 openpi-client 为本地开发模式
使用 -e（editable）方式安装本地 openpi-client 包，允许实时修改代码而无需重装
5 启动主程序控制机器人（主线程）
启动 Pi0 控制机器人动作的主循环，包括摄像头读取、电机控制、感知更新等
此模块会连接 ROS 节点并与推理服务通信
ROS节点服务
其作用为：启动机器人硬件驱动、启动摄像头节点、提供底层硬件接口
————————————————
具体而言，启动ROS驱动
```
roslaunch aloha ros_nodes.launch
```
使用 ROS 启动 aloha 平台硬件驱动，包括：
控制机械臂的 Dynamixel 电机驱动
相机接口节点
ROS 的 topic 广播、TF 树等
这一步是让机器人硬件对接 ROS 网络层，确保后续主控程序可调用硬件资源
PS：如姚博士所说，此处 ROS 配置根据项目子模块配置，以及 ROS 系统主要针对 ALOHA 一类的舵机机器人
openpi策略服务器，相当于WebSocket服务器
其作用为：加载训练好的openpi模型，监听WebSocket连接，以及接收观察数据运行策略推理，从而最终返回动作序列
————————————————
具体而言，运行策略推理服务器
```
uv run scripts/serve_policy.py --env ALOHA --default_prompt='take the toast out of the toaster'
```
上述代码相当于启动 serve_policy.py 推理服务：
加载一个策略（如 pi0）和预训练权重
等待来自主控制进程的请求（语言提示 + 视觉输入）
返回动作序列给控制主进程

综上，三进程间的协同流程可以总结为：

[ROS系统（终端2）] <== 硬件数据 ==> [主控进程 main.py（终端1）] <== 请求 ==> [推理服务 serve_policy.py（终端3）]

即三个终端分别主要负责：

启动虚拟环境 + 控制主逻辑，控制主程序需要同步感知与动作控制
启动 ROS 节点驱动机器人硬件，ROS 启动通常是独立的进程
启动语言策略模型的推理服务，推理服务需常驻监听 socket 请求

// 待更

RabbitMQ

Wed, 26 Nov 2025 17:21:57 +0800

在消息队列（MQ）中，确保消息成功传递是一个关键问题。消息传递的过程通常包括以下几个阶段：publisher（生产者） -> exchange（交换机） -> queue（队列） -> consumer（消费者）。为了确保消息在每个阶段都能成功传递，我们需要采取一系列措施来保证消息的可靠性。

生产者的可靠性

重试机制

当生产者与交换机（或队列，如果没有交换机）之间的连接不稳定时，生产者发送的消息可能会在传输过程中丢失。在这种情况下，生产者需要等待一段时间以获取响应。如果未收到响应，生产者应尝试重新发送消息。重试次数应有限制，以防止因持续重试而占用过多资源。此外，重试之间应有一定的间隔时间，以避免频繁重试导致资源浪费。

由于发送消息时会占用通道，其他业务操作可能会被阻塞，直到消息发送完成（无论成功或失败）。因此，对发送消息的重试机制进行限制是必要的，以防止因连接问题导致资源被长时间占用。

以下是一个在Spring Boot中配置生产者重试机制的示例：

spring:
  rabbitmq:
    connection-timeout: 1s  # 连接超时时间
    template:
      retry:
        enabled: true  # 开启重试机制
        initial-interval: 1000ms  # 初始重试间隔时间
        multiplier: 1  # 重试间隔时间的倍数
        max-attempts: 5  # 最大重试次数

在这个配置中：

connection-timeout 设置了连接超时的时间。
enabled: true 开启了重试机制。
initial-interval: 1000ms 设置了在连接失败后，首次重试前的等待时间。
multiplier: 1 设置了每次重试后等待时间的倍数（在此例中，等待时间保持不变）。
max-attempts: 5 设置了最大重试次数，超过该次数后将不再重试。

通过这种配置，生产者在发送消息失败后会自动进行重试，直到达到最大重试次数或消息成功发送。这种机制有效地提高了消息传递的可靠性，同时避免了因持续重试而导致的资源浪费

根据您提供的信息，我们可以分析MQ连接失败时的重试行为。以下是详细的分析：

连接超时时间：设置为1秒。这意味着如果MQ在1秒内未能成功连接，连接尝试将被视为失败。
初始重试间隔：设置为1秒。在第一次连接失败后，系统会等待1秒再进行下一次连接尝试。
等待时间倍数：设置为1。这意味着每次重试的等待时间保持不变。因此，每次重试的间隔时间为1秒（等待时间） + 1秒（连接超时时间） = 2秒。
最大重试次数：设置为5次。系统会在达到最大重试次数后停止尝试连接。

根据这些设置，系统在连接失败后的行为如下：

第一次连接失败后，等待1秒，然后进行第二次连接尝试。
第二次连接失败后，再次等待1秒，然后进行第三次连接尝试。
第三次连接失败后，系统将停止尝试连接。

如果multiplier设置为2，重试行为将有所不同：

第一次连接失败后，等待1秒，然后进行第二次连接尝试。
第二次连接失败后，等待时间翻倍为2秒，然后进行第三次连接尝试。
第三次连接失败后，等待时间再次翻倍为4秒，然后系统将停止尝试连接。

这种配置确保了系统在连接失败时能够进行有限次数的重试，同时通过调整等待时间倍数来控制重试的频率，以避免过度占用资源。

确认机制

在正常情况下，消息传递到MQ后不会发生丢失，但我们仍需对消息丢失有所防备。为了及时发现消息丢失，MQ通常使用Publisher Confirm和Publisher Return两种机制来进行预警。

消息丢失的可能情况

MQ内部故障：MQ服务器内部出现问题，导致消息丢失。
交换机或队列不存在：消息无法找到目标交换机或队列。
RoutingKey不匹配：消息的路由键（RoutingKey）没有匹配的队列。
其他异常情况：例如网络故障、消息过期（TTL）等。

消息传递的几种状态

路由失败，但投递成功：
- 消息传递到MQ后，MQ服务器成功接收并存储了消息（投递成功）。
- 但由于路由失败（例如RoutingKey不匹配或目标队列不存在），消息无法被正确路由到下一个节点（例如队列）。
- MQ服务器会返回ACK确认投递成功，同时通过Publisher Return机制返回路由失败的异常信息。
临时消息投递成功：
- 临时消息传递到MQ后，被存储在内存中。
- MQ服务器返回ACK，表示消息已成功投递。
持久化消息投递成功：
- 持久化消息传递到MQ后，被持久化存储到磁盘。
- MQ服务器返回ACK，表示消息已成功投递。
投递失败：
- 如果消息未能成功传递到MQ（例如网络故障或MQ服务器不可用），MQ服务器会返回NACK，表示投递失败。

Publisher Confirm 和 Publisher Return 的作用

Publisher Confirm：
- 通过返回ACK或NACK，告知生产者消息是否成功投递到MQ服务器。
- ACK表示消息已成功投递（无论是临时消息还是持久化消息）。
- NACK表示消息投递失败。
Publisher Return：
- 当消息成功投递到MQ服务器但路由失败时，通过Publisher Return返回异常信息。
- 帮助生产者及时发现消息无法被正确路由的问题。

总结

投递成功：消息成功到达MQ服务器并被存储（临时消息存储在内存中，持久化消息存储在磁盘中），MQ返回ACK。
路由失败：消息成功投递到MQ服务器，但无法被正确路由到目标队列，MQ通过Publisher Return返回异常信息。
投递失败：消息未能成功传递到MQ服务器，MQ返回NACK。

通过Publisher Confirm和Publisher Return机制，生产者可以及时了解消息的投递状态，从而有效预防和发现消息丢失问题。

spring:
rabbitmq:
publisher-confirm-type: correlated
publisher-returns: true

默认情况下, Publisher Confirm和Publisher Return两个机制是关闭的. publisher-confirm-type有三钟状态

none: 默认关闭, 就是这个状态

simple: 同步进行, 需要等到回复状态之后才会继续业务

correlated: 异步进行, 在等待回复状态的同时, 业务可以继续进行处理

MQ的可靠性

数据持久化

确保消息队列（MQ）可靠性的关键措施

在分布式系统中，消息队列（MQ）是异步通信的核心组件。然而，即使在正常情况下，消息在到达MQ后仍有可能丢失。因此，确保MQ的可靠性至关重要。本文将介绍几种常见的MQ可靠性方案。

1. 消息丢失的原因

MQ通常将消息存储在内存中进行处理和传递，这种方式虽然高效，但在MQ服务重启或崩溃时，内存中的消息会丢失。为了解决这一问题，我们需要采取以下措施来增强MQ的可靠性。

2. MQ可靠性方案

2.1 交换机的持久化

交换机（Exchange）是MQ中路由消息的关键组件。通过将交换机设置为持久化，可以确保在MQ重启后，交换机的配置和元数据不会丢失，从而保证消息能够继续被正确路由。

2.2 队列的持久化

队列（Queue）是消息的存储载体。将队列设置为持久化后，即使MQ服务重启，队列中的消息也不会丢失。持久化队列会将消息存储到磁盘中，而不是仅仅依赖内存。

2.3 消息的持久化

除了交换机和队列的持久化，消息本身也需要进行持久化处理。持久化的消息在未被消费前会一直存储在磁盘中，只有在被成功消费后才会被删除。这种方式可以有效避免因MQ重启或崩溃导致的消息丢失。

3. 生产者确认机制

为了进一步增强MQ的可靠性，可以启用生产者确认机制（Publisher Confirms）。当消息被持久化存储到磁盘后，MQ会向生产者发送一个确认（ACK），告知消息已安全存储。这种机制可以确保消息不会在传输过程中丢失。

4. 批量持久化与异步处理

为了提高性能，MQ通常不会逐条持久化消息，而是采用批量持久化的方式。这种方式可以显著减少磁盘I/O操作，提升系统的整体效率。同时，推荐使用异步方式进行持久化，以避免阻塞消息的处理流程。

5. 总结

通过交换机的持久化、队列的持久化、消息的持久化以及生产者确认机制，可以显著提升MQ的可靠性。此外，批量持久化和异步处理能够在不牺牲可靠性的前提下，进一步提高系统的性能。在实际应用中，建议根据业务需求合理配置这些机制，以确保消息的可靠传递。

LazyQueue

消息存储在内存中的优势与挑战

RabbitMQ 默认将消息存储在内存中，因为内存的读写速度远高于硬盘，这可以显著提高消息处理的效率。然而，这种设计也带来了一些潜在的问题，尤其是在消息量激增的情况下。

1. 内存存储的优势

高效读写： 内存的访问速度比硬盘快得多，因此将消息存储在内存中可以大幅降低消息收发的延迟，提升系统性能。
低延迟： 对于实时性要求较高的场景，内存存储能够确保消息快速传递。

2. 内存存储的挑战

尽管内存存储有诸多优势，但在某些情况下，可能会面临以下问题：

2.1 生产者消息激增

当生产者的消息发送速率突然增加时，可能会导致消息在内存中大量堆积。

2.2 消费者处理能力不足

如果消费者的处理速度跟不上生产者的发送速度，消息会在内存中积压，占用大量内存资源。

2.3 内存限制与 PageOut

内存的容量是有限的，当消息积压超过内存的极限时，RabbitMQ 会将部分消息从内存转移到硬盘中，这个过程称为 PageOut。在 PageOut 过程中：

生产者可能会被拒绝发送消息（流控机制生效）。
消费者也无法消费消息，因为部分消息正在从内存转移到硬盘。
内存资源会被占用，影响系统的整体性能。

3. Lazy Queue 的引入

为了解决上述问题，RabbitMQ 在 3.12 版本 之后引入了 Lazy Queue 机制。Lazy Queue 的核心设计思想是：

消息优先存储到硬盘： 消息不会直接存储在内存中，而是批量写入硬盘。
懒加载机制： 只有当消费者需要消费消息时，才会将消息从硬盘加载到内存中。
支持海量消息存储： Lazy Queue 可以轻松处理百万级甚至更多的消息量，而不会对内存造成过大压力。

4. Lazy Queue 的优势

降低内存压力： 消息主要存储在硬盘中，内存占用大幅减少。
提高系统稳定性： 即使消息量激增，也不会因为内存不足而导致消息丢失或系统崩溃。
适合高吞吐量场景： 对于消息量大但实时性要求不高的场景，Lazy Queue 是一个理想的选择。

消费者的可靠性

当RabbitMq将消息传递给消费者, 依旧会存在之前的消息丢失, 比如消费者处理异常, 发送失败, 消费者宕机等情况

消费者确认机制

当RabbitMQ将消息传递给消费者时，仍然可能存在消息丢失的情况，例如消费者处理异常、发送失败、消费者宕机等。为了确保消息的可靠传递，RabbitMQ提供了消费者确认机制，允许消费者在处理消息后向RabbitMQ反馈消息的处理状态。

消费者确认机制

RabbitMQ的消费者确认机制主要分为以下三种状态：

ack（确认）：消息处理成功，RabbitMQ会将该消息从队列中删除。
nack（否定确认）：消息处理异常，RabbitMQ会将消息重新加载回队列，进行重试。
reject（拒绝）：消息处理异常，且消息无法被处理，RabbitMQ会直接删除该消息。reject通常用于消息类型不匹配或无法处理的场景。

Spring AMQP中的确认模式

Spring AMQP为开发者提供了三种确认模式，以简化消息确认的处理：

none模式：
- 当消息发送到消费者后，RabbitMQ会立即确认（ack）并删除消息，无论消费者是否成功处理。
- 适用于对消息丢失不敏感的场景。
manual模式：
- 开发者需要手动调用ack、nack或reject来确认消息的处理状态。
- 适用于需要精细控制消息确认的场景，但可能会造成业务代码的污染。
auto模式：
- Spring AMQP会自动根据消息处理的结果发送确认。
  - 如果业务正常处理，返回ack。
  - 如果发生业务异常，返回nack，消息会重新入队进行重试。
  - 如果发生消息校验或处理异常，返回reject，消息会被直接删除。
- 适用于大多数常见的业务场景。

配置示例

在Spring Boot中，可以通过以下配置来设置确认模式：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto # 可选值为 none, manual, auto

异常处理的最佳实践

在auto模式下，建议根据异常类型决定是返回nack还是reject。例如，对于可重试的业务异常（如网络抖动），可以返回nack；对于不可恢复的异常（如消息格式错误），可以返回reject。

失败重试机制

业务异常处理与重试机制

在消息队列中，如果不对业务异常处理进行合理的限制，每次业务发生异常时，消息会重新入队并进行重试。如果没有适当的控制机制，消息可能会不断重试，导致系统资源被空耗，甚至引发消息堆积、系统崩溃等问题。

在Spring AMQP中，可以通过配置RabbitMQ的监听器重试机制来避免这种情况。以下是一个典型的配置示例：

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true          # 开启重试机制
          max-attempts: 5         # 最大重试次数
          initial-interval: 1000  # 初始重试间隔时间（毫秒）
          multiplier: 1           # 重试间隔时间的倍数
          stateless: true         # 是否启用无状态重试

配置项说明：

enabled: 是否启用重试机制。
max-attempts: 最大重试次数。
initial-interval: 初始重试间隔时间。
multiplier: 重试间隔时间的倍数。
stateless: 是否启用无状态重试。设置为true时，表示每次重试都是独立的，不会保留前一次重试的状态。这对于一些涉及上下文代码或变量传递的业务场景非常重要。如果业务逻辑依赖于某些变量的初始状态，启用无状态重试可以确保每次重试时变量都恢复到初始值，避免前一次重试的结果影响后续业务。

无状态重试的重要性

在某些业务场景中，业务逻辑可能会依赖于某些变量的初始状态。例如：

int x = 5;  // 初始值为5
// 业务逻辑处理
x = 8;      // 业务处理后，x的值变为8

如果在下一次重试时，变量x仍然保留上一次处理后的值（即8），而不是恢复到初始值（5），这可能会导致业务逻辑出现错误。通过设置stateless: true，可以确保每次重试时，业务逻辑中的变量都会恢复到初始状态，从而避免这种问题。

失败处理策略

当消息在本地多次重试失败后, 超过重试次数的限制, 会被队列删除, 但是这对于消息可靠性要求高的业务并不友好, 所以提供了一个MessageRecovery接口, 这个接口较好的实现是RepublishMessageRecovery类, 它是把异常信息放到一个单独的队列中, 后续人工介入处理

业务幂等性

在执行某个业务操作时，无论该操作被执行一次还是多次，最终的业务结果都是一致的，这就是幂等性。幂等性在分布式系统和网络通信中尤为重要，因为它可以有效避免重复操作带来的问题。

举个例子：当你在购物时进行扣款操作，如果由于网络延迟或其他原因，扣款请求被重复发送了两次，而你被扣了两次钱，这显然是不可接受的。为了避免这种情况，我们需要确保扣款操作的幂等性。

唯一消息ID

在消息队列系统中，为了确保消息的幂等性（即多次处理同一消息不会产生重复的效果），通常会在发送消息时附带一个唯一ID。这个唯一ID可以是全局唯一标识符（UUID）或根据业务规则生成的唯一值。消费者在接收到消息后，首先会检查该ID是否已经存在于数据库中。如果该ID已经存在，说明该消息已经被处理过，消费者会直接跳过该消息，避免重复处理；如果该ID不存在，消费者则会处理该消息，并将该ID存储到数据库中，以确保后续重复消息不会被重复处理。

业务判断

除了使用唯一消息ID来确保消息的幂等性外，业务判断也是处理重复消息或请求的重要手段。业务判断是通过业务本身的逻辑来进行判断，确保即使消息ID不同，但业务内容相同的请求也不会被重复处理。例如，在订单系统中，可以通过订单号、用户ID等业务字段来判断是否已经处理过相同的请求。

通过结合唯一消息ID和业务判断，可以有效地确保消息队列系统中的消息处理是幂等的，从而避免重复处理带来的业务问题。

兜底方案

尽管MQ消息系统已经尽可能减少了消息丢失的可能性，但在实际应用中，消息丢失的情况仍然可能发生。因此，我们需要主动采取措施来确保业务结果的准确性。具体来说，可以通过定时任务来实现这一目标。定时任务会每隔一段时间对业务结果进行比对和检查，确保数据的完整性和一致性。

这种兜底方案的核心在于通过定期检查来弥补消息丢失可能带来的影响，从而保证业务的最终一致性。

消息幂等与性能优化

在高数据量的场景下，消息幂等性不仅要保证消息处理的正确性，还需要提升消息的存取速度和系统效率。以下是几种常见的优化方案：

集群部署：通过集群化部署，可以提高系统的稳定性和吞吐量，分散单节点的压力，从而提升整体性能。
分库分表：对于数据量巨大的场景，可以采用分库分表的方式，将数据分散到多个数据库或表中，减少单表的数据量，提升查询和写入效率。
数据生命周期管理：通过定时任务对数据进行归档、移动或删除，确保数据的时效性，避免无效数据占用存储资源，从而提升系统性能。

这些方法结合使用，可以在保证消息幂等性的同时，有效提升系统的处理能力和效率

延时消息

在网上售卖货物时，由于商品数量有限，当用户下单后，数据库会自动扣减商品库存。然而，如果用户未在规定时间内完成付款，这些商品就会被该用户占用，导致其他购物者无法购买。为了解决这一问题，我们需要对未付款的用户设置时间限制，要求他们在规定时间内完成付款，否则库存将被释放。

这种需要在一定时间后再执行的任务被称为延时任务。消息队列（MQ）提供了两种处理延时任务的解决方案：死信交换机和延时消息插件。

死信交换机

死信是指以下几种情况下的消息：

超时未被处理的消息：消息在队列中等待时间过长，未被消费者处理。
队列满员：当队列达到最大容量时，新消息无法进入，成为死信。
消费失败的消息：消费者在处理消息时返回nack或reject，并且将requeue设置为false，表示消息消费失败且不再重新入队。

当一个队列中存在死信消息时，可以通过配置dead-letter-exchange参数，将这些死信消息转发到一个特定的交换机，这个交换机被称为死信交换机。死信交换机会与一个或多个队列绑定，用于接收和处理这些死信消息。

延时消息插件

延时消息插件是一种简化消息延时处理的工具，相较于使用死信交换机的方案，它减少了实现延时消息所需的步骤。延时消息插件在基本的消息流程（生产者 -> 交换机 -> 队列 -> 消费者）中，通过将交换机设计为具备延时和暂时存储消息的能力，从而直接实现消息的延时投递。这种方式避免了传统死信交换机方案中需要额外设置死信队列、绑定死信交换机等复杂操作，简化了系统的设计和维护。

HTTP请求解剖图：从请求头到请求体，小白秒懂

Wed, 26 Nov 2025 17:21:54 +0800

当你刷网页、点外卖、传文件时，浏览器都在悄悄发送"网络快递"——这就是HTTP请求！它像精心包装的包裹，包含地址标签（请求头）和实际货物（请求体）。学会拆解这个包裹，你就掌握了网络通信的核心密码！

📦 一、HTTP请求就像快递包裹

想象你寄快递：

📝 请求头(Headers) = 快递单（写清收件地址、物品类型、特殊要求）
🎁 请求体(Body) = 包裹里的实际货物（表单数据、文件、JSON等）

📜 二、HTTP请求完整结构

一个标准的HTTP请求包含三部分：

示例：登录请求的原始数据

POST /login HTTP/1.1                   👉 请求行
Host: www.example.***
Content-Type: application/json         👉 请求头
User-Agent: Chrome/115.0

{"username":"tom","password":"123456"} 👉 请求体

📋 三、请求头：包裹的"快递单"

高频请求头大全（表格对比）

请求头字段	作用	常见值示例
Host	目标服务器地址	`www.baidu.***`
User-Agent	客户端身份标识	`Mozilla/5.0 (Windows NT 10.0)`
Content-Type	请求体的数据类型	`application/json`, `multipart/form-data`
Aept*	期望的响应格式	`text/html`, `image/*`
Authorization	身份验证凭证	`Bearer xxxxxx`
Cookie	客户端存储的数据	`sessionId=abc123`

💡 关键提示：Content-Type是请求体的"说明书"，服务器靠它解析数据格式！

📦 四、请求体：包裹的"实际货物"

根据Content-Type分三大类型：

1. 表单数据 (`application/x-www-form-urlencoded`)

网页表单提交的标准格式，如登录框

POST /login HTTP/1.1
Content-Type: application/x-www-form-urlencoded

username=tom&password=123456  👉 键值对用&连接

2. JSON数据 (`application/json`)

API接口主流格式，结构清晰

POST /api/users HTTP/1.1
Content-Type: application/json

{
  "name": "Lucy",
  "age": 25,
  "hobbies": ["coding","reading"]
}  👉 结构化数据

3. 文件上传 (`multipart/form-data`)

上传图片/视频等二进制文件

POST /upload HTTP/1.1
Content-Type: multipart/form-data; boundary=----boundary123

------boundary123
Content-Disposition: form-data; name="avatar"; filename="photo.jpg"
Content-Type: image/jpeg

<这里是图片的二进制数据...>
------boundary123--

🔍 五、如何查看HTTP请求？

浏览器开发者工具实战

按 F12 打开控制台 → ***work 标签
刷新页面 → 点击任意请求
查看 Headers 和 Request Payload

代码示例：Python发送POST请求

import requests

# 发送JSON请求体
headers = {"Content-Type": "application/json"}
data = {"key": "value"}
response = requests.post("https://api.example.***", json=data, headers=headers)

# 发送文件
files = {'file': open('report.xlsx', 'rb')}
requests.post("https://upload.***", files=files)

🚀 六、不同请求方法的身体差异

请求方法	是否携带请求体	典型场景
GET	❌ 无	获取网页、搜索查询
POST	✅ 有	提交表单、创建资源
PUT	✅ 有	更新整个资源
PATCH	✅ 有	更新资源部分字段
DELETE	⚠️ 通常无	删除资源

⚠️ 注意：GET请求的参数在URL中传输：
https://api.***/search?keyword=http&page=1

💎 七、核心总结

请求行 = 动作指令（GET/POST等）
请求头 = 元数据标签（Content-Type决定身体类型）
请求体 = 传输的实际数据（JSON/表单/文件）

✨ 掌握HTTP请求结构，你就能：

调试API接口更高效 🛠️

理解前后端数据交互 🔄

快速定位网络问题 🔍

📚 扩展阅读

HTTP/2协议详解
Postman接口测试工具实战

下次遇到API报错时，先检查请求头和请求体吧！ 👨‍💻
点赞 ▲ 收藏 ⭐ 关注 ➕ 素质三连走起~ ❤️

异构多活架构引领医疗信创：浙人医案例解析

Wed, 26 Nov 2025 17:21:52 +0800

作为浙江省卫健委直属，省内规模最大、实力最强的综合性三甲医院，浙江省人民医院（下称“浙人医”）庞大的服务体量与业务规模，使其成为省内卫健系统信创试点的核心选择，承担着探索和表率双重使命。电科金仓以“异构多活容灾架构”为核心的技术体系，不仅助力浙人医突破瓶颈，打造国内首个LIS系统国产化异构多院区多活改造案例，更构建了一套适配集团化医院信创的“全链路解决方案”，为行业提供了可落地的技术范本。

一、集团化医院信创的三重难题

浙人医目前拥有朝晖、望江山、越城、富阳四大已运行院区、滨江、萧山两个在建院区及全面托管的八家分院，横跨杭州绍兴两地。浙人医的信创难点，源于其跨区域、多院区、高负荷的运营特性与医疗业务零中断、高安全的本质要求相互交织。此外由于各院区和分院在信息化建设初期技术能力、资源及政策要求等条件不一，受历史因素影响，各主体之间信息化建设的情况差异较大。信创启动前，浙人医院内并存Oracle、SQL Server、MySQL、PostgreSQL等多种数据库，院内的100余个业务系统由多个开发商建设，各系统对国产数据库的适配能力参差不齐。这些问题都需要借助信创契机一并解决。

面对业务需求和政策要求带来的双重压力，浙人医最终选定电科金仓作为数据库合作伙伴，以LIS系统为突破口，开启信创之路。

具体到数据库的选型来看，核心面临以下三重挑战：

兼容性挑战：集团化医院一体化管理模式要求各院区间信息化平台实现全面功能对接和数据共享，异构数据集成和互通的现实需求，对国产数据库的兼容能力提出高要求。
业务连续性挑战：浙人医LIS系统作为核心业务载体之一，支撑着全院日均2万余个标本的处理需求，任何停机都可能导致样本积压、报告延迟，甚至影响临床诊断决策，这对数据库的不停机迁移和数据同步能力提出严苛要求。
高可用与信息安全挑战：此前浙人医的LIS系统业务由越城院区主系统承载，系统压力大，院区间如遇网络中断需要手动拉起灾备系统。医疗数据的敏感性与合规要求下，浙人医提出“容灾恢复能力达到6级标准”（数据零丢失，RPO=0；故障恢复时间 RTO<10 分钟）和“业务连续性达到 99.99%”的硬性指标，同时需要进行密评改造。

二、异构多活架构铸就安全底座

针对浙人医的痛点，电科金仓基于医疗行业特性与集团化医院需求，打造以“异构多活”为核心的技术路线，在完美替代原非信创业务系统的同时，让浙人医数据安全和处理效率有了质的提升。

在浙人医LIS信创项目中，电科金仓实现了四大技术创新：

1. 多活容灾保安全

电科金仓创新设计的异地多活容灾架构，核心在于“多院区互为主备”，彻底改变了浙人医此前面对数据安全的被动局面。目前浙人医拥有朝晖院区、越城院区和富阳院区三个数据中心，各中心之间互为主备，发生故障无需手动拉起灾备系统，可以实现秒级切换，帮助医院实现RTO≤10min、RPO=0的容灾目标，有效提高了业务系统的高可用性。同时各中心之间支持动态负载均衡，横向拆分了各院区压力，稳定性显著提升。

2. 异构组网高兼容

当前浙人医数据库信创处在双轨并行状态，具体部署如下：

朝晖院区：保持非信创系统活跃状态，金仓数据库仅承担局部业务正式访问；
越城院区：以金仓数据库为主，非信创系统作为只读库；
富阳院区（新建）：主数据库和只读库均为国产数据库，非信创系统仅用作数据备份。

金仓数据库凭借其对非信创数据库的原生兼容能力，有效满足了异构组网需求。

3. 多写同步提效率

浙人医三大数据中心支持双向多写。基于业务存量数据校验技术和增量数据校验技术，电科金仓异构数据同步软件KFS可提供全周期数据实时一致性校验且无需中断业务。三大数据中心通过KFS工具实现环状数据同步，异地保持了数据库的三个全量副本，进一步提升了容灾能力。

4. 卫星方案降成本

此外浙人医还联合电科金仓，在小型院区重要工作位置布置轻量化卫星站，通过小型化节点保留核心功能，减少网络依赖，提升应急能力，在降低部署成本的同时进一步提升了系统可用性。

三、项目成效与行业价值

LIS系统在多院区的成功落地，为浙人医信创建设的全面推进奠定了基础，其信创版图持续扩张。富阳院区作为全省首个医疗全栈信创样板间，更实现“云原生+国产化”双重突破。富阳院区试运行以来，运行高效稳定，核心指标表现优异：

系统每小时访问量达40多万次；
数据库IOPS（每秒读写操作数）达到1万以上；
与原系统相比，数据调用时间平均缩短了0.8秒，效率提升了60%；
业务高峰时，系统响应延迟时间≤0.3秒，为患者和医护人员提供了更高效、更快速的服务体验。

浙人医LIS系统关键操作性能对比表

NO.	操作	设计性能指标	非信创数据库实测	金仓数据库实测
1	切换单元	<2秒	<2秒	<2秒
2	样本列表刷新	<1秒	<1秒	<1秒
3	切换样本	20个项目以内<0.3秒；50个项目以内<0.5秒；有图形样本<0.5秒	<0.15秒（后台接口返回：57ms）；<0.2秒（后台接口返回：63ms）；<0.2秒（后台接口返回：69ms）	<0.15秒（后台接口返回：68ms）；<0.2秒（后台接口返回：79ms）；<0.2秒（后台接口返回：84ms）
4	单样本审核	<0.5秒	<0.2秒（后台接口返回：130ms）	<0.2秒（后台接口返回：127ms）
5	批量样本审核	<N*0.5秒	<N*0.2秒	<N*0.2秒
6	单样本发布	<0.5秒	<0.2秒（后台接口返回：90ms）	<0.2秒（后台接口返回：124ms）
7	批量样本发布	<N*0.5秒	<N*0.2秒	<N*0.2秒

四、案例启示与电科金仓实力

浙人医案例为集团化医院信创提供了清晰的可复制路径：

以LIS系统等业务重要、规模可控的系统为突破口，降低初期风险；
采用电科金仓异构多活、双轨并行架构，平衡安全与效率；
依托KFS等工具实现低侵扰改造，减少对现有业务的冲击。

电科金仓凭借杰出的产品力和对医疗行业业务场景的深入理解，已经获得301医院、西京医院、常德二院等数十家医疗机构的认可，蝉联国产数据库销量榜首（点击了解详情）。此次与浙人医的合作，再次证明金仓数据库不仅能够替代国外产品，更能通过架构创新解决医疗场景的特殊需求。电科金仓的异构多活架构，也将持续为更多集团化医院破解信创难题，为构建自主可控的医疗信息化体系注入核心动力。

五、电科金仓数据库常用sql语句

1. 新增数据（INSERT）

KingbaseES中使用INSERT语句向表中添加新数据，基本语法为：

INSERT INTO 表名(字段1, 字段2, ...) VALUES(值1, 值2, ...);

示例：向用户表（user_info）插入一条新记录

INSERT INTO user_info(id, username, age) VALUES(1, '张三', 25);

可同时插入多条记录，只需在VALUES后添加多组值，用逗号分隔。

2. 查询数据（SELECT）

SELECT语句用于从表中查询数据，基本语法为：

SELECT 字段1, 字段2, ... FROM 表名 WHERE 条件;

示例1：查询用户表中所有记录

SELECT * FROM user_info;

示例2：查询年龄大于20的用户姓名

SELECT username FROM user_info WHERE age > 20;

可通过ORDER BY进行排序，使用LIMIT限制返回条数。

3. 更新数据（UPDATE）

UPDATE语句用于修改表中已有数据，基本语法为：

UPDATE 表名 SET 字段1=新值1, 字段2=新值2 WHERE 条件;

示例：将id为1的用户年龄更新为26

UPDATE user_info SET age=26 WHERE id=1;

注意必须添加WHERE条件，否则会更新表中所有记录。

4. 删除数据（DELETE）

DELETE语句用于删除表中的记录，基本语法为：

DELETE FROM 表名 WHERE 条件;

示例：删除id为1的用户记录

DELETE FROM user_info WHERE id=1;

同样需要注意WHERE条件，若省略将删除表中所有数据。

以上四个操作构成了KingbaseES数据库的基本数据操作，通过合理组合使用这些语句，可以实现对数据库的完整管理。在实际应用中，需注意SQL语句的规范性和安全性，尤其是涉及删除和更新操作时要谨慎处理条件判断。

【Java 开发日记】RabbitMQ 里面的交换机是什么，你用过哪种？

Wed, 26 Nov 2025 17:21:50 +0800

交换机是什么？

我用过的交换机类型（及详细介绍）

① 直连交换机

② 扇出交换机

③ 主题交换机

④ 头交换机

总结与对比

交换机是什么？

在 RabbitMQ 中，交换机 是消息路由机制的核心。你可以把它想象成一个邮局分拣员。

生产者 发送消息时，它不是直接把消息放到队列里，而是发送到 交换机。
然后，交换机根据一个特定的规则（这个规则叫做 “绑定” 和 “路由键”）来决定把消息投递到哪些队列中。
绑定是连接交换机和队列的桥梁，你可以为这个桥梁设定一个路由键。

简单流程：
生产者 -> 交换机 -> (根据绑定规则) -> 一个或多个队列 -> 消费者

如果没有交换机，生产者需要直接知道所有队列的存在，这在复杂的、需要灵活路由的系统里是几乎不可行的。交换机解耦了生产者和队列，使得消息的路由策略变得非常灵活和强大。

我用过的交换机类型（及详细介绍）

RabbitMQ 主要提供了四种类型的交换机，每种都有不同的路由行为。最常用的是前三种。

① 直连交换机

类型：direct
行为：一个消息的路由键如果 完全匹配 某个队列的绑定键，那么这条消息就会被路由到该队列。
类比：就像公司里的邮件系统，你写对了工号（路由键），邮件就会被准确地投递到那个人的邮箱（队列）。
使用场景：
- 有明确一对一或一对多任务分发的场景。例如，将错误日志（routing_key: 'error'）只发送给记录错误的队列，将订单消息（routing_key: 'order.paid'）只发送给处理已支付订单的队列。
我的使用经验：这是最常用、最简单的交换机类型。我们在处理不同优先级的任务时经常使用，比如 task.high 和 task.low 分别绑定到不同的队列，由不同性能的消费者来处理。

② 扇出交换机

类型：fanout
行为：它会把发送到该交换机的所有消息广播到所有与它绑定的队列中。它完全忽略路由键。
类比：就像公司里的群发邮件或公告板，一条消息发出，所有订阅了的人都收到一份。
使用场景：
- 发布/订阅模式。例如，用户成功注册后，需要同时执行多个操作：发送欢迎邮件、初始化用户画像、发放新手优惠券。你可以让这三个任务对应的队列都绑定到同一个 fanout 交换机上，这样一条“用户注册成功”的消息会被这三个队列同时接收并处理。
我的使用经验：在需要做事件驱动架构，一个事件触发多个下游服务的场景下非常好用。我们用它来同步不同服务间的缓存数据失效通知。

③ 主题交换机

类型：topic
行为：功能最强大的交换机。它使用路由键和一种模式进行匹配。绑定键（Binding Key）可以包含两种特殊的通配符：
- * (星号)：匹配一个单词。
- # (井号)：匹配零个或多个单词。
- 单词之间用点号 . 分隔，例如 usa.news， europe.weather.serious。
类比：就像新闻订阅系统，你可以订阅“所有美国地区的新闻”（usa.#），或者“所有地区的严重天气”（*.weather.serious）。
使用场景：
- 根据消息的多个属性进行灵活路由。例如，一个日志处理系统，你可以根据日志的严重程度（info, error）和来源（auth, order, payment）来路由。绑定键 *.error 会接收所有服务的错误日志，而 order.* 会接收订单服务的所有日志。
我的使用经验：在构建复杂的消息路由规则时，topic 交换机是首选。我们用它来构建日志收集和业务通知系统，可以根据不同的标签组合将消息精准地投递给感兴趣的消费者。

④ 头交换机

类型：headers
行为：它不依赖于路由键的匹配规则，而是根据消息的 headers 属性 来路由。在绑定时，你需要指定一组键值对。当发送来的消息的 headers 属性与绑定时指定的键值对完全匹配时，消息就会被路由到该队列。
使用场景：
- 用于需要基于多个消息属性（而不仅仅是一个路由键）进行路由的复杂场景。但因为性能比 topic 交换机差，且使用起来更复杂，所以在实际开发中非常少见。
我的使用经验：我个人在实际项目中几乎没有使用过头交换机，topic 交换机已经能满足绝大多数复杂路由的需求，而且更直观高效。

总结与对比

交换机类型	路由行为	使用场景
直连 (direct)	精确匹配 `routing_key`	任务分发、RPC
扇出 (fanout)	广播给所有绑定队列	发布/订阅、事件广播
主题 (topic)	通配符匹配 `routing_key`	灵活的多维消息路由
头 (headers)	匹配 `headers` 属性	复杂属性匹配（不常用）

回答“你用过哪种？”：

在实际工作中，我最常用的是 直连交换机、扇出交换机 和 主题交换机。

用 direct 处理简单的任务分发。
用 fanout 处理需要广播消息的发布/订阅场景。
用 topic 处理需要根据多种条件（如日志级别和模块）进行灵活路由的复杂业务。

如果小假的内容对你有帮助，请点赞，评论，收藏。创作不易，大家的支持就是我坚持下去的动力！

对比nginx、kong、apisix、zuul、gateway网关

Wed, 26 Nov 2025 17:21:47 +0800

Nginx、Kong、APISIX、Zuul、Spring Cloud Gateway 均是 API 网关或反向代理工具，但定位、技术栈和适用场景差异显著。以下从核心定位、优缺点、适用场景三个维度对比分析：

1. Nginx

核心定位

高性能 HTTP 服务器、反向代理服务器、负载均衡器，用 C 语言开发，是网关领域的“基础设施”。
主要功能：静态资源托管、反向代理、负载均衡、SSL 终结、简单路由转发。

优点

性能极强：C 语言编写，异步非阻塞架构，单机支持数十万并发，资源消耗低。
稳定性高：经过数十年验证，适合作为流量入口的“第一道防线”。
功能丰富：支持 URL 重写、缓存、限流（简单）、SSL 配置等基础网关能力。
生态成熟：大量第三方模块（如 ngx_lua）可扩展功能，社区文档丰富。

缺点

动态配置弱：修改配置需重启或 reload（热加载但有短暂阻塞风险），不适合频繁变更的场景。
API 网关特性弱：缺乏原生的服务发现、细粒度限流、熔断、监控等微服务网关功能，需通过 Lua 脚本或第三方模块扩展，开发成本高。
开发门槛高：扩展功能需熟悉 Lua 或 C 模块，对运维人员要求高。

适用场景

作为全局流量入口（边缘网关），处理静态资源、SSL 终结、初步负载均衡。
搭配 ngx_lua 实现简单的 API 路由和限流（如 OpenResty 方案）。
对性能要求极高、配置变更不频繁的场景（如电商大促流量入口）。

2. Kong

核心定位

基于 Nginx + OpenResty（Lua 扩展）的开源 API 网关，专注于 API 全生命周期管理。
主要功能：API 路由、负载均衡、认证授权、限流熔断、监控日志、插件扩展。

优点

性能优秀：基于 Nginx 内核，性能接近原生 Nginx，支持高并发。
插件生态丰富：内置 100+ 插件（如 JWT 认证、OAuth2、限流、监控），支持自定义 Lua 插件。
动态配置：通过 Admin API 实时修改配置（无需重启），适合动态扩缩容的场景。
支持服务发现：可对接 Consul、etcd、Kuber***es 等注册中心。

缺点

学习成本高：插件开发依赖 Lua，运维和扩展需熟悉 OpenResty 生态。
资源消耗较高：相比原生 Nginx 略重，内存占用更大。
企业级功能（如多租户、高级监控）需依赖商业版（Kong Enterprise）。

适用场景

中小型微服务架构的 API 网关，需要丰富的插件功能（如认证、限流）。
混合架构（既有传统服务也有微服务）的统一 API 入口。
对动态配置和可扩展性有要求，但团队能接受 Lua 技术栈的场景。

3. APISIX

核心定位

云原生、高性能的开源 API 网关，基于 Nginx + etcd（配置存储），由中国团队开发，兼容 OpenResty 生态。
主要功能：动态路由、负载均衡、限流熔断、服务发现、监控告警、多语言插件。

优点

性能顶尖：基于 Nginx 内核，采用 etcd 存储配置，动态配置生效速度快（毫秒级），性能略优于 Kong。
云原生友好：原生支持 Kuber***es、服务网格（Service Mesh），适合容器化部署。
插件生态灵活：支持 Lua、Go、Java 等多语言开发插件，内置丰富的微服务相关插件（如 SkyWalking 追踪、普罗米修斯监控）。
中文支持好：文档和社区有大量中文资源，国内用户适配成本低。

缺点

生态成熟度略逊于 Kong（插件数量和社区规模较小）。
部分高级功能（如多集群管理）仍在迭代中。

适用场景

云原生/微服务架构，尤其是 Kuber***es 环境下的 API 网关。
对动态配置速度和性能要求极高的场景（如金融、电商核心服务）。
国内团队（中文文档和社区支持更友好）。

4. Zuul

核心定位

基于 Java 的开源 API 网关，由 ***flix 开发，是 Spring Cloud 早期的默认网关组件（Zuul 1.x），后推出 Zuul 2.x（异步非阻塞）。
主要功能：路由转发、认证授权、负载均衡、简单限流。

优点

Spring Cloud 生态集成：与 Spring Boot 无缝衔接，适合 Java 技术栈团队快速上手。
开发门槛低：用 Java 开发过滤器，符合 Java 开发者习惯。

缺点

性能较差：Zuul 1.x 是同步阻塞架构，高并发下性能瓶颈明显（已被 Spring Cloud Gateway 替代）。
功能简陋：缺乏高级特性（如细粒度限流、动态配置），需大量自定义开发。
社区活跃度低：Zuul 2.x 发布缓慢，生态逐渐萎缩，已不是主流选择。

适用场景

早期 Spring Cloud 项目的过渡期网关（新项目不推荐）。
低并发、对性能要求不高的小型 Java 微服务架构。

5. Spring Cloud Gateway

核心定位

基于 Spring Boot 2.x、Spring WebFlux（响应式编程）的开源 API 网关，是 Spring Cloud 官方推荐的网关（替代 Zuul）。
主要功能：动态路由、负载均衡、熔断限流（集成 Resilience4j/Sentinel）、服务发现（集成 Eureka/Consul）、监控追踪。

优点

Spring 生态无缝集成：天然支持 Spring Cloud 组件（如服务发现、配置中心），Java 团队零学习成本。
性能优秀：基于 ***ty 异步非阻塞架构，性能远超 Zuul 1.x，接近 Nginx 级（但略低于 Kong/APISIX）。
功能丰富：内置路由断言、过滤器（如 JWT 认证、请求重写），支持动态配置（结合 Spring Cloud Config）。
开发便捷：用 Java/Kotlin 开发自定义过滤器，符合微服务开发习惯。

缺点

仅支持 Java 技术栈，对非 Spring 项目适配性差。
性能略逊于 Nginx/Kong/APISIX（JVM 语言的天然开销）。
依赖 Spring 生态，灵活性受限于框架设计。

适用场景

纯 Spring Cloud 微服务架构，需要与 Java 技术栈深度融合的场景。
对开发效率要求高，需快速集成服务发现、熔断等微服务特性的场景。
中高并发场景（性能满足大多数企业级需求）。

对比总结表

特性	Nginx	Kong	APISIX	Zuul	Spring Cloud Gateway
技术栈	C/Lua	Nginx + Lua	Nginx + Lua + etcd	Java（同步阻塞）	Java（***ty 响应式）
性能	极高（10万+ QPS）	高（接近 Nginx）	高（略优于 Kong）	低（万级 QPS）	中高（5万+ QPS）
动态配置	弱（需 reload）	强（Admin API）	极强（etcd 实时同步）	弱（需重启）	强（配置中心集成）
生态	最成熟（第三方模块）	丰富（100+ 插件）	快速成长（多语言插件）	萎缩（基本停滞）	完善（Spring 生态）
开发成本	高（Lua/C）	中（Lua）	中（多语言支持）	低（Java）	低（Java/Spring）
适用场景	边缘网关、静态资源	混合架构 API 网关	云原生/高性能网关	老旧 Spring 项目	Spring 微服务网关

选型建议

追求极致性能 + 静态资源/边缘网关：选 Nginx（搭配 OpenResty 扩展）。
云原生/高动态配置 + 多语言支持：选 APISIX（国内团队优先）。
插件生态丰富 + 混合架构：选 Kong（商业版适合企业级需求）。
纯 Spring Cloud 微服务：选 Spring Cloud Gateway（开发效率最高）。
避免使用：Zuul（性能差，已被替代）。

Spring AI

Wed, 26 Nov 2025 17:21:43 +0800

基本概念

什么是 AI

模型（Model）

大语言模型 (LLM)

提示词（Prompt）

词元（Token）

Spring AI 是什么

快速入门

环境要求

申请 API Key

项目创建

接口编写

核心接口

ChatModel

ChatClient

消息类型

SystemMessage

UserMessage

AssistantMessage

输出格式

结构化输出

流式输出

SSE 协议介绍

SSE 数据格式

data

event

retry

SSE 使用示例

Flux

Advisors

基本概念

什么是 AI

AI：也就是 人工智能（Artificial Intelligence），顾名思义，就是让机器模拟人类智能的科学与技术

我们通过一个示例来对比理解：

普通计算机程序：像一台自动售货机。你按下特定的按钮（输入），它就给你一瓶特定的饮料（输出）。它的所有行为都是程序员预先设定好的规则

人工智能程序：更像是一个正在学习的孩子。你给它看很多猫和狗的图片（数据），并告诉它哪个是猫，哪个是狗。经过学习后，当你给它一张它从未从未见过的猫咪图片时，它也能识别出来。它自己从数据中学会了规律，而不是依赖硬编码的规则

因此，AI 的核心是从经验中学习，并根据所学做出决策或预测

而目前最主流、最引人注目的 AI 分支是 生成式人工智能，也就是现在常说的 AIGC（Artificial Intelligence Generated Content，人工智能生成内容）。它与传统 AI（主要用于分析数据，比如识别人脸）不同，它的目标是利用人工智能技术自动生成或创造出各类数字内容，比如写文章、报告、翻译、编程

为了更好的理解 AI，我们先来理解其中的一些常见术语

模型（Model）

模型（Model）是 AI系统的核心，它是通过算法在数据数据上训练后得到的结果。模型本质上是一个数学函数，它接收输入数据，并进行计算，然后产生输出。

我们常说的“调用一个AI”，实际上就是在使用这个“模型”。模型文件大小不一，可以从几MB到几十GB

可以把 AI 模型想象成一个 “虚拟大脑”。这个大脑通过在大量数据上进行“训练”或“学习”，掌握了一些技能和知识。而当被提问时，就需要运用这个大脑掌握的知识来解决问题

大语言模型 (LLM)

LLM（ Large Language Model，大语言模型）：一种基于深度学习的、使用海量文本数据训练的模型。它的主要任务是理解和生成人类语言。LLM是当前生成式AI热潮的代表。它们的特点是“大”，体现在训练数据量大、模型参数数量巨大

可以将其看做一个进行了超大规模训练的 “专家大脑” 。它通过学习互联网上几乎所有的文本，掌握了语言的语法、句法、事实知识以及上下文逻辑，拥有数百亿甚至数千亿个参数，并且因为它什么都学过，所以能应对各种各样的话题和任务

提示词（Prompt）

提示词（Prompt）：用户提供给AI模型的指令、问题或上下文信息。模型根据提示词来生成相应的回复。提示词的质量直接决定了 AI 回答的质量。

而设计和优化提示词的过程被称为“提示词工程”，是一门新兴的技能。

提示词就像是给AI这位“天才天才”下达的“工作订单”。订单越清晰、越具体，完成的工作质量就越高。

例如：

简单提示词：“法国的首都是哪里？” -> 模型回答：“巴黎。”

复杂提示词（角色扮演）：“假设你是一位资深营养师，请为我（一位办公室久坐的上班族）设计一份为期一周的健康午餐食谱。” -> 模型会以营养师的口吻提供一份详细的食谱。

词元（Token）

词元（Token）：是模型处理和理解的基本文本单位。它不是完全等同于一个英文单词或一个汉字。模型在处理前，会先将文本拆分成词元，同时，词元也是计费和衡量模型处理长度的基本单位。

英文中，单词 “unbelievable” 可能会被拆分成三个词元[“un”, “believe”, “able”]

中文中，“我喜欢编程”这句话，很可能会被拆分成四个词元[“我”, “喜”, “欢”, “编程”]

不同模型的分词规则不同，同一个词在不同模型中可能被拆分成不同词元

了解了 AI 的基本概念，接下来，我们来看 Spring AI 相关内容

Spring AI 是什么

Spring AI 是一个基于 Spring 生态系统的开源人工智能应用框架，它的核心目标是简化 AI 功能在 Java 应用程序中的集成过程，让 Java 开发者也能高效地构建生成式 AI 应用

官方文档：简介 :: Spring AI 参考文档 - Spring 框架

Spring AI 提供了作为开发 AI 应用基础的抽象。这些抽象具有多种实现，可以通过最少的代码更改轻松实现组件切换。

Spring AI 提供了一系列强大而实用的功能，使其成为一个功能完备的 AI 应用开发框架：

1. 统一的多模型支持：支持与众多主流的 AI 模型提供商进行交互，包括 OpenAI、Microsoft、Amazon、Google 和 Anthropic 等，无论是云端模型还是本地部署的模型（如通过 Ollama），都能通过一致的接口进行调用

2. 强大的数据集成能力：这是 Spring AI 的一大亮点。它内置了对向量数据库（如 Chroma、Pinecone、Redis 等）的支持

3. 与 Spring 生态无缝集成：作为 Spring 大家庭的一员，它能自然地与 Spring Boot、Spring Data 等其他知名项目协同工作

4. 简化的开发模式：允许 AI 模型根据需要请求执行客户端定义的函数，从而接入实时信息或触发具体动作

了解了相关概念后，我们就来上手体验一下 Spring AI

快速入门

环境要求

JDK 版本：JDK 17 或以上 (推荐 JDK 21)，这是强制要求，因为 Spring Boot 3.x 本身就需要 JDK 17+

Spring Boot 版本：Spring Boot 3.2 或以上 ，具体版本可以是 3.3.3、3.4.3 或 3.5.0，选择一个稳定的3.x最新版本即可。

AI 服务凭证：有效的 API Key，需要一个来自 AI 服务提供商（如 OpenAI、DeepSeek、阿里百炼等）的账户和 API Key

在本篇文章中，我们以 DeepSeek 作为示例来进行学习

申请 API Key

我们访问 DeepSeek 官网：DeepSeek | 深度求索

进入 API 开放平台：

创建 API Key：

点击创建之后输入名称即可完成创建，但需要注意的是 API key 仅在创建时可见可复制

然后需要进行充值（充值前需要进行实名认证）：

只是学习使用，1块就够了

项目创建

Spring AI 专门为 OpenAI 及兼容 API 服务设计了 spring-ai-openai-spring-boot-starter，用于快速集成大模型语言能力到 Spring Boot 应用中：

正常创建 Maven 项目（注意JDK 和 Spring Boot 版本），并添加 Spring AI 依赖：

        <dependency>
            <groupId>org.springframework.ai</groupId>
            <artifactId>spring-ai-openai-spring-boot-starter</artifactId>
            <version>1.0.0-M6</version>
        </dependency>

依赖版本可参考：https://docs.spring.io/spring-ai/reference/getting-started.html

在 application.yml 中配置 API 密钥：

spring:
  ai:
    openai:
      #DeepSeek
      api-key: 申请的 API Key
      base-url: https://api.deepseek.***
      chat:
        options:
          model: deepseek-chat
          temperature: 0.7

其中：

spring.ai.openai.base-url：要连接的 URL

spring.ai.openai.api-key：申请的 DeepSeek API 密钥

spring.ai.openai.chat.options.model：要使用的 DeepSeek LLM 模型

spring.ai.openai.chat.options.temperature：用于控制模型生成文本的随机性和创造性：低温 (接近 0.0) → 保守、确定、可预测；高温 (接近 2.0) → 冒险、多样、富有想象力。也就是说，temperature 值越低，相同的提问得到的结果越类似。此外，不建议在同一个补全请求中同时修改 temperature 和 top_p，因为这两个设置的交互作用难以预测。

配置项可参考：https://docs.spring.io/spring-ai/reference/api/chat/deepseek-chat.html

此时，就已经完成了 项目的创建 和 DeepSeek 的接入，接下来，我们通过编写接口来调用模型

接口编写

@RestController
@RequestMapping("/deepseek")
public class DeepSeekChatController {
    @Autowired
    private OpenAiChatModel deepSeekChatModel;

    @GetMapping("/chat")
    public String generate(String message) {
        return deepSeekChatModel.call(message);
    }
}

运行，并访问 127.0.0.1:8080/deepseek/chat?message=你是谁进行测试

上述，我们通过 ChatModel 完成了与模型的交互

而在 Spring AI 框架中，ChatModel 和 ChatClient 是构建对话式AI 应用的两大核心接口，接下来，我们分别来看这两个接口

核心接口

ChatModel

ChatMode 直接与底层 AI 模型（如 GPT-4、Claude 等）通信，处理原始的请求和响应

因此，ChatModel 更底层，使用更灵活

@Service
public class ChatService {
    
    @Autowired
    private ChatModel chatModel; // 例如 OpenAiChatModel
    
    public String askQuestion(String question) {
        // 1. 构造消息
        UserMessage userMessage = new UserMessage(question);
        
        // 2. 创建 Prompt
        Prompt prompt = new Prompt(List.of(userMessage));
        
        // 3. 调用并获得完整响应
        ChatResponse response = chatModel.call(prompt);
        
        // 4. 从响应中提取内容
        return response.getResult().getOutput().getContent();
    }
}

ChatClient

ChatClient 在 ChatModel 之上提供了一层流畅的 API，简化了常见的使用模式

即 ChatClient 是对 ChatMode 的一层包装

因此，ChatClient 更高级，也更简洁

@Service
public class ChatService {
    
    @Autowired
    private ChatClient chatClient;
    
    public String askQuestion(String question) {
        // 一行搞定
        return chatClient.call(question);
    }
}

可以看到，ChatClient 的使用更加简洁直观

ChatMode 与 ChatClient 对比：

维度	ChatModel	ChatClient
抽象层级	底层，接近原始模型	高层，面向业务使用
返回值	ChatResponse（包含丰富元数据的完整响应对象）	ChatResponse（直接获得内容的纯文本）或流式响应
使用方法	需要手动构造 Prompt 对象	提供流式的 builder 模式
控制粒度	精细控制	快捷简便

消息类型

在 Spring AI 中，所有消息类型都实现了 org.springframework.ai.chat.messages.Message 接口，系统中的消息被设计用来模拟一个多轮对话中的不同参与者

消息类型	对应角色	核心作用
SystemMessage	系统 / 导演	设定 AI 的背景、角色、行为和回复风格。通常在对话开始时提供，为整个会话定下基调。
UserMessage	用户 / 提问者	代表人机交互中的人类一方，是驱动对话前进的源泉。
AssistantMessage	助理 / AI 本身	代表 AI 在之前轮次中做出的回复。是多轮对话连贯性的保障。
FunctionMessage	函数 / 工具	代表 AI 通过函数调用获得的额外信息或操作结果。
ToolMessage	工具	功能与 ToolMessage 完全相同，是 ToolMessage 的别名。
MediaMessage	多媒体	表示除文本外的其他类型消息数据，例如图像。

其中，最常使用的是 SystemMessage、UserMessage 和 AssistantMessage

SystemMessage

SystemMessage 通常用于设定 AI 助手的身份、性格、行为准则和对话规则，一般位于对话的开头，为整个对话设定基调

例如，我们可以为其进行角色预设：

@RequestMapping("/chat")
@RestController
public class ChatClientController {
    private ChatClient chatClient;
    public ChatClientController(ChatClient.Builder chatClientBuilder) {
        this.chatClient = chatClientBuilder
                .defaultSystem("你叫小小鱼，是一款专业的智能答疑AI助手，擅长Java和Python，以友好的态度来回答问题")
                .build();
    }

    @GetMapping("/call")
    public String generation(String userInput) {
        return this.chatClient.prompt()
                .user(userInput) // 用户输入
                .call() // 调用 API
                .content(); // 返回响应
    }
}

在 ChatClient 中，通过 defaultSystem 来设置 AI 模型的默认系统消息，通过 ChatClient.Builder 链式调用设置的系统消息会作为对话的 "初始指令"，注入到每次对话的上下文中，引导 AI 的回复风格或身份设定

此时，我们访问接口，再次询问其身份：

UserMessage

UserMessage 表示我们提出的具体问题或指令，上述输入的 "你是谁"，就是 UserMessage

AssistantMessage

AssistentMessage：是AI模型给出的回复

AssistentMessage 是实现 连贯多轮对话 的关键。每次 AI 回复后，可以将这个回复作为 AssistantMessage 保存下来，并在下一次请求时将其作为历史上下文的一部分发送给 AI

输出格式

结构化输出

若想要从 LLM 接收结构化输出，Spring AI 支持将 ChatModel/ChatClient 方法的返回类型从 Spring 更改为其他类型

通过 entity() 方法将模型输出转化为自定义实体

例如：

@RequestMapping("/chat")
@RestController
public class ChatClientController { 
    private ChatClient chatClient;
    public ChatClientController(ChatClient.Builder chatClientBuilder) {
        this.chatClient = chatClientBuilder
                .build();
    }

   @GetMapping("/entity")
    public String entity(String userInput) {
        Recipe entity = this.chatClient.prompt()
                .user(String.format("请帮我生成%s的菜谱", userInput))
                .call()
                .entity(Recipe.class);
        return entity.toString();
    }

    record Recipe(String dis, List<String> ingredients) {}
}

流式输出

首先我们对比来看什么是流式输出：

传统输出（非流式）：等待全部生成完成后才一次性返回，用户长时间等待 → 突然显示完整答案。像寄送一封平信，写完所有内容才寄出，对方一次性收到整封信。

流式输出：边生成边返回，立即推送部分结果，几乎立即开始显示 → 逐字逐句增长。像打电话一样，对方一边说话，你一边就能听到。

流式输出过程：

用户提问: "请写一篇关于春天的短文"

AI 模型生成过程:

"春天"... (立即返回)

"春天来了"... (继续返回)

"春天来了，万物复苏"... (持续返回)

直到生成完整回答SSE 协议

Spring AI 主要通过 响应式编程 来实现流式输出，使用 stream() 方法生成 Flux<String> 流

@RequestMapping("/chat")
@RestController
public class ChatClientController { 
    private ChatClient chatClient;
    public ChatClientController(ChatClient.Builder chatClientBuilder) {
        this.chatClient = chatClientBuilder
                .build();
    }

    @GetMapping(value = "/stream", produces = "text/html;charset=utf-8")
    public Flux<String> stream(String userInput) {
        return this.chatClient.prompt()
                .user(userInput)
                .stream()
                .content();
    }

    record Recipe(String dis, List<String> ingredients) {}
}

但是，我们思考这样一个问题：由于 HTTP 协议本身设计为无状态的请求-响应模式，也就是严格来说，无法做到服务器主动推送消息到客户端，那么我们要如何实现服务器的流式响应呢？

我们可以通过 SSE（Server-Sent Events，服务器发送事件）来实现流式传输，允许服务器主动向浏览器推送数据流

SSE 协议介绍

SSE 是一种基于 HTTP 的轻量级实时通信协议，浏览器通过内置的 EventSource API 接收并处理这些实时事件

服务器向客户端声明：接下来发送的是 流消息（streaming）、

此时客户端不会关闭连接，会一直等待服务器发送过来新的数据流

SSE 核心特点：

1. 单向通信：数据流只能从服务器推送到客户端。客户端不能通过这个连接向服务器发送数据（除了最初的建立连接请求）。

2. 基于 HTTP/HTTPS：SSE 使用标准的 HTTP 协议，这意味着它可以轻松地穿越大多数防火墙和代理服务器，无需特殊的配置。

3. 长连接：客户端发起一个普通的 HTTP 请求，但服务器会保持这个连接处于打开状态，而不是在发送一次响应后就关闭它。

4. 文本数据流：服务器通过这个持久的连接，持续地向客户端发送遵循特定格式的文本数据流。

5. 自动重连：SSE 协议内建了重连机制。如果连接意外断开，浏览器会自动尝试重新连接到服务器。

SSE 数据格式

服务器向浏览器发送 SSE 数据，需要设置必须的 HTTP 头信息

Content-Type: text/event-stream;charset=utf-8
Connection: keep-alive

整个数据流由一系列消息组成，每条消息（message）由一行或多行文本构成，每行文本以一个字段名开头，后跟一个冒号和一个空格，然后是字段的值，每条消息以一个空行（即两个连续的换行符 \n\n）结束

每一行格式：[field]: value\n

field 的常见取值有：data、event、id、retry

data

data：消息主体，是最重要的字段，用于承载消息的实际内容，如果一个消息包含多个 data 行，客户端会将它们用换行符 (\n) 连接起来，形成一个完整的数据字符串。可用于传递 JSON 字符串、纯文本、XML 等任何文本数据

示例：

data: 这是一条简单的消息\n\n

data: Hello\n

data: World\n

data: !\n\n

event

event：事件类型，用于指定消息的自定义类型，若提供了此字段，客户端将触发对该特定事件名的监听器；否则，将触发通用的 onmessage 事件，可用于对不同类型的消息进行分类处理

示例：

event: userJoined
data: Alice

id

id：事件id，用于为消息设置一个唯一的 ID（字符串），如果连接中断，当客户端重新连接时，会在 HTTP 请求头 Last-Event-ID 中自动发送最后一个接收到的 ID。可以用于实现消息的幂等性和断点续传。

示例：

id: msg-123
data: 这是一条重要消息

retry

retry：重连时间，表示建议浏览器在连接断开后再次尝试连接之前应等待的毫秒数，由于这不是一个强制命令，浏览器可能会忽略它。用于避免在服务器出现故障时，客户端过于频繁地重试。

示例：告诉浏览器，如果连接失败，请等待 10 秒后再尝试重连

retry: 10000

我们通过一个简单的示例来看 SSE 协议的使用

SSE 使用示例

后端接口：

@Slf4j
@RequestMapping("/sse")
@RestController
public class SseController {
    @RequestMapping("/end")
    public void end(HttpServletResponse response) throws IOException, InterruptedException {
        log.info("发起请求: event");
        response.setContentType("text/event-stream;charset=utf-8");
        PrintWriter writer = response.getWriter();
        for (int i = 0; i < 10; i++) {
            // 事件 foo 事件
            String s = "event: foo\n";
            s += "data: " + new Date() + "\n\n";
            writer.write(s);
            writer.flush();
            Thread.sleep(1000L);
        }
        //定义end事件, 表示当前流传输结束
        writer.write("event: end\ndata: EOF\n\n");
        writer.flush();
    }
}

前端实现：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>SSE</title>
</head>
<body>
    <div id="sse"></div>
    <script>
        let eventSource = new EventSource("/sse/end");
        eventSource.addEventListener("foo", function(event) {
            console.log(event);
            document.getElementById("sse").innerHTML = event.data;
        });
        eventSource.addEventListener("end", function(event) {
            console.log("连接关闭")
            eventSource.close();
        });
    </script>
</body>
</html>

运行，观察控制台打印的日志：

可以看到成功传输消息，并在消息传输完毕后关闭连接

而在 Spring 中，可以通过 WebFlux 优雅地实现 SSE 协议，也就是我们之前使用的 Flux，它是 WebFlux 中的核心组件，我们来看 Flux 的使用和常见操作

Flux

Flux 的使用流程：创建 → 转换 → 过滤 → 消费

创建 Flux：

import reactor.core.publisher.Flux;

// 1.1 从固定值创建
Flux<String> fixedFlux = Flux.just("Hello", "World", "!");

// 1.2 从集合创建
List<String> list = Arrays.asList("A", "B", "C");
Flux<String> fromCollection = Flux.fromIterable(list);

// 1.3 数值范围
Flux<Integer> rangeFlux = Flux.range(1, 5); // 1,2,3,4,5

// 1.4 动态生成
Flux<Long> intervalFlux = Flux.interval(Duration.ofSeconds(1)).take(5);

// 1.5 从数组创建
Flux<String> arrayFlux = Flux.fromArray(new String[]{"X", "Y", "Z"}));

// 1.6 空流
Flux<String> emptyFlux = Flux.empty();

// 1.7 从 Future 创建（适配传统异步API）
***pletableFuture<String> future = ***pletableFuture.supplyAsync(() -> "Result");

Flux<String> futureFlux = Flux.fromFuture(future);

转化操作：

// 2.1 map - 一对一转换
Flux<String> original = Flux.just("apple", "banana", "cherry");
Flux<String> uppercased = original.map(String::toUpperCase); // APPLE, BANANA, CHERRY

// 2.2 flatMap - 一对多转换（异步展平）
Flux<String> words = Flux.just("hello world", "spring ai");
Flux<String> splitWords = words.flatMap(word -> 
    Flux.fromArray(word.split(" "))
);

// 2.3 cast - 类型转换
Flux<Object> objects = Flux.just("text1", "text2");
Flux<String> strings = objects.cast(String.class);

// 2.4 scan - 累积计算
Flux<Integer> numbers = Flux.range(1, 4);
Flux<Integer> cumulativeSum = numbers.scan((a***, current) -> a*** + current); // 1,3,6,10

过滤操作：

// 3.1 filter - 条件过滤
Flux<Integer> allNumbers = Flux.range(1, 10);
Flux<Integer> evenNumbers = allNumbers.filter(n -> n % 2 == 0); // 2,4,6,8,10

// 3.2 distinct - 去重
Flux<String> withDuplicates = Flux.just("A", "B", "A", "C");
Flux<String> uniqueItems = withDuplicates.distinct()); // A,B,C

// 3.3 take - 取前 N 个
Flux<String> limited = original.take(2); // apple, banana

// 3.4 skip - 跳过前 N 个
Flux<String> skipped = original.skip(1); // banana, cherry

// 3.5 takeWhile / skipWhile - 条件取/跳
Flux<Integer> sequence = Flux.range(1, 100);
Flux<Integer> firstPart = sequence.takeWhile(n -> n < 10); // 1,2,3,...,9

// 3.6 sample - 采样（定期取最新元素）
Flux<Long> sampled = Flux.interval(Duration.ofMillis(100)))
        .sample(Duration.ofSeconds(1)))
        .take(3); // 每隔1秒取样，共取3次

消费操作：

// 4.1 subscribe - 最基本的消费方式
Flux<String> data = Flux.just("one", "two", "three");
data.subscribe(
    item -> System.out.println("Received: " + item),
    error -> System.err.println("Error: " + error)),
   ,
    () -> System.out.println("***pleted!"))
);

// 4.2 collectList - 收集所有元素到 List
Mono<List<String>> listMono = data.collectList());

// 4.3 blockFirst / blockLast - 阻塞获取（仅用于测试）
// String first = data.blockFirst();

// 4.4 reduce - 归约操作
Mono<Integer> sum = numbers.reduce(0, Integer::sum));

// 4.5 count - 计数
Mono<Long> count = data.count();

// 4.6 hasElement - 检查是否有元素
Mono<Boolean> hasData = data.hasElements();

// 4.7 then - 忽略元素，只在完成后触发
Mono<Void> ***pletionSignal = data.then();

Advisors

Advisors 是 Spring AI 中的一种拦截器机制，允许我们在 AI 调用链的特定节点注入自定义逻辑。

Advisors 在两个关键的时机点介入：

Before Call（调用前）：在请求发送到 AI 模型之前执行，主要用于修改提示词

After Call（调用后）：在收到 AI 响应后、返回给客户端之前执行

其执行流程为：

用户输入 → Advisor1.before() → Advisor2.before() → AI 模型调用 → Advisor2.after() → Advisor1.after() → 最终响应

在 Spring AI 中内置了一些 Advisor，如 SimpleLoggerAdvisor，其主要功能是进行日志记录，只需要将其添加到 Advisor 链中，就可以自动记录 Advisor 的聊天请求和响应：

@RequestMapping("/chat")
@RestController
public class ChatClientController { 
    private ChatClient chatClient;
    public ChatClientController(ChatClient.Builder chatClientBuilder) {
        this.chatClient = chatClientBuilder
                .defaultSystem("你叫小小鱼，是一款专业的智能答疑AI助手，擅长Java和Python，以友好的态度来回答问题")
                .build();
    }

    @GetMapping("/advisor")
    public String advisor(String userInput) {
        return this.chatClient.prompt()
                .advisors(new SimpleLoggerAdvisor())
                .user(userInput)
                .call()
                .content();
    }

    record Recipe(String dis, List<String> ingredients) {}
}

我们将日志级别配置为 debug 来观察：

logging:
  level:
    org.springframework.ai.chat.client.advisor: debug

观察打印的日志内容：

成功打印了对应的请求和响应日志

清晰易懂的 PHP 安装与配置教程

Wed, 26 Nov 2025 17:21:41 +0800

初学者也能看懂的 PHP 安装与配置教程

本教程将手把手教你如何在 Windows 系统上安装 PHP，并配置 ***poser（PHP 的依赖管理工具）的缓存位置，即使你是零基础小白，也能轻松完成！

一、准备工作

操作系统：Windows 10/11。
下载工具：浏览器（推荐 Chrome 或 Edge）。
存储空间：至少预留 200MB 可用空间。

二、安装 PHP

1. 下载 PHP

访问 PHP 官网下载页面：https://windows.php.***/download
选择适合你系统的版本：
- 推荐下载 Non Thread Safe (NTS) 版本（如 php-8.2.10-nts-Win32-vs16-x64.zip）。
- 如果你的系统是 32 位，选择 x86 版本。

2. 解压 PHP

将下载的 ZIP 文件解压到一个目录（如 D:\PHP）。
解压后，你会看到以下文件：
- php.exe：PHP 解释器。
- php.ini-development：PHP 配置文件模板。

3. 配置 PHP

将 php.ini-development 文件重命名为 php.ini。

打开 php.ini，找到以下配置项并修改：

启用扩展（去掉前面的分号 ;）：

extension_dir = "ext"
extension=curl
extension=gd
extension=mbstring
extension=mysqli
extension=pdo_mysql
extension=openssl

设置时区：
```
date.timezone = Asia/Shanghai
```

4. 配置环境变量

右键“此电脑” → 属性 → 高级系统设置 → 环境变量。
在“系统变量”中找到 Path，点击“编辑”。
点击“新建”，输入 PHP 的安装路径（如 D:\PHP）。
点击“确定”保存。

5. 验证安装

打开命令提示符（Win + R → 输入 cmd → 回车）。

输入以下命令：

php -v
# 输出示例：PHP 8.2.10 (cli) (built: Aug 29 2023 12:00:00)

三、安装 ***poser（PHP 依赖管理工具）

1. 下载 ***poser

访问 ***poser 官网：https://get***poser.org/download
下载 Windows 安装程序（***poser-Setup.exe）。

2. 安装 ***poser

双击 ***poser-Setup.exe，启动安装程序。
选择 PHP 路径（如 D:\PHP\php.exe）。
勾选“Add ***poser to your PATH”，点击“Install”。
完成安装后，点击“Finish”。

3. 验证 ***poser

打开命令提示符，输入以下命令：

***poser --version
# 输出示例：***poser version 2.6.5 2023-10-06 10:11:52

四、配置 ***poser 缓存位置

默认情况下，***poser 会将下载的依赖包缓存到 C:\Users\<你的用户名>\AppData\Local\***poser。如果你想将缓存位置改为其他目录，可以按照以下步骤操作：

1. 设置环境变量

右键“此电脑” → 属性 → 高级系统设置 → 环境变量。
在“系统变量”中点击“新建”，输入：
- 变量名：***POSER_CACHE_DIR
- 变量值：自定义路径（如 D:\***poserCache）。
点击“确定”保存。

2. 验证缓存路径

打开命令提示符，输入以下命令：

***poser config --global cache-dir
# 输出示例：D:\***poserCache

五、编写并运行第一个 PHP 程序

创建文件 hello.php，输入以下代码：
```
<?php
echo "你好，世界！";
?>
```
运行程序：
- 打开命令提示符，进入文件所在目录（如 cd D:\PHP）。
- 输入以下命令：
```
php hello.php
# 输出：你好，世界！
```

六、常见问题

php 命令无效：
- 检查是否将 PHP 安装路径添加到环境变量 Path 中。
***poser 安装失败：
- 确保 PHP 已正确安装，并且 php.ini 中启用了 openssl 扩展。
缓存路径不生效：
- 确保环境变量 ***POSER_CACHE_DIR 已正确设置，并重启命令提示符。

七、总结

通过本教程，你已成功完成以下操作：

安装 PHP 并验证环境。
安装 ***poser 并配置缓存位置（解放 C 盘空间）。
编写并运行第一个 PHP 程序。

接下来可以学习：

使用 ***poser 安装第三方库（如 laravel/framework）。
探索 PHP 基础语法和 Web 开发（如 Laravel 框架）。

遇到问题欢迎留言讨论，祝你早日成为 PHP 高手！ 🐘🚀

AJ教程_站长资源网

为何最终我放弃了 Go 的 sync.Pool

一、使用场景

1.1、引入：

1.2、什么是sync.pool？

二、如何使用

2.1、声明对象池

2.2、GET & PUT

三、实例：

3.1、标准库中的应用

3.1.1: fmt.Printf

3.2、Gin框架的应用(context)

3.2.1、定义对象池

3.2.2、初始化对象池

3.2.3、从池中获取Context

3.2.4、处理请求后放回池中

四、我在项目中的实战

4.1、为何最初选择sync.Pool

4.2、又为何选择放弃sync.pool

4.2.1、存储驱动通常是无状态的

五、总结

六、sync.Pool的底层剖析

6.1 底层结构体

6.2 重点

6.2.1 local unsafe.Pointer

6.2.2 victim

七、性能测试

7.1 测试主函数

7.2 对象的复用率

7.3 对象复用性能测试

八、自测

【MCP探索实践】Google GenAI Toolbox：Google开源的企业级AI数据库中间件、5分钟搞定LLM-SQL安全互联

系列篇章💥

前言

一、项目概述

二、技术原理

（一）、整体架构

（二）、性能与安全

三、主要功能

（一）、核心能力

（二）、高级特性

四、应用场景

（一）、RAG 知识库

（二）、NL2SQL 数据助手

（三）、智能客服 Agent

（四）、低代码 BI

（五）、AIOps

五、快速使用

（一）、环境准备

（二）、5 分钟上手

（三）、Docker 一键部署

（四）、Kuber***es 生产级部署示例

（五）、常见踩坑与排查

结语

项目地址

π0源码(openpi)剖析——从π0模型架构的实现：如何基于PaLI-Gemma和扩散策略去噪生成动作，到基于C/S架构下的模型训练与部署

前言

第一部分 π0模型架构的实现：src下models的全面分析与解读

1.1 models/model.py：核心基础模型的定义

1.1.1 基础组件和关键常量

1.1.2 `Observation` 类与Actions类型的详解

1.1.3 preprocess_observation

1.1.4 BaseModelConfig(abc.ABC)

1.1.5 class BaseModel(nnx.Module, abc.ABC)

1.1.6 restore_params

1.2 models/pi0.py的实现

1.2.1 make_attn_mask：注意力掩码生成函数

1.2.2 posemb_sincos：位置编码函数

1.2.3 class Pi0Config：含inputs_spec、get_freeze_filter

1.2.3.1 模型配置参数的定义

1.2.3.2 inputs_spec：定义了π0模型本身接收的输入数据格式

1.2.3.3 get_freeze_filter：参数冻结器，包含谁则相当于谁被冻结/过滤

1.2.4 class Pi0：含特征嵌入(embed_prefix/embed_suffix)、损失函数(训练去噪的准确性)、推理(去噪生成动作)

1.2.4.1 初始化方法 `__init__`

1.2.4.2 特征嵌入方法：embed_prefix(图像和文本输入)、embed_suffix(状态和动作信息)

1.2.4.3 损失函数***pute_loss：训练模型去噪的准确率

注解 LeRobotDataset：训练数据集的来源(即训练数据集长什么样)

1.2.4.4 推理函数 `sample_actions`：基于扩散模型逆向采样(即去噪)，生成机器人动作序列

1.3 语言模型实现：models/gemma.py

1.4 视觉模型实现：models/siglip.py

1.2.4.1 初始化方法 `init`

1.6.2 Einsum类中的call

1.6.4 FeedForward类中的setup、call、_dot

第二部分策略适配接口：src下policy的全面分析与解读

第三部分模型训练的配置：src下training模块的全面分析与解读

第四部分模型的训练与部署：基于客户端-服务器C/S架构——openpi-Client/Scripts

4.2.1 init.py