full-stack-learning_go-python-ts-react

go

前面先让 ai 去帮我整理了 go 的系列知识点 Go-MOC，看了 l3 以下等级，大致浏览后，学了 boot.dev 中的几门课程 Boot.dev Learn Go

开始在之前用 ai vibe 的实际项目中学习 go：

1. 先看已有的代码
   1. 学实践中的新知识
   2. 把前面看到的知识点和实践结合
2. 手搓新的代码
   1. 完善项目
   2. 应用

阅读已有代码

建立整体心智模型

阅读了 user 相关的整个流程 http > handler > service > repository > model > postgresql

model

type User struct {
    ID           uuid.UUID  `gorm:"type:uuid;primaryKey;default:uuid_generate_v4()" json:"id"`
    Email        string     `gorm:"type:varchar(255);uniqueIndex;not null" json:"email"`
    PasswordHash string     `gorm:"type:varchar(255);not null" json:"-"`
    CreatedAt    time.Time  `gorm:"not null;default:now()" json:"created_at"`
    LastLoginAt  *time.Time `json:"last_login_at,omitempty"`
}

func (User) TableName() string {

    return "users"

}

从这里能够看到，该项目首先使用了 GORM，也就是 Go 语言里的一个 ORM（Object-Relational Mapping，对象关系映射）框架。

所谓的 ORM，主要是让开发者可以直接用指定的语言去实现 SQL 相关的操作，而不需要手写原生 SQL。它通过一套映射语言，让开发者直接用自己熟悉的语言来编写数据库操作。目前来看，它比较适合简单的 CRUD 操作；但对于比较复杂的查询操作，主要还是适合写原生 SQL，因为当请求复杂之后，使用 ORM 写的语句其实和原生 SQL 也没什么两样。

还有一个具体细节，这里大致是通过 struct（结构体）来建立用户对象的模型（包含 ID、email 等）：

1. 第二列是它的数据类型。
2. 第三列用反引号（飘号）标注的，是用于 GORM 和 JSON 转换的 tag 注释。这样在进行 GORM 操作或 JSON 转换时，程序就能知道对应的转换数据结构和数据类型是什么样的。
3. 此外，在 GORM 中，它默认是通过一个 TableName 方法（比如 user 的 TableName 方法）来获取该模型在数据库中对应的表名。
4. 还有一个点就是其中的 last_login_at，它的类型是星号 time.Time。它为什么要使用指针类型呢？我后面了解到是因为 last_login_at 是可选字段，所以它可能是没有值的。而 Go 语言里面，就应该会需要用到指针类型，去使用 nil 来表示空值。如果是正常的非指针类型，没有值的话，会初始化为零值；而时间类型的零值，就是一个特殊值，可能会导致误解或者错误。如果是正常的非指针类型，没有值的话，会初始化为零值；而时间类型的零值，就是一个特殊值（即 0001-01-01 00:00:00 UTC），可能会导致误解或者错误。

repository

package repository

import (
	"context"
	"time"

	"github.com/bodysense/api/internal/model"
	"github.com/google/uuid"
	"gorm.io/gorm"
)

// 把“数据库操作”集中封装起来，让上层业务代码不要直接到处调用 GORM。
// 保存一个数据库访问工具，然后提供一堆和 User 有关的方法。
// UserRepository handles user database operations.
type UserRepository struct {
	// 传入一个数据库操作对象 db *gorm.DB，UserRepository 将使用它来执行数据库操作。这样可以实现依赖注入，使得 UserRepository 更加灵活和可测试。
	// 这里的 * 是为了表示 db 是一个指针类型，指向 gorm.DB 的实例。gorm.DB 是 GORM 框架中用于与数据库交互的主要对象，它封装了数据库连接和操作方法。
	// 数据库操作入口 / database handle
	db *gorm.DB
}

// NewUserRepository creates a new UserRepository.
func NewUserRepository(db *gorm.DB) *UserRepository {
	return &UserRepository{db: db}
}

// Create creates a new user.
func (r *UserRepository) Create(ctx context.Context, user *model.User) error {
	return r.db.WithContext(ctx).Create(user).Error
}

// FindByEmail finds a user by email.
func (r *UserRepository) FindByEmail(ctx context.Context, email string) (*model.User, error) {
	var user model.User
	err := r.db.WithContext(ctx).Where("email = ?", email).First(&user).Error
	if err != nil {
		return nil, err
	}
	return &user, nil
}

// FindByID finds a user by ID.
func (r *UserRepository) FindByID(ctx context.Context, id uuid.UUID) (*model.User, error) {
	var user model.User
	err := r.db.WithContext(ctx).Where("id = ?", id).First(&user).Error
	if err != nil {
		return nil, err
	}
	return &user, nil
}

// UpdateLastLoginAt updates the user's last login timestamp.
func (r *UserRepository) UpdateLastLoginAt(ctx context.Context, userID uuid.UUID) error {
	now := time.Now()
	return r.db.WithContext(ctx).Model(&model.User{}).Where("id = ?", userID).Update("last_login_at", now).Error
}

// EmailExists checks if an email already exists.
func (r *UserRepository) EmailExists(ctx context.Context, email string) (bool, error) {
	var count int64
	err := r.db.WithContext(ctx).Model(&model.User{}).Where("email = ?", email).Count(&count).Error
	return count > 0, err
}

// DeleteByID deletes a user by ID.
// WARNING: Callers must also invalidate the user's session cache.
func (r *UserRepository) DeleteByID(ctx context.Context, id uuid.UUID) error {
	return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.User{}).Error
}

接下来是与上面一层对应的仓储层（Repository 层），这里主要进行数据库操作的集合：

1. 定义 UserRepository 结构体： UserRepository 持有一个 *gorm.DB 数据库操作句柄，Repository 可以复用这个初始化后的数据库入口；GORM 的链式调用又会基于它创建相应的 *gorm.DB 会话，并共享底层数据库资源。
2. 基础初始化（NewUserRepository）： 新建一个真正的 User 仓储实例。我们需要把上面提到的数据库操作入口传进去，构造一个用户仓储对象并返回。需要注意到，这里返回的也是一个地址（即指针对象），创建一个 UserRepository，然后返回它的地址。。
3. 基础的 Find 方法（通过邮箱查找用户）： 这里重点关注它的参数中有一个 ctx context.Context（上下文）。这是 Go 的一个基础特性，能够通过 Context 传递取消、超时/截止时间和请求范围的数据。

