M01 Go 语言 AI 开发基础

ai-agent/
├── cmd/                    # 可执行入口（每个子目录是一个 binary）
│   ├── server/             # HTTP API 服务
│   ├── agent/              # 命令行 Agent
│   └── ingest/             # 文档摄取工具
│
├── internal/               # 私有包，外部无法 import
│   ├── llm/                # LLM Provider 封装
│   │   ├── provider.go     # 统一接口定义
│   │   ├── openai.go       # OpenAI 实现
│   │   └── stream.go       # 流式处理
│   ├── agent/              # Agent 核心逻辑
│   ├── tools/              # 工具系统
│   ├── memory/             # 记忆与 RAG
│   └── transport/          # HTTP 客户端、重试、SSE 解析
│
├── pkg/                    # 可对外暴露的通用工具包
│   └── retry/              # 通用重试逻辑
│
├── go.mod                  # 模块定义
├── go.sum                  # 依赖哈希锁定
├── Makefile                # 常用命令
└── .env.example            # 环境变量模板

# 创建项目
mkdir ai-agent && cd ai-agent

# 初始化 Go module（即使不发布到 GitHub，也用这个格式）
go mod init github.com/yourname/ai-agent

# 安装核心依赖
go get github.com/sashabaranov/go-openai  # OpenAI SDK

# Makefile

.PHONY: run build test lint clean

# 启动开发服务器
run:
	go run ./cmd/server/

# 构建所有二进制
build:
	CGO_ENABLED=0 go build -ldflags="-w -s" -o bin/server ./cmd/server/
	CGO_ENABLED=0 go build -ldflags="-w -s" -o bin/agent  ./cmd/agent/

# 运行测试（含竞态检测，这个习惯要从第一天养成）
test:
	go test -race -count=1 -coverprofile=coverage.out ./...

# 查看测试覆盖率
cover:
	go tool cover -html=coverage.out

# 代码检查
lint:
	golangci-lint run ./...

# 清理构建产物
clean:
	rm -rf bin/ coverage.out

同步方式：发请求 → 等待（3-8秒） → 收到完整回复 → 显示
流式方式：发请求 → 立刻收到第一个 token → 实时显示 → ... → 收到结束标记

func StreamTokens(ctx context.Context, ...) <-chan Token {
    tokens := make(chan Token, 64)

    go func() {
        defer close(tokens)
        // 读取并发送 token
    }()

    return tokens
}

// StreamChunk 是流式输出的最小单元。
// 约定：Content 与 Err 互斥；Err != nil 时这是最后一个有意义的块。
type StreamChunk struct {
	Content string // 本次增量文本
	Err     error  // 出错时非空
}

// fakeStream 模拟一个会逐字吐字的模型，用于教学/测试。
func fakeStream(ctx context.Context, text string) <-chan StreamChunk {
	out := make(chan StreamChunk)
	go func() {
		defer close(out) // 关键：无论如何都要关 channel，否则消费者死等
		for _, r := range text {
			select {
			case <-ctx.Done(): // 上游取消/超时，立刻收手并把原因带出去
				out <- StreamChunk{Err: ctx.Err()}
				return
			case out <- StreamChunk{Content: string(r)}:
				time.Sleep(30 * time.Millisecond) // 模拟生成耗时
			}
		}
	}()
	return out
}

func consume(ctx context.Context) error {
	for chunk := range fakeStream(ctx, "你好，世界") {
		if chunk.Err != nil {
			return chunk.Err
		}
		print(chunk.Content) // 逐字打印
	}
	return nil
}

HTTP Handler → Agent.Run → LLM.Complete → HTTP 请求模型服务

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel() // 重要：即使超时前完成，也要释放资源

ctx, cancel := context.WithCancel(parent)
// 某个条件触发后主动 cancel()
go func() {
    <-userClosed
    cancel()
}()

ctx, cancel := context.WithCancelCause(parent)
cancel(fmt.Errorf("用户取消了对话"))
// 下游：context.Cause(ctx) 取出真正原因

type contextKey string
const requestIDKey contextKey = "request_id"

ctx = context.WithValue(ctx, requestIDKey, "req-123")

// 在任何地方取出
reqID := ctx.Value(requestIDKey).(string)

func (a *Agent) Run() (string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    return a.callLLM(ctx) // 传入了，但如果上层的 ctx 已取消，这个新 ctx 不受影响
}