在具体实现上，它通过 Go 的 GORM 语法去调用仓储实例的 r.db，携带上下文，然后调用 .Where() 传入查询字段 “email = ?”。这里的问号是一个占位符，随后传入对应的变量，通过这种参数化查询的方法来避免 SQL 注入。最后调用 .First() 找到第一条数据，并写入到定义好的 user 变量中。

这里我关注到一个 Go 语言的特性： 在很多地方，Go 的写入方法与 JS 等语言有所不同。一般情况下，在 JS 中都是通过方法返回值，再赋值给定义的变量；但在 Go 里面，存在很多方法是需要你传入一个变量的指针（地址），然后直接将值写入到该地址对应的变量中。这里确实很有 Go 语言的独特特性。

service

package service

import (
	"context"
	"errors"
	"fmt"
	"log"

	"github.com/bodysense/api/internal/auth"
	"github.com/bodysense/api/internal/cache"
	"github.com/bodysense/api/internal/database"
	"github.com/bodysense/api/internal/dto"
	"github.com/bodysense/api/internal/model"
	"github.com/bodysense/api/internal/repository"
	"github.com/google/uuid"
	"golang.org/x/crypto/bcrypt"
	"gorm.io/gorm"
)

// AuthService handles authentication business logic.
type AuthService struct {
	userRepo     *repository.UserRepository
	jwtConfig    auth.JWTConfig
	sessionCache *cache.UserSessionCache
}

// NewAuthService creates a new AuthService.
func NewAuthService(userRepo *repository.UserRepository, jwtConfig auth.JWTConfig, sessionCache *cache.UserSessionCache) *AuthService {
	return &AuthService{
		userRepo:     userRepo,
		jwtConfig:    jwtConfig,
		sessionCache: sessionCache,
	}
}

// Register creates a new user account.
func (s *AuthService) Register(ctx context.Context, req dto.RegisterRequest) (*dto.AuthResponse, error) {
	// Check if email already exists
	exists, err := s.userRepo.EmailExists(ctx, req.Email)
	if err != nil {
		return nil, fmt.Errorf("failed to check email existence: %w", err)
	}
	if exists {
		return nil, errors.New("registration failed")
	}

	// Hash password with bcrypt (cost >= 12)
	hashedPassword, err := bcrypt.GenerateFromPassword([]byte(req.Password), 12)
	if err != nil {
		return nil, fmt.Errorf("failed to hash password: %w", err)
	}

	// Create user
	user := &model.User{
		ID:           uuid.New(),
		Email:        req.Email,
		PasswordHash: string(hashedPassword),
	}

	if err := s.userRepo.Create(ctx, user); err != nil {
		return nil, fmt.Errorf("failed to create user: %w", err)
	}

	// Generate tokens (also writes session cache)
	return s.generateTokens(ctx, user)
}

// Login authenticates a user and returns tokens.
func (s *AuthService) Login(ctx context.Context, req dto.LoginRequest) (*dto.AuthResponse, error) {
	// Find user by email
	user, err := s.userRepo.FindByEmail(ctx, req.Email)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, errors.New("invalid email or password")
		}
		return nil, fmt.Errorf("failed to find user: %w", err)
	}

	// Verify password
	if err := bcrypt.CompareHashAndPassword([]byte(user.PasswordHash), []byte(req.Password)); err != nil {
		return nil, errors.New("invalid email or password")
	}

	// Update last login time
	if err := s.userRepo.UpdateLastLoginAt(ctx, user.ID); err != nil {
		// Log error but don't fail login
		log.Printf("Warning: failed to update last login time: %v", err)
	}

	// Generate tokens (also writes session cache)
	return s.generateTokens(ctx, user)
}

// RefreshToken refreshes an access token using a refresh token.
func (s *AuthService) RefreshToken(ctx context.Context, req dto.RefreshRequest) (*dto.AuthResponse, error) {
	// Get user ID from Redis
	redisClient := database.RedisClient
	key := fmt.Sprintf("refresh_token:%s", req.RefreshToken)

	userIDStr, err := redisClient.Get(ctx, key).Result()
	if err != nil {
		return nil, errors.New("invalid or expired refresh token")
	}

	// Delete old refresh token
	redisClient.Del(ctx, key)

	// Parse user ID
	userID, err := uuid.Parse(userIDStr)
	if err != nil {
		return nil, errors.New("invalid refresh token")
	}

	// Find user — if user no longer exists, clean up and reject
	user, err := s.userRepo.FindByID(ctx, userID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			// User was deleted — clear stale session cache
			_ = s.sessionCache.Delete(ctx, userID)
			return nil, errors.New("user no longer exists")
		}
		return nil, fmt.Errorf("failed to find user: %w", err)
	}

	// Generate new tokens (also writes session cache)
	return s.generateTokens(ctx, user)
}

// Logout invalidates a refresh token and clears the user's session cache.
// Looks up the user ID from the refresh token in Redis before cleaning up.
func (s *AuthService) Logout(ctx context.Context, refreshToken string) error {
	redisClient := database.RedisClient
	refreshKey := fmt.Sprintf("refresh_token:%s", refreshToken)

	// Look up user ID from refresh token (before deleting it)
	userIDStr, err := redisClient.Get(ctx, refreshKey).Result()
	if err != nil {
		// Refresh token not found or expired — nothing to clean up
		return nil
	}

	// Delete refresh token from Redis
	redisClient.Del(ctx, refreshKey)

	// Delete session cache if we have a valid user ID
	if userID, parseErr := uuid.Parse(userIDStr); parseErr == nil {
		if err := s.sessionCache.Delete(ctx, userID); err != nil {
			log.Printf("[AuthService] Failed to delete session cache on logout: %v", err)
		}
	}

	return nil
}