func (a *Agent) Run(ctx context.Context, goal string) (string, error) {
    // 每次 LLM 调用单独设置超时，但基于父 ctx 派生
    // 这样父 ctx 取消时，子 ctx 也会立刻取消
    for step := 0; step < a.maxSteps; step++ {
        // 检查父 ctx 是否已取消
        if err := ctx.Err(); err != nil {
            return "", fmt.Errorf("agent cancelled: %w", err)
        }

        // 派生子 ctx，单次 LLM 调用不超过 30 秒
        llmCtx, llmCancel := context.WithTimeout(ctx, 30*time.Second)
        resp, err := a.llm.Complete(llmCtx, req)
        llmCancel() // ← 注意：不用 defer，在循环里 defer 会积累！
        
        if err != nil {
            return "", err
        }
        // ...
    }
}

if err := ctx.Err(); err != nil {
    return err
}

select {
case result := <-resultChan:
    return result, nil
case <-ctx.Done():
    return nil, ctx.Err()
}

req, err := http.NewRequestWithContext(ctx, "POST", url, body)

// ❌ 不可维护：每增加一个 provider，Agent 层都要修改
func (a *Agent) callLLM(ctx context.Context, messages []Message) (string, error) {
    switch a.providerType {
    case "openai":
        return a.openaiClient.Complete(ctx, messages)
    case "claude":
        return a.claudeClient.Complete(ctx, messages)
    case "doubao":
        return a.doubaoClient.Complete(ctx, messages)
    // 每加一个 provider 就要改这里...
    }
}

// Provider 是所有大模型供应商的统一抽象。
// 接口刻意保持最小：先跑起来，能力随模块增长再扩展。
type Provider interface {
    // Name 返回 Provider 标识符，用于日志和监控
	Name() string

    // Chat 发送请求并等待完整响应
	Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error)

    // ChatStream 发送请求并通过 channel 返回流式 token
	ChatStream(ctx context.Context, req ChatRequest) (<-chan StreamChunk, error)
}

type Role string

const (
	RoleSystem    Role = "system"
	RoleUser      Role = "user"
	RoleAssistant Role = "assistant"
	RoleTool      Role = "tool"
)

type Message struct {
	Role    Role   `json:"role"`
	Content string `json:"content"`
}

type ChatRequest struct {
	Model       string    `json:"model"`
	Messages    []Message `json:"messages"`
	Temperature float64   `json:"temperature,omitempty"`
	MaxTokens   int       `json:"max_tokens,omitempty"`
	Stream      bool      `json:"stream,omitempty"`
}

type ChatResponse struct {
	Content      string
	InputTokens  int
	OutputTokens int
}

// ParseInto 把模型输出的 JSON 文本解析成目标类型 T。
// 用泛型避免每个调用点都写一遍 unmarshal + 类型断言。
func ParseInto[T any](raw string) (T, error) {
	var v T
	if err := json.Unmarshal([]byte(raw), &v); err != nil {
		return v, fmt.Errorf("解析为 %T 失败: %w", v, err)
	}
	return v, nil
}

// 用法：
//   type Weather struct{ City string; Temp float64 }
//   w, err := ParseInto[Weather](modelOutput)

// Ptr 返回任意值的指针，常用于构造带可选字段的请求体。
func Ptr[T any](v T) *T { return &v }

// 用法： req.Temperature = Ptr(0.7)

// Processor 是泛型流处理器，T 可以是任何类型
// 用法：stream.Process(ctx, tokenChan, func(tok Token) error { ... })
func Process[T any](
    ctx context.Context,
    ch <-chan T,
    handler func(T) error,
) error {
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case item, ok := <-ch:
            if !ok {
                return nil // channel 关闭，正常结束
            }
            if err := handler(item); err != nil {
                return err
            }
        }
    }
}

// Collect 将 channel 中的所有值收集到 slice
func Collect[T any](ctx context.Context, ch <-chan T) ([]T, error) {
    var result []T
    err := Process(ctx, ch, func(item T) error {
        result = append(result, item)
        return nil
    })
    return result, err
}

resp, err := http.Get("https://api.openai.com/v1/...")

// HTTPConfig 收拢 HTTP 客户端的可调项。
type HTTPConfig struct {
	DialTimeout         time.Duration // 建连超时
	KeepAlive           time.Duration
	MaxIdleConns        int
	MaxIdleConnsPerHost int // 默认只有 2，调模型时远远不够
	IdleConnTimeout     time.Duration
	TLSHandshakeTimeout time.Duration
}

// DefaultHTTPConfig 是一组适合调 LLM API 的稳妥默认值。
func DefaultHTTPConfig() HTTPConfig {
	return HTTPConfig{
		DialTimeout:         5 * time.Second,
		KeepAlive:           30 * time.Second,
		MaxIdleConns:        100,
		MaxIdleConnsPerHost: 20,
		IdleConnTimeout:     90 * time.Second,
		TLSHandshakeTimeout: 5 * time.Second,
	}
}

// HTTPOption 覆盖默认配置。
type HTTPOption func(*HTTPConfig)

func WithDialTimeout(d time.Duration) HTTPOption { return func(c *HTTPConfig) { c.DialTimeout = d } }
func WithMaxIdleConnsPerHost(n int) HTTPOption   { return func(c *HTTPConfig) { c.MaxIdleConnsPerHost = n } }
// 其余字段按同一模式按需添加 WithXxx。

// newTransport 把配置组装成 *http.Transport —— 与 client 构造分离，单一职责。
func newTransport(c HTTPConfig) *http.Transport {
	return &http.Transport{
		DialContext: (&net.Dialer{
			Timeout:   c.DialTimeout,
			KeepAlive: c.KeepAlive,
		}).DialContext,
		MaxIdleConns:          c.MaxIdleConns,
		MaxIdleConnsPerHost:   c.MaxIdleConnsPerHost,
		IdleConnTimeout:       c.IdleConnTimeout,
		TLSHandshakeTimeout:   c.TLSHandshakeTimeout,
		ExpectContinueTimeout: 1 * time.Second,
	}
}

// NewHTTPClient 返回一个适合调 LLM API 的客户端；不传参用默认值，也可用 Option 覆盖。
func NewHTTPClient(opts ...HTTPOption) *http.Client {
	cfg := DefaultHTTPConfig()
	for _, opt := range opts {
		opt(&cfg)
	}
	// 注意：不设 client.Timeout —— 流式(SSE)是长连接，整体超时会把它掐断；
	// 超时交给 context 逐次控制，这里只兜底拨号/握手等阶段。
	return &http.Client{Transport: newTransport(cfg)}
}

type RetryConfig struct {
	MaxRetries int           // 最多重试次数（不含首次）
	BaseDelay  time.Duration // 退避基数，如 500ms
	MaxDelay   time.Duration // 退避上限，如 10s
}

func DefaultRetryConfig() RetryConfig {
	return RetryConfig{
		MaxRetries: 3,
		BaseDelay: 500 * time.Millisecond,
		MaxDelay: 10 * time.Second,
	}
}

// Limiter 抽象限流器。标准库没有内置限流，生产中通常用 golang.org/x/time/rate.Limiter
// （它的 Wait(ctx) error 正好满足这个接口）。接口定义在使用方，transport 无需 import rate。
type Limiter interface {
	Wait(ctx context.Context) error
}

// Client 把"连接池 + 重试退避 + 可选限流"封装成一个可复用的客户端。
// Do 是唯一出站入口：重试与限流统一在这里执行，调用方无法绕过。
type Client struct {
	http    *http.Client
	retry   RetryConfig
	limiter Limiter // 可空：nil 表示不限流
}

type Option func(*Client)

func WithRetry(cfg RetryConfig) Option { return func(c *Client) { c.retry = cfg } }
func WithLimiter(l Limiter) Option     { return func(c *Client) { c.limiter = l } } // 如 rate.NewLimiter(10, 1)
func WithHTTPOptions(opts ...HTTPOption) Option {
	return func(c *Client) { c.http = NewHTTPClient(opts...) }
}

// NewClient 返回生产级 LLM HTTP 客户端；零参开箱即用，后面各章都用它。
func NewClient(opts ...Option) *Client {
	c := &Client{http: NewHTTPClient(), retry: DefaultRetryConfig()}
	for _, opt := range opts {
		opt(c)
	}
	return c
}