// DeleteUser deletes a user from the DB and invalidates all their sessions.
// Any future request with the user's JWT will be rejected by the middleware.
func (s *AuthService) DeleteUser(ctx context.Context, userID uuid.UUID) error {
	// Delete from DB first
	if err := s.userRepo.DeleteByID(ctx, userID); err != nil {
		return fmt.Errorf("failed to delete user: %w", err)
	}

	// Clear session cache — middleware will now reject this user's tokens
	if err := s.sessionCache.Delete(ctx, userID); err != nil {
		log.Printf("[AuthService] Failed to delete session cache for deleted user %s: %v", userID, err)
	}

	return nil
}

// generateTokens generates access and refresh tokens for a user,
// and writes the user session cache entry.
func (s *AuthService) generateTokens(ctx context.Context, user *model.User) (*dto.AuthResponse, error) {
	// Generate access token
	accessToken, err := auth.GenerateAccessToken(s.jwtConfig, user.ID, user.Email)
	if err != nil {
		return nil, fmt.Errorf("failed to generate access token: %w", err)
	}

	// Generate refresh token
	refreshToken, err := auth.GenerateRefreshToken()
	if err != nil {
		return nil, fmt.Errorf("failed to generate refresh token: %w", err)
	}

	// Store refresh token in Redis
	redisClient := database.RedisClient
	key := fmt.Sprintf("refresh_token:%s", refreshToken)
	if err := redisClient.Set(ctx, key, user.ID.String(), s.jwtConfig.RefreshTokenTTL).Err(); err != nil {
		return nil, fmt.Errorf("failed to store refresh token: %w", err)
	}

	// Write user session cache (best-effort, don't fail token generation)
	if err := s.sessionCache.Set(ctx, user.ID); err != nil {
		log.Printf("[AuthService] Failed to cache user session for %s: %v", user.ID, err)
	}

	return &dto.AuthResponse{
		AccessToken:  accessToken,
		RefreshToken: refreshToken,
		ExpiresIn:    int64(s.jwtConfig.AccessTokenTTL.Seconds()),
	}, nil
}

这份 AuthService 已经基本覆盖了 Go 后端开发中非常高频的一批语言习惯。尤其是 value, err、if err := …; err != nil、%w、errors.Is、指针 receiver、Context 往下传，后面几乎每个 Go 项目都会反复遇到。

handler

package handler

import (
	"net/http"

	"github.com/bodysense/api/internal/dto"
	"github.com/bodysense/api/internal/service"
	"github.com/gin-gonic/gin"
)

// AuthHandler handles authentication HTTP requests.
type AuthHandler struct {
	authService *service.AuthService
}

// NewAuthHandler creates a new AuthHandler.
func NewAuthHandler(authService *service.AuthService) *AuthHandler {
	return &AuthHandler{authService: authService}
}

// Register handles user registration.
func (h *AuthHandler) Register(c *gin.Context) {
	var req dto.RegisterRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		respondError(c, http.StatusBadRequest, "VALIDATION_ERROR", err.Error())
		return
	}

	resp, err := h.authService.Register(c.Request.Context(), req)
	if err != nil {
		status := http.StatusInternalServerError
		code := "REGISTRATION_FAILED"

		if err.Error() == "registration failed" {
			status = http.StatusConflict
		}

		respondError(c, status, code, "registration failed")
		return
	}

	c.JSON(http.StatusCreated, resp)
}

// Login handles user login.
func (h *AuthHandler) Login(c *gin.Context) {
	var req dto.LoginRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		respondError(c, http.StatusBadRequest, "VALIDATION_ERROR", err.Error())
		return
	}

	resp, err := h.authService.Login(c.Request.Context(), req)
	if err != nil {
		respondError(c, http.StatusUnauthorized, "AUTHENTICATION_FAILED", "invalid email or password")
		return
	}

	c.JSON(http.StatusOK, resp)
}

// RefreshToken handles token refresh.
func (h *AuthHandler) RefreshToken(c *gin.Context) {
	var req dto.RefreshRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		respondError(c, http.StatusBadRequest, "VALIDATION_ERROR", err.Error())
		return
	}

	resp, err := h.authService.RefreshToken(c.Request.Context(), req)
	if err != nil {
		respondError(c, http.StatusUnauthorized, "REFRESH_FAILED", "invalid or expired refresh token")
		return
	}

	c.JSON(http.StatusOK, resp)
}

// Logout handles user logout (invalidates refresh token + session cache).
func (h *AuthHandler) Logout(c *gin.Context) {
	var req dto.RefreshRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		respondError(c, http.StatusBadRequest, "VALIDATION_ERROR", err.Error())
		return
	}

	if err := h.authService.Logout(c.Request.Context(), req.RefreshToken); err != nil {
		respondError(c, http.StatusInternalServerError, "LOGOUT_FAILED", "failed to logout")
		return
	}

	c.JSON(http.StatusOK, gin.H{"message": "Logged out successfully"})
}

// Me returns the current authenticated user info.
func (h *AuthHandler) Me(c *gin.Context) {
	userID, exists := c.Get("user_id")
	if !exists {
		respondError(c, http.StatusUnauthorized, "UNAUTHORIZED", "authentication required")
		return
	}

	uid, ok := userID.(string)
	if !ok {
		respondError(c, http.StatusInternalServerError, "INTERNAL_ERROR", "invalid user id type")
		return
	}

	email, _ := c.Get("email")
	emailStr, _ := email.(string)

	c.JSON(http.StatusOK, dto.UserResponse{
		ID:    uid,
		Email: emailStr,
	})
}

这里首先开头又是一个经典的构造，即 AuthService 对象的引用。

这里它的返回 Service.AuthService 本质上其实不是为了保证单例，而主要是为了让 Handler 持有一个已经构造好的 AuthService 对象的引用，然后所有请求都通过这个 Service 去执行认证业务。

这样做有几个好处：

• 不复制整个 AuthService struct
• 多个地方可以引用同一个 service 对象
• 符合 Go 里 service/repository 这种长期对象的常见使用方式
• AuthService 自己的方法 receiver 本身也是 *AuthService

c *gin.Context 这里是一个传入的参数。我看了一下，是基于一个 Go 的框架 GIN。这个框架本质上是建立在 Go 的基础框架 net/http 这个基础库上，然后进行了一层封装。是一个为一次 HTTP 请求准备的综合工作台。它封装主要是对 http.Request, ResponseWriter, Router, Middleware 状态做了一层便利封装。

阅读开源项目

突然想到我的这个本来就是 vibe coding 的项目，可能在代码质量上面不足，应该直接学习开源的被大众认可的高质量的项目，看他们的工程化和代码。

ardanlabs-service

这个项目是 GPT 推荐的一个项目，它的工程化做得比较好，也是领域驱动设计的。

整体架构设计分析

这个项目的整体架构是先按架构进行垂直分层，内部再按照领域业务进行内聚。 项目结构顶部有一层 api、foundation 等大文件夹，即基础的垂直分层，大致分为：

• 对外的接口层 • 业务层 • 基础设施层

这和我的 MemoFlow 项目在架构设计上有相通之处，但我的设计像是把它“旋转了 90 度”：

• 我的架构（MemoFlow）： 整体上先按照业务进行水平拆分。顶部是按照业务拆分的包（packages），比如 goal、task、reminder 等。在每个包内部，再按照不同的架构层级进行拆分，例如每个 package 内部包含 domain（领域层）、application（应用层）、infrastructure（基础设施层）和接口层。 • 该项目的架构： 先整体进行垂直分层（将接口、业务、基础设施作为顶层目录），再在内部按领域业务进行内聚。

顺便提一下，在这次学习和询问的过程中，我突然意识到一件事：DDD 架构中的 Presentation（展示层）并不是我之前理解的前端那种页面展示，而是更类似于接口不同的暴露形式，比如 CLI、RESTful 接口或 gRPC 接口，类似我现在搞的接口层/Router 层。

总之，它的整体架构和 MemoFlow 比较相似，不过里面有一些工程细节需要进一步学习。

──────────

内部细节分析：以“创建用户”线路为例

这个项目适用于 K8s 微服务场景。它在 api 目录下存放的是它的 runtime 层（即组装层/组合根）。

作为一个微服务项目，它会在这里分别建立并拆分出各个服务，比如：

• auth（认证服务） • sales（销售业务服务）

在具体的服务（如 sales 服务）内部，进行具体业务的组装。它会需要联合几个子模块，比如将用户模块、商品模块等组合到一起。

具体的组装流程：

1. 初始化基础设施： 比如数据库。
2. 注入基建： 数据库等基础设施就绪后，创建并传入这些基建。
3. 创建服务： 基于基建创建具体的业务服务。
4. 包装 Handle 与中间件： 创建服务之后，创建 Handler、中间件等，进行一层一层的包装。
5. 挂载运行： 拿到所需的所有组装件后，开始挂载并运行服务。

package userapp

import (
	"net/http"

	"github.com/ardanlabs/service/app/sdk/auth"
	"github.com/ardanlabs/service/app/sdk/authclient"
	"github.com/ardanlabs/service/app/sdk/mid"
	"github.com/ardanlabs/service/business/domain/userbus"
	"github.com/ardanlabs/service/foundation/logger"
	"github.com/ardanlabs/service/foundation/web"
)

// Config contains all the mandatory systems required by handlers.
type Config struct {
	Log        *logger.Logger
	UserBus    userbus.ExtBusiness
	AuthClient authclient.Authenticator
}

// Routes adds specific routes for this group.
func Routes(app *web.App, cfg Config) {
	const version = "v1"

	authen := mid.Authenticate(cfg.AuthClient)
	ruleAdmin := mid.Authorize(cfg.AuthClient, auth.RuleAdminOnly)
	ruleAuthorizeUser := mid.AuthorizeUser(cfg.AuthClient, cfg.UserBus, auth.RuleAdminOrSubject)
	ruleAuthorizeAdmin := mid.AuthorizeUser(cfg.AuthClient, cfg.UserBus, auth.RuleAdminOnly)

	api := newApp(cfg.UserBus)

	app.HandlerFunc(http.MethodGet, version, "/users", api.query, authen, ruleAdmin)
	app.HandlerFunc(http.MethodGet, version, "/users/{user_id}", api.queryByID, authen, ruleAuthorizeUser)
	app.HandlerFunc(http.MethodPost, version, "/users", api.create, authen, ruleAdmin)
	app.HandlerFunc(http.MethodPut, version, "/users/role/{user_id}", api.updateRole, authen, ruleAuthorizeAdmin)
	app.HandlerFunc(http.MethodPut, version, "/users/{user_id}", api.update, authen, ruleAuthorizeUser)
	app.HandlerFunc(http.MethodDelete, version, "/users/{user_id}", api.delete, authen, ruleAuthorizeUser)
}

我觉得它的路由层和接口层做得很好。它封装了一个 HandleFunc 方法，直接将请求方法、路由和对应的 Handler 传进去，包装得非常优雅。这样一来，请求方法、路由路径以及需要调用的 Handler，还有前面需要经过哪些中间件处理，都表达得非常清晰。

比如在查询或创建 User 时，它会先进行 Auth 中间件处理，之后进入 API Handler 层。这一层主要负责 HTTP 数据处理，从 HTTP 中获取 JSON 数据，进行简单的转换和错误校验，拿到正确的数据后再交给业务层服务去处理。最终返回时，它也会把数据转回 JSON 格式。而且，它的 HandleFunc 似乎也封装了返回数据，在 Handler 中直接 return 即可，它会自动处理 HTTP 状态码等封装，后续就不需要再额外处理了。

业务层写起来比较基础，类似常见的写法，按业务逻辑处理，比如生成密码、保存用户等。

这里有一个比较特殊、也是我第一次见到的设计，就是它的仓储层（Store）分为了两层：一层缓存层，一层真实的数据库。对于业务层来讲，它并不需要知道缓存层的存在，调用时就像正常使用一样，通过传入的依赖调用 Store 里的 Create 等方法。但在初始化封装时，它是先传入基础数据库实例，创建一个缓存层，再把缓存层传给 Service。所以对 Service 来说，它当前使用的 Store 其实是一层缓存层。在缓存层里，它会先尝试读取缓存数据，有的话直接返回，没有再去查真实的数据库。