// Do 执行请求并按需重试 429/5xx（指数退避 + 抖动，优先尊重 Retry-After）。
// 签名刻意对齐标准库 http.Client.Do：ctx 通过 http.NewRequestWithContext 装进请求里。
func (c *Client) Do(req *http.Request) (*http.Response, error) {
	ctx := req.Context()
	if c.limiter != nil {
		if err := c.limiter.Wait(ctx); err != nil { // 限流：等到有令牌，或被 ctx 取消
			return nil, err
		}
	}
	var lastErr error
	for attempt := 0; attempt <= c.retry.MaxRetries; attempt++ {
		// 重试前重建一次性的 Body —— 否则 POST 重试会发空体。
		// http.NewRequest 对 bytes/strings reader 会自动设好 GetBody，这里直接用。
		if req.Body != nil && req.GetBody != nil {
			body, err := req.GetBody()
			if err != nil {
				return nil, err
			}
			req.Body = body
		}
		resp, err := c.http.Do(req)
		var wait time.Duration
		switch {
		case err != nil:
			lastErr = err // 网络层错误，可重试
			wait = backoff(attempt, c.retry)
		case resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500:
			wait = retryAfter(resp) // 优先尊重服务端的 Retry-After
			if wait <= 0 {
				wait = backoff(attempt, c.retry)
			}
			resp.Body.Close()
			lastErr = fmt.Errorf("服务端返回 %d", resp.StatusCode)
		default:
			return resp, nil // 2xx/4xx(非429) 直接返回，交给上层处理
		}

		if attempt == c.retry.MaxRetries { // 最后一次仍失败：不再等待，直接退出
			break
		}
		if !sleep(ctx, wait) {
			return nil, ctx.Err()
		}
	}
	return nil, fmt.Errorf("重试 %d 次后仍失败: %w", c.retry.MaxRetries, lastErr)
}

// backoff 计算第 attempt 次重试的等待时长：指数增长 + 抖动，且不超过 MaxDelay。
func backoff(attempt int, cfg RetryConfig) time.Duration {
	d := cfg.BaseDelay << attempt // 0.5s, 1s, 2s, 4s...
	if d > cfg.MaxDelay {
		d = cfg.MaxDelay
	}
	// 加抖动，让结果落在约 0.9~1.1 倍之间（±10%），避免大量客户端同时重试造成“惊群”
	jitter := time.Duration(rand.Int63n(int64(d) / 5))
	return d - (d / 10) + jitter
}

// retryAfter 解析 429/503 响应里的 Retry-After 头：支持「秒数」与「HTTP-date」两种形式。
func retryAfter(resp *http.Response) time.Duration {
	v := resp.Header.Get("Retry-After")
	if v == "" {
		return 0
	}
	if sec, err := strconv.Atoi(v); err == nil { // 形式一：延迟秒数，如 "3"
		return time.Duration(sec) * time.Second
	}
	if t, err := http.ParseTime(v); err == nil { // 形式二：HTTP-date
		if d := time.Until(t); d > 0 {
			return d
		}
	}
	return 0
}