关于数据结构体的定义，它有一些文件专门保存，有的则直接写在业务文件中。比如 UserService.go 里，就会有它本身的 Service 接口，以及它所依赖的 Repository（仓储）接口，明确需要实现哪些方法。此外，Store 层会定义 Store 的结构体，并且它处理异步时有一个特点：通过 Context 中的属性（比如是否为事务）来判断，如果是事务的话，就不会走缓存。

关于中间件，虽然在 Express 框架里也是重要概念，但这里的实现让我有了更进一步的了解。比如鉴权部分，它分了很多步：

1. 通过中间件从 Bearer Token 中拿到个人信息，并存入 Context 上下文。
2. 如果是创建或修改用户，会再加一个中间件，去获取目标用户并校验是否有权限修改。
3. 还有基础的埋点类中间件，用来查看请求状态等。

它把这些逻辑连接起来，处理得非常优雅。

Memos

这个项目也是一个 GitHub 上面开源的高星大项目，基于 React 和 Go，并且它有了比较丰富的更新迭代的经验吧。 它是更接近前端 React 加上后端 Go 的架构。

它的大致结构和整个架构会有比较明显的变化：

1. 它有一个比较巨大的契约层（Proto）： 这是它比较独特的一个点。这个 Proto 进行了比较大的封装，确保了前端 TS 和后端 Go 都有相同的契约。
2. 传统的层级： 包括 Service、Store、Internal（业务层级、存储以及一些基础的工具设施），还有一个 Web 层。

它最独特的点在于，不仅通过 Proto 确保了前后端的定义统一，还对接口层（Transport 层）进行了彻底的封装（包含 gRPC 或者 RESTful 风格的接口），将其完全封装隐藏了。

在后续开发的时候，前端直接发起一个请求，后端跳转到 Go 后端代码时，由于接口层已经被封装起来，就会直接跳转到后端的业务层。

这有点类似我的 MemoFlow 项目中的 Contract（契约层），但它做得更多：我的契约层只是确保前后端的定义统一，而它不仅确保了定义统一，还把中间的 Transport 层也给封装了

python

学习目标

1. ai application（重点）
2. python 工程化（如何写代码，如何组织代码，主要是基础的语法和独特的接口， 代码组织应该和传统后端差不多）

当前理解（通过---来记录更新的理解）

当前只知道可以只接入ai能力，大致是 ai(deps): structured data

AI 内部主要有两种工作模式：

1. Workflow：按照一定的流程，有规划好、有顺序的节点去执行。
2. Agent：更偏向于让 AI 在运行时及时地规划自己后续要干什么，一个关键词是 ReAct。

总之，它们最后都需要返回结构化的数据，这是两种大致的工作模式。

至于这两个模式内部的其他工作细节，

1. 调用 LLM 大模型，通过大模型去进行迭代、循环或者进入下一步。
2. prompt 提示词， 通过提示词去约束大模型的输出
3. 大模型本身只能生成文本、输出内容，这时候就需要接入一些 Function Tool（工具）让大模型去调用。这个调用过程叫 Function Calling，通过它来给大模型提供更多的能力。后续让 AI 程序选择工具，以及进行业务层的数据持久化，也都是通过 Function Calling 来实现的。

AI 层负责“不确定性的工作”：理解自然语言、推理、生成结构化候选数据、选择工具、规划执行步骤。
业务后端负责“确定性的工作”：身份、权限、业务规则、事务、状态机、一致性和持久化。

---

在学习 Pydantic AI 的过程中，ChatGPT 给了我一个专题，让我去边做边学。我突然想在我的 bodysense 项目中边做边学，却发现原来我用的是 LangGraph 框架，并没有用 Pydantic AI。

于是我问 AI，这个 LangGraph 是否也可以拿来学习，是不是也是一个比较好的学习资料？但结果并不是。LangGraph 的 workflow 做的比较好，但是它的代码可能并不适合初学者去学习，不适合作为一个 Python 基础写法之类的学习材料。

那我就想看看其他推荐的几个项目是否有用 Pydantic AI 的，或者说，我应该拿什么更适合作为学习的材料？

和 AI 一顿对话后，我更明确了下面这几点知识： 首先，就是它这些 Pydantic AI 或者说 LangGraph 的框架，它不是最基本的抽象，它本质上是已经对基本抽象的一些实现。

如果把所有 LangGraph、PydanticAI、LangChain、Dify、Langflow、OpenHands 这些名字全部擦掉，我认为一个现代 AI Application 最核心可以拆成：

                     AI Application

                           │
                           ▼

                 Application / Use Case
                           │
                           ▼
                  Workflow / Agent
                           │
                    ┌──────┴──────┐
                    ▼             ▼
                  Nodes          Tools
                    │             │
                    └──────┬──────┘
                           ▼
                    Context / State
                           │
                           ▼
                      Model API
                           │
                           ▼
                       Provider
                           │
                           ▼
                         Model

然后有另一套东西横着覆盖它：

Runtime
Persistence
Checkpoint
Streaming
Retry
Concurrency
Observability
Evals
Governance
Security

这一点非常重要：

AI 系统不是一个纯粹的纵向“七层架构”。

它更像：

纵向能力层 + 横向运行时能力。

---

ai应用的抽象（分层）

model

底层的大模型 gpt、gork、deepseek等

provider

例如：

OpenAI
Anthropic
Google
Azure OpenAI
AWS Bedrock
Ollama
OpenRouter
Groq

这些是 Provider。

严格来说还应该再区分：

Model
      GPT-5.x

Provider
      OpenAI

API Adapter
      OpenAIModel

Gateway
      LiteLLM / 自己的 ModelRouter

Provider 层主要处理：

认证
HTTP API
请求格式
响应格式
stream protocol
rate limit
retry
usage
模型名称
provider-specific parameters

因此好的 AI Application 不应该出现：

# 到处都是
openai.responses.create(…)

而应该：

Application
     ↓
Model abstraction
     ↓
Provider adapter
     ↓
OpenAI / Anthropic / Gemini

PydanticAI 当前就明确提供 Models & Providers 这一层，并把 model-agnostic 作为核心目标之一。

model interface/ contract

非常重要的一层。

因为 Provider 上面通常还应该有：

Messages
Tool Schema
Output Schema
Events
Usage
ModelRequest
ModelResponse

例如：

class ConsultationResult(BaseModel):
    diagnosis: str
    confidence: float
    needs_more_info: bool

实际上你是在告诉 AI：

不要随便吐一坨字符串

我要的是：

ConsultationResult

这一层就是：

AI 世界里的 Contract Layer。

很像传统后端里的：

Request DTO
Response DTO
Command
Query
Domain Event

所以 Pydantic 在 AI 项目里为什么特别重要？

因为 LLM 天生是：

概率性的
非结构化的
不稳定的

而生产系统需要：

确定 schema
类型
validation
error handling
contracts

二者之间必须有一个“收口层”。

context engineering

这可能是传统后端里最没有直接对应物的一层。

以前大家喜欢叫：

Prompt Engineering

现在更准确的概念其实是：

Context Engineering

因为送给模型的东西远远不只是 prompt。

可能是：

System Instructions
        +
User Message
        +
Conversation History
        +
User Profile
        +
Workflow State
        +
Retrieved Documents
        +
Memory
        +
Tool Descriptions
        +
Current Time
        +
Domain Rules
        +
Previous Tool Results
        ↓
       LLM

所以：

Prompt
只是 Context 的一部分

RAG

其实属于 Context 层，而不是 Agent 的本质

例如：

用户：
“我上个月体重有什么变化？”

        ↓

Retrieval
        ↓
PostgreSQL / Vector DB
        ↓
取出相关记录
        ↓
Context Builder
        ↓

用户问题
+
相关记录

        ↓
LLM

这就是 RAG 最本质的事情：

在 inference 前动态构造 context。

所以不要把 RAG 神秘化。

从架构角度：

RAG =
Retrieval
+
Context Construction

当然生产 RAG 后面还有：

chunking
embedding
index
reranking
citation
document ingestion

但在 Agent runtime 看来，它最终是在解决：

给模型什么上下文？

memory 和 state

这是 AI 系统里特别容易混乱的一件事。

实际上至少存在三类东西。

Conversation State

User: …
Assistant: …
Tool: …
Assistant: …

即 message history。

---

Workflow State

例如 BodySense：

class ConsultationState:
    user_id: str
    symptoms: list[str]
    assessment: Assessment | None
    waiting_for_user: bool
    tool_results: list[ToolResult]

这是：

当前 workflow 执行到了什么状态。

LangGraph 特别强调这一类 shared State；其 Graph API 把 State、Nodes、Edges列为三个核心 graph primitive。

---

Long-term Memory

例如：

用户偏好
过去 consultation
健康记录摘要
长期目标
历史事实

它可能来自：

Postgres
Redis
Vector DB
Memory Store

所以：

Messages ≠ State ≠ Memory

虽然框架有时会把它们包装到一起。

tool

AI 获得“行动能力”的核心抽象

就是提到的 Function Calling。

但这里要做一个很重要的区分：

Tool
≠
Function Calling

Tool 是能力

例如：

get_health_record()
search_medical_knowledge()
schedule_followup()
calculate_bmi()
send_notification()

Tool 本质上就是：

AI 可以使用的一项外部 Capability。

完整 Tool 往往包含：

name
description
input schema
output schema
executor
authorization
validation
timeout
retry
approval policy

---

Function Calling 是协议

Function Calling 实际上只是：

LLM：
“我希望调用 get_health_record，
参数是 {user_id: 123}”

模型并没有真的执行 Python。

真实流程是：

LLM
 │
 │ tool_call
 ▼
Runtime
 │
 │ validate
 ▼
Tool Executor
 │
 │ execute
 ▼
Database/API
 │
 │ result
 ▼
Runtime
 │
 │ tool_result
 ▼
LLM

所以：

Function Calling 是 Model ↔ Application 之间表达 Tool Invocation Intent 的协议。

Tool 才是实际能力。

PydanticAI 当前也明确把 Function Tools / Toolsets / Deferred Tools / Native Tools 作为独立的一整组能力。

node

Node 其实并不是真正 AI 独有的概念。

Node 更准确的定义是：

Workflow 中最小的、可独立调度的 Execution Unit。

例如：

┌──────────────┐
│ CollectInfo  │
└──────┬───────┘
       ↓
┌──────────────┐
│   Assess     │
└──────┬───────┘
       ↓
┌──────────────┐
│   Decide     │
└──────┬───────┘
       ↓
┌──────────────┐
│ Finalize     │
└──────────────┘

Node 可以是什么？

几乎什么都可以：

Node
 ├─ 普通 Python function
 ├─ LLM call
 ├─ PydanticAI Agent
 ├─ RAG retrieval
 ├─ Tool execution
 ├─ HTTP request
 ├─ human approval
 ├─ sub-workflow
 └─ database transaction

所以：

Node 不是“AI 功能”。Node 是 orchestration unit。

agent

Agent 最核心的抽象其实是：

Observe
   ↓
Think / Decide
   ↓
Act
   ↓
Observe result
   ↓
Decide again
   ↓
…

用伪代码表示：

while not done:
    context = observe()

    decision = model(context)

    if decision.tool_call:
        result = execute_tool(decision.tool_call)
    else:
        return decision.output

这就是一个最基本的 Agent Loop。

PydanticAI 当前官方也明确说它自己“ships the agent loop”，并围绕 agent、tools、dependencies、outputs、capabilities 等构建。

所以：

Agent
=
Model
+
Context
+
Tools
+
Decision Loop

Workflow

Workflow 才真正开始进入 orchestration。

Workflow 描述：

Node
Edge
State
Branch
Loop
Parallel
Join
Interrupt
Human approval
Subflow

例如 BodySense：

START
  ↓
collect_context
  ↓
assess
  ↓
needs_more_info?
 ┌───────┴────────┐
yes               no
 ↓                 ↓
ask_user          plan
 ↓                 ↓
interrupt         tools
 ↓                 ↓
resume          finalize
 └──────────┬──────┘
            ↓
           END

这就是：

Workflow Definition。

LangGraph 非常典型：它明确把 workflow 建模为 State + Nodes + Edges。

Runtime

Workflow 只是：