// sleep 在等待 d 的同时监听 ctx；被取消则返回 false。
func sleep(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

resp, err := transport.NewClient().Do(req)

// retryTransport 是一层中间件式的 RoundTripper：先限流，再带重试地委托给底层 base。
type retryTransport struct {
	base    http.RoundTripper
	retry   RetryConfig
	limiter Limiter // 可空
}

func (t *retryTransport) RoundTrip(req *http.Request) (*http.Response, error) {
	ctx := req.Context()
	if req.Body != nil {
		defer req.Body.Close() // RoundTripper 负责关闭传入 Body；重试用 GetBody 取独立副本
	}
	if t.limiter != nil {
		if err := t.limiter.Wait(ctx); err != nil {
			return nil, err
		}
	}
	var lastErr error
	for attempt := 0; attempt <= t.retry.MaxRetries; attempt++ {
		// RoundTripper 契约：不应改动传入的 req（它可能被并发读）。
		// 每次重试都 Clone 一份、用 GetBody 重建一次性的 Body。
		r := req
		if req.Body != nil && req.GetBody != nil {
			body, err := req.GetBody()
			if err != nil {
				return nil, err
			}
			r = req.Clone(ctx)
			r.Body = body
		}
		resp, err := t.base.RoundTrip(r) // 委托给真正发请求的底层 transport
		var wait time.Duration
		switch {
		case err != nil:
			lastErr, wait = err, backoff(attempt, t.retry)
		case resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500:
			if wait = retryAfter(resp); wait <= 0 {
				wait = backoff(attempt, t.retry)
			}
			resp.Body.Close()
			lastErr = fmt.Errorf("服务端返回 %d", resp.StatusCode)
		default:
			return resp, nil
		}
		if attempt == t.retry.MaxRetries {
			break
		}
		if !sleep(ctx, wait) {
			return nil, ctx.Err()
		}
	}
	return nil, fmt.Errorf("重试 %d 次后仍失败: %w", t.retry.MaxRetries, lastErr)
}

// NewRetryHTTPClient 产出一个“自带重试/限流”的普通 *http.Client：调用方照常用即可。
func NewRetryHTTPClient(retry RetryConfig, limiter Limiter, opts ...HTTPOption) *http.Client {
	cfg := DefaultHTTPConfig()
	for _, opt := range opts {
		opt(&cfg)
	}
	return &http.Client{Transport: &retryTransport{base: newTransport(cfg), retry: retry, limiter: limiter}}
}

type toolCall struct {
	Name string          `json:"name"`
	Args json.RawMessage `json:"arguments"` // 先不解析，按工具类型再 ParseInto
}

// ContentPart 是多模态消息的一个分块（文本或图片）——“联合类型”的典型场景：
// 同一个字段在不同情况下是不同结构。
type ContentPart struct {
	Type     string    `json:"type"`                // "text" | "image_url"
	Text     string    `json:"text,omitempty"`      // type=="text" 时有值
	ImageURL *ImageURL `json:"image_url,omitempty"` // type=="image_url" 时有值
}

type ImageURL struct {
	URL string `json:"url"`
}

// MessageContent 兼容 content 字段的两种形态：纯文本字符串 或 多模态分块数组。
type MessageContent struct {
	Text  string        // 纯文本形态
	Parts []ContentPart // 多模态形态（图片、音频…）
}

func (m *MessageContent) UnmarshalJSON(data []byte) error {
	// 先试最常见的纯字符串
	var s string
	if err := json.Unmarshal(data, &s); err == nil {
		m.Text = s
		return nil
	}
	// 再试分块数组
	var parts []ContentPart
	if err := json.Unmarshal(data, &parts); err == nil {
		m.Parts = parts
		return nil
	}
	return fmt.Errorf("content 既不是 string 也不是数组: %s", data)
}

// 出站方向对称处理：有分块就发数组，否则发纯字符串——与上面的解析一一对应，能无缝往返。
func (m MessageContent) MarshalJSON() ([]byte, error) {
	if len(m.Parts) > 0 {
		return json.Marshal(m.Parts)
	}
	return json.Marshal(m.Text)
}

// SafeJSON 安全地序列化对象，失败时返回错误字符串而非 panic
func SafeJSON(v any) string {
    data, err := json.Marshal(v)
    if err != nil {
        return fmt.Sprintf(`{"error": "marshal failed: %v"}`, err)
    }
    return string(data)
}

package transport

import (
	"net/http"
	"testing"
	"time"
)

func TestRetryAfter(t *testing.T) {
	tests := []struct {
		name string
		val  string // Retry-After 头的值（空表示不设该头）
		want time.Duration
	}{
		{"无头", "", 0},
		{"正常秒数", "3", 3 * time.Second},
		{"非法值", "soon", 0},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			resp := &http.Response{Header: http.Header{}}
			if tt.val != "" {
				resp.Header.Set("Retry-After", tt.val)
			}
			if got := retryAfter(resp); got != tt.want {
				t.Errorf("got %v want %v", got, tt.want)
			}
		})
	}
}

package transport

import (
	"context"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

func TestClientDo(t *testing.T) {
	var hits int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if atomic.AddInt32(&hits, 1) == 1 {
			w.WriteHeader(http.StatusTooManyRequests) // 第 1 次：429，应触发重试
			return
		}
		w.WriteHeader(http.StatusOK) // 第 2 次：成功（GetBody 已把 POST body 重发一遍）
		_, _ = w.Write([]byte(`{"ok":true}`))
	}))
	defer srv.Close()

	cfg := DefaultRetryConfig()
	cfg.BaseDelay = time.Millisecond // 测试里别真等几百毫秒
	// ctx 装进请求；strings.Reader 让 http 自动设好 GetBody，重试时 Client 据此重建 Body
	req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, srv.URL, strings.NewReader(`{}`))
	if err != nil {
		t.Fatal(err)
	}

	resp, err := NewClient(WithRetry(cfg)).Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("期望最终 200，得到 %d", resp.StatusCode)
	}
	if hits != 2 {
		t.Fatalf("期望命中 2 次（429 后重试一次），实际 %d", hits)
	}
}

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"strings"
	"syscall"
)

func main() {
	// Ctrl+C → context 取消
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	question := strings.Join(os.Args[1:], " ")
	if question == "" {
		fmt.Println("用法: minicall <你的问题>")
		os.Exit(1)
	}

	cfg := loadConfigFromEnv() // TODO: 读取 BASE_URL / API_KEY / MODEL
	if err := runOnce(ctx, cfg, question); err != nil {
		fmt.Fprintln(os.Stderr, "\n出错:", err)
		os.Exit(1)
	}
}

// runOnce: 构造请求 → transport.NewClient().Do(req) → 解析 choices[0].message.content 与 usage → 打印。
// TODO: 由你实现，这是本模块的综合验收。
func runOnce(ctx context.Context, cfg Config, question string) error {
	panic("TODO")
}