定义应该怎么跑。

Runtime 才是：

真正把它跑起来。

这是一个特别重要的区别：

Workflow Definition
        ↓
       Runtime
        ↓
Actual Execution

就像：

Python code
   ≠
Python interpreter

类似地：

Workflow
   ≠
Workflow Runtime

Runtime 负责的东西非常多：

Node scheduling
state propagation
concurrency
parallel execution
retry
timeout
cancellation
streaming
checkpoint
resume
interrupt
durable execution
task lifecycle
event emission
error handling

LangGraph 官方现在甚至直接把自己描述成 orchestration framework and runtime，并特别强调 durable execution、streaming、HITL 和 persistence。

所以你之前 BodySense 的：

checkpointing.py
runtime_event
ask_user
resume
governance
thread

其实已经不是“Agent 代码”了。

它们属于：

阅读开源项目

Pydantic AI

学习优雅的 Python AI 写法，现代 Python 如何把 LLM 变成可靠软件能力

Python
  ↓
如何描述 AI 能力
  ↓
如何把不确定的 LLM 包装成确定的软件接口
  ↓
如何支持多个模型
  ↓
如何调用 Tool
  ↓
如何做 structured output
  ↓
如何 streaming
  ↓
如何保存状态
  ↓
如何测试
  ↓
如何 eval
  ↓
如何变成生产级 AI Service

专题一

不看源码解释下面整个过程：

BodySenseDeps
      │
      │ 类型
      ▼
RunContext[BodySenseDeps]
      │
      │
      ▼
get_recent_workouts(
    ctx,
    days: int
)
      │
      │ inspect.signature + type hints
      ▼
FunctionSchema
      │
      ├──────────────┐
      │              │
      ▼              ▼
SchemaValidator   JSON Schema
                     │
                     ▼
                  LLM API
                     │
                     │
                 {"days": 7}
                     │
                     ▼
              runtime validation
                     │
                     ▼
                 days = 7
                     │
        RunContext ──┤
                     ▼
         get_recent_workouts(
             ctx,
             days=7,
         )
                     │
                     ▼
               list[Workout]

Onyx

Langflow

重构（bodysense 的一条支线（diagnosis）中学习）（跨三端）

决定直接在实际项目的重构中学习。 ai 结合了 pydanticAI，langGraph，langflow 帮我创建了一个重构计划

阶段	你学什么	BodySense 实际改什么
0	阅读已有系统	Characterization tests + 调用链
1	Pydantic / typing	Diagnosis、Symptom、Context contracts
2	PydanticAI Agent	重构 DiagnosisService
3	Structured Output	删除 diagnosis json.loads() 路径
4	Deps / Context	DiagnosisDeps + DiagnosisContextBuilder
5	PydanticAI Tools	search_knowledge、get_posture_analysis
6	Model / Provider	分离 ModelPolicy 和 provider implementation
7	LangGraph State	重构 ConsultationThreadState
8	Node / Workflow	拆 consultation_thread.py
9	HITL Runtime	独立 AskUserNode + interrupt/resume
10	Tool taxonomy	extraction 不再做 Tool
11	Langflow 思想	typed/inspectable node contract
12	Evals	Diagnosis dataset + routing eval
13	Observability	agent/tool/node tracing

DMR-001

摘要

不是为了证明“新设计正确”，而是为了把旧 Python Diagnosis 边界钉死。

简单讲就是补充一系列测试文件，确保在重构的时候不会意外改动，导致意外的变化。

比如：

• 默认值的表现
• 错误的传递的表现

主要学习内容

学习 python 的测试文件（pytest 框架）怎么写：
测试函数 test_ 前缀+三个步骤（AAA模式）

class _FakeDiagnosisService: def __init__(self, result=None, error: Exception | None = None):
        self.result = result
        self.error = error
        self.captured = None

    async def generate_diagnosis(self, **kwargs):
        self.captured = kwargs
        if self.error is not None:
            raise self.error
        return self.result

# client 和 monkeypatch 是 pytest 提供的依赖注入，交给 pytest 处理
def test_analyze_diagnosis_preserves_http_request_and_response_contract(client, monkeypatch):
    expected = {
        "diagnoses": [
            {
                "name": "头前伸倾向",
                "confidence": "中",
                "severity": "轻度",
                "basis": "久坐后颈肩酸胀",
                "typical_symptoms": "颈肩酸胀",
            }
        ],
        "citations": [{"title": "头前伸自测"}],
        "governance": {
            "verdict": "accepted",
            "kind": "diagnosis",
            "reasons": [],
            "issues": [],
        },
    }
    # Arrange（准备） 造假对象、预期结果、注入替身
    fake = _FakeDiagnosisService(result=expected)
    monkeypatch.setattr(
        "src.api.routes.diagnosis.get_diagnosis_service",
        lambda: fake,
    )

    payload = {
        "extracted_info": [{"body_part": "颈椎", "symptom_type": "酸胀"}],
        "profile": {"age": 30, "occupation": "程序员"},
        "conversation_summary": "久坐后颈肩酸胀",
        "rag_context": "## 知识库\n头前伸相关资料",
        "rag_results": [{"title": "头前伸自测"}],
        "use_case": "llm.json",
    }

  
	# Act（执行）： 调用被测对象
    response = client.post("/api/diagnosis/analyze", json=payload)

	# Assert（断言）：验证行为
    assert response.status_code == 200
    assert response.json() == expected
    assert fake.captured == payload

DMR-002

摘要

go 中的测试

用户请求“生成诊断”
        │
        ▼
DiagnosisHandler
        │
        ▼
检查 consultation 是否存在
        │
        ▼
读取 profile / extracted info
        │
        ▼
调用 Python Diagnosis AI
        │
        ▼
检查 governance
        │
        ├── rejected
        │      └── 不保存 diagnosis
        │          不推进 phase
        │
        └── accepted
               └── 保存 diagnosis
                   phase -> analysis_ready

未来改 Diagnosis 业务逻辑之前，先用自动化测试把这几个重要行为固定下来。

上面的 Diagnosis 业务比较复杂。在进行诊断前，它大致需要满足以下几个步骤：

1. 确保当前的会话存在
2. 读取当前的会话信息（即会话诊断的结果信息）
3. 获取用户的私人信息
4. 调用一个 AI Service（即诊断服务）

以上就是诊断服务内部的流程，整体确实比较复杂。

如果我们要进行测试，且所有环节都走真实流程的话，测试就会变得非常重——需要准备真实的数据库环境、服务环境等，最终演变成一个比较重的集成测试或端到端测试。因此，我们需要对这个测试进行优化。

大致的优化内容和结果如下：

					Test
                     │
                     ▼
               Gin Context
                     │
                     ▼
           REAL DiagnosisHandler
                     │
          ┌──────────┼───────────┐
          ▼          ▼           ▼
 REAL Consultation  REAL       REAL
     Service      Profile     AIClient
       │          Service        │
       ▼             │           ▼
 Fake Repository   GORM       HTTP
                     │           │
                  sqlmock    httptest.Server

注意：

Handler 是真的。 ConsultationService 也是真的。 AIClient 也是真的。

我们只把真正昂贵、不稳定、外部的部分替换掉：

PostgreSQL → Fake / sqlmock  
Python AI Service → httptest.Server

所以这个测试既不是纯 Unit Test，也不是完整 E2E。 更准确地说，它是：

Handler / Application Boundary Test

同时它的主要目的又是：

Characterization Test

对应“三条轴”：

层级 / scope  
→ Handler/Application boundary

目的 / purpose  
→ Characterization

手段 / technique  
→ Fake + Stub + sqlmock

主要学习内容

根据上面的要求，去给 DiagnosisHandler.handleHTTP 这一层级写测试文件，然后要把仓储层 mock 一下，构建一个测试用的仓储层(Fake Repo)： 直接创建一个 diagnosisHandlerRepo struct，然后实现所有 真实 ConsultationSession 的方法

[!note] 为什么这个 diagnosisHandlerRepo 却是实现 consultationSession 的方法 其实当前 diagnosis 就是归 consultationSession 管理。 从设计角度，这其实也告诉了我们一件事：consultationRepository interface 有点宽。 后续应该拆分。

使用依赖的对象，不负责创建它的依赖。

[!question] 拆分疑问 拆分后这个测试文件会大改吗 是的话，是否应该先设计（重构）好领域，再搞测试文件

实际上没有进行手搓代码， 二是让 ai 实施了，然后先进行一个阅读学习。发现了一下问题：

1. 当前的 diagnosis 服务 没有时间比较好的依赖注入，写死了 ai.service 之类的参数，导致测试比较困难。应该在后续重构为更优雅的依赖注入模式

DMR-100

摘要

临时添加的，进行上面提到的 diagnosis service 的重构，让 diagnosis 不自己构建 aiservice，二是通过组合根创建爱你

主要学习内容

1. 在 diagnosis service 的 def init 函数中添加 aiservice 参数
2. python 的 protocol （类似go中的interface）
3. 依赖注入和依赖倒置

工程化一点可以叫 consumer-owned interface + Interface Segregation：

• consumer-owned：接口由使用方需求决定；
• interface segregation：只暴露使用方真正需要的最小方法集合；
• dependency inversion：高层 DiagnosisService 依赖抽象 AIExecutor，低层 AIService 去满足这个抽象。

DMR-101

摘要

为什么现在 DiagnosisItem / DiagnosisResponse 这些 Pydantic Model 还不够，它们与真正的 typed domain models 有什么区别

大致就是，目前是使用了一个 base model 来定义 diagnosis item 的一些属性，也就是它长什么样，但并不足以把这个领域对象给清楚建模。

比如说它的 severity（严重程度），当前只是简单的 string（字符串）字段。但真正建模好的话，应该限制它的字段值，一般是作为一个枚举（比如说中、低、高之类的）。现在这个就完全没有清楚建模。

主要学习内容

• 领域建模
• 用 @dataclass(slots=True) 把 DiagnosisDependencies 变成一个“字段固定的数据对象”

[!importance] 一个很重要的能力 不要因为某个参数恰好出现在同一个函数里，就认为它们属于同一个 context。 要看 ownership 和生命周期。

DX-001

摘要

• AIExecutor(Protocol) 为什么存在？
• constructor dependency injection 是什么？
• PydanticAI deps_type 为什么不是 constructor DI 的替代品？（“既然 PydanticAI 已经提供了一套 dependency injection 机制，我们为什么还需要让 DiagnosisService 使用 constructor DI？”）
• 为什么 output_type=DiagnosisAgentOutput 后，可以删掉手工 JSON parsing？

主要学习内容

第一个和第二个问题： DiagnosisService 不关心“你到底是不是 AIService”，只关心“你有没有我需要的 generate 能力”。典型的 Dependency Inversion + constructor injection。 而 AIExecutor(Protocol) 就是能够描述这个能力的接口。所以，给 DiagnosisService 传入这个作为参数，然后后续在实际构造的时候就可以传入任何符合这个接口的实现。

第三个问题： 因为本质上讲，Hydrantic AI 提供的 Depstype 和 Constructor DI 不是同一个层级的，它们注入的是依赖关系图里的两条不同的边。Constructor DI 上面是在 Application level 去使用的。然后，Pydantic 的这个 dependency injection 机制是单纯地在 agent 里使用的。差不多就是 constructor DI 决定了我要使用长什么样的 agent，然后 agent 的里面的 depth type 决定了 agent 想要长什么样的依赖。

第四个： output type 这个参数是在告诉 Pydantic AI 这个 agent 的最终合法输出类型，就是 diagnosis agent output。Pydantic AI 会基于这个类型建立 structured output schema，并验证模型返回的数据，最后确保返回的数据是 diagnosis agent output 类型。

写了一个 Agent 能力的接口，也就是搜索能力,用于后面给 diagnosis service 实现一个依赖注入。

from __future__ import annotations # python 现代注解库， 比如 ：str ：int 这些类型注解
from typing import Protocol, Any # python 中的类型

class EvidenceSearcher(Protocol):
    async def search(self, query: str, *, top_k: int = 5) -> list[dict[str, Any]]: …

DMR-005

摘要

主要学习内容

领域拆分

工程化

react

go

python