调用 LLM（流式）→ 解析出 N 个 tool_call → 并行执行工具 → 汇总结果 → 下一轮
                     ↑ 任一环节都要：超时、取消、限流、不泄漏

go version                 # 确认安装，本仓库用 1.23+
mkdir hello-agent && cd hello-agent
go mod init github.com/duang777/hello-agent   # 初始化模块，生成 go.mod

package main // 可执行程序的入口包必须叫 main

import "fmt" // 引入标准库的 fmt 包（格式化输出）

func main() { // 程序从 main 函数开始执行
    fmt.Println("hello, agent")
}

go run .          # 编译并运行当前目录
go build -o app . # 编译出二进制文件 app，可以直接拷到服务器上跑
go get golang.org/x/sync@latest   # 引入第三方库，自动写进 go.mod / go.sum
go mod tidy       # 清理没用到的依赖、补上缺的

hello-agent/
├── go.mod
├── main.go            # 入口：读配置、启动 HTTP 服务
├── agent/             # package agent：Agent Loop
│   ├── loop.go
│   └── loop_test.go   # 测试文件以 _test.go 结尾
└── tools/             # package tools：各种工具实现
    └── search.go

package tools

func Search(q string) string { return "..." } // 大写开头：其他包能用 tools.Search
func buildURL(q string) string { return "..." } // 小写开头：只有 tools 包内部能用

var name string = "search"   // 完整写法：var 变量名 类型 = 值
var count int                // 不赋值也行，自动是“零值”：int 是 0，string 是 ""，bool 是 false
model := "glm-4"             // 短声明：类型自动推断，只能在函数内部用（最常用）
const maxTurns = 10          // 常量，编译期确定，不能改

timeout := 20 * time.Second  // time.Duration 类型，常用于超时

func divide(a, b float64) (float64, error) { // 返回两个值：结果和错误
    if b == 0 {
        return 0, errors.New("除数不能为 0")
    }
    return a / b, nil // nil 表示“没有错误”
}

result, err := divide(10, 2) // 同时接收两个返回值
if err != nil {              // Go 里最常见的三行
    fmt.Println("出错了：", err)
    return
}
fmt.Println(result) // 5

_, err = divide(1, 0) // 不需要的返回值用 _ 丢掉

var ErrToolNotFound = errors.New("tool not found") // “哨兵错误”：包级变量，用来做比较

func findTool(name string) error {
    if name != "search" {
        return ErrToolNotFound
    }
    return nil
}

func runTool(name string) error {
    if err := findTool(name); err != nil {
        return fmt.Errorf("run %q: %w", name, err) // %w：把原错误“包”进新错误
    }
    return nil
}

err := runTool("weather")
fmt.Println(err)                              // run "weather": tool not found
fmt.Println(errors.Is(err, ErrToolNotFound))  // true：能穿透包装，找到里面的原错误

type ToolError struct {
    Tool string
    Code int
}

func (e *ToolError) Error() string { // 只要有 Error() string 方法，就是 error
    return fmt.Sprintf("tool %s failed with code %d", e.Tool, e.Code)
}

var te *ToolError
if errors.As(err, &te) { // errors.As：沿包装链找“这种类型”的错误，找到就赋值给 te
    fmt.Println(te.Code)
}

type Message struct {   // 定义一个新类型 Message
    Role    string      // "user" / "assistant" / "tool"
    Content string
}

type Conversation struct {
    ID       string
    Messages []Message  // 字段可以是 slice
}

// 方法：在 func 和名字之间写“接收者”(c *Conversation)
func (c *Conversation) Add(role, content string) {
    c.Messages = append(c.Messages, Message{Role: role, Content: content})
}

func (c Conversation) Len() int { // 值接收者：只读，不修改
    return len(c.Messages)
}

conv := &Conversation{ID: "c1"} // & 取地址，得到 *Conversation（指针）
conv.Add("user", "你好")
fmt.Println(conv.Len()) // 1

x := 1
p := &x   // p 是指向 x 的指针，类型 *int
*p = 2    // 通过指针改 x
fmt.Println(x) // 2

// slice：可变长度的数组
tools := []string{"search", "calc"}
tools = append(tools, "weather")       // 追加，必须接收返回值
fmt.Println(len(tools), tools[0])      // 3 search

for i, name := range tools {           // 遍历：下标 + 值
    fmt.Println(i, name)
}

results := make([]string, 3)           // 提前开好长度为 3 的 slice，元素是零值 ""

// map：键值对
registry := map[string]int{"search": 1}
registry["calc"] = 2                   // 写
v, ok := registry["weather"]           // 读：ok 表示键是否存在
if !ok {
    fmt.Println("没有这个工具", v)     // v 是零值 0
}
delete(registry, "calc")               // 删

type Tool interface {          // 接口：只规定“有哪些方法”
    Name() string
    Run(ctx context.Context, args string) (string, error)
}

type Calculator struct{}       // 一个具体工具

func (Calculator) Name() string { return "calc" }
func (Calculator) Run(ctx context.Context, args string) (string, error) {
    return "42", nil
}

// Calculator 没有写任何 "implements Tool"，但它有 Tool 要求的全部方法，所以它就是 Tool
var t Tool = Calculator{}

tools := map[string]Tool{}     // 工具注册表：名字 → 工具
tools[t.Name()] = t
out, err := tools["calc"].Run(context.Background(), `{"expr":"6*7"}`)

// 给 B6 的 Message 加上 JSON 标签后，就能直接用来调模型 API
type ChatRequest struct {
    Model       string    `json:"model"`                 // 标签：JSON 里叫 model
    Messages    []Message `json:"messages"`
    Temperature float64   `json:"temperature,omitempty"` // omitempty：零值时不输出
    Stream      bool      `json:"stream"`
}

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

// struct → JSON（编码）
req := ChatRequest{Model: "glm-4", Messages: []Message{{Role: "user", Content: "你好"}}, Stream: true}
b, err := json.Marshal(req) // b 是 []byte
// {"model":"glm-4","messages":[{"role":"user","content":"你好"}],"stream":true}

// JSON → struct（解码）
var resp struct {           // 匿名 struct：只用一次，不用起名字
    Choices []struct {
        Message Message `json:"message"`
    } `json:"choices"`
}
err = json.Unmarshal([]byte(`{"id":"x","choices":[{"message":{"role":"assistant","content":"hi"}}]}`), &resp)
fmt.Println(resp.Choices[0].Message.Content) // hi

func readConfig(path string) ([]byte, error) {
    f, err := os.Open(path)
    if err != nil {
        return nil, err
    }
    defer f.Close() // 登记：函数返回时关闭文件，不管从哪个 return 出去

    return io.ReadAll(f)
}

func f() {
    defer fmt.Println("3")  // 先登记，最后才执行
    defer fmt.Println("2")
    fmt.Println("1")
}
// 输出：1 2 3  —— 多个 defer 按“后登记先执行”

func Map[T, R any](xs []T, f func(T) R) []R { // [T, R any]：类型参数
    out := make([]R, 0, len(xs))
    for _, x := range xs {
        out = append(out, f(x))
    }
    return out
}

list := []Tool{Calculator{}}
names := Map(list, func(t Tool) string { return t.Name() }) // 自动推断 T=Tool, R=string，结果 ["calc"]

// divide_test.go，和 divide.go 放在同一目录、同一个包
package calc

import "testing"

func TestDivide(t *testing.T) { // 测试函数：Test 开头，参数是 *testing.T
    cases := []struct {         // 表驱动：把输入和期望写成一张表
        name    string
        a, b    float64
        want    float64
        wantErr bool
    }{
        {"正常", 10, 2, 5, false},
        {"除零", 1, 0, 0, true},
    }

    for _, tc := range cases {
        t.Run(tc.name, func(t *testing.T) { // 每行一个子测试，失败时能看到是哪一行
            got, err := divide(tc.a, tc.b)
            if (err != nil) != tc.wantErr {
                t.Fatalf("err = %v, wantErr %v", err, tc.wantErr)
            }
            if got != tc.want {
                t.Errorf("got %v, want %v", got, tc.want)
            }
        })
    }
}

go test ./...            # 跑所有测试
go test -run TestDivide  # 只跑名字匹配的
go test -v ./...         # 打印每个子测试
go test -race ./...      # 检测数据竞争（并发部分必用）

go doSomething()        // 在后台执行 doSomething，当前代码不等它，继续往下走

go func() {             // 也可以直接写一个匿名函数丢到后台
    fmt.Println("我在后台跑")
}()                     // 注意最后的 ()：定义完立刻调用

ch := make(chan string)     // 造一根能传 string 的管道（无缓冲）
ch2 := make(chan string, 3) // 带缓冲：管道里最多能“存” 3 个值

ch <- "hello"   // 往管道里放（发送）
v := <-ch       // 从管道里取（接收）
close(ch)       // 关闭管道：告诉接收方“不会再有数据了”

func produce() <-chan int   // 返回值类型 <-chan int：调用方只能从里面读
func consume(ch chan<- int) // 参数类型 chan<- int：函数只能往里面写

results := make([]string, 3)
for i := 0; i < 3; i++ {
    go func() {
        results[i] = "done" // 直接用外面的 results 和 i
    }()
}

results := make(chan string)      // 无缓冲 channel
go func() {
    results <- "tool A done"      // 后台：把结果放进管道
}()
fmt.Println(<-results)            // 前台：在这里等，直到拿到结果

主 goroutine:  创建 channel → 启动后台 → 执行 <-results（卡住等）……………… 拿到值，打印
后台 goroutine:                          执行 results <- "..."（等有人接）→ 交接完成，结束

func produce(n int) <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch) // 规则 1：发送方负责关闭
        for i := 0; i < n; i++ {
            ch <- i     // 每放一个，都要等调用方取走
        }
    }()                 // 后台开始生产
    return ch           // 立刻返回管道，不等生产完
}

for v := range produce(3) { fmt.Println(v) } // 0 1 2

select {
case tok := <-tokens:                       // 情况 A：等到一个 token
    fmt.Print(tok)
case <-time.After(30 * time.Second):        // 情况 B：30 秒都没等到
    return errors.New("LLM 30 秒没吐 token")
case <-ctx.Done():                          // 情况 C：被取消了
    return ctx.Err()
}

用户关掉网页
  → HTTP 服务端发现连接断了，r.Context() 被取消
    → 你传给 LLM 调用的 ctx 被取消 → 到模型供应商的连接断开，停止计费
    → 你传给每个工具的 ctx 被取消 → 工具停止执行

func handleChat(w http.ResponseWriter, r *http.Request) {
    // 从请求的 ctx 派生一个子 ctx，并加上“最多 60 秒”的限制
    ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
    defer cancel() // 必写：函数结束时主动释放计时器

    if err := runAgent(ctx); err != nil {   // 把 ctx 一路往下传
        if errors.Is(err, context.Canceled) {
            slog.Info("client gone, stopped") // 用户自己走了，不算错误
            return
        }
        http.Error(w, err.Error(), 500)
    }
}

串行：[ 工具A 1s ][ 工具B 2s ][ 工具C 3s ]          = 6s
并行：[ 工具A 1s ]
      [ 工具B 2s     ]
      [ 工具C 3s          ]                        = 3s

import "golang.org/x/sync/errgroup"

// 模型返回的一次工具调用请求
type ToolCall struct {
    ID   string          // 模型给的调用 ID，回传结果时要带上，模型靠它对号入座
    Name string          // 工具名，比如 "search"
    Args json.RawMessage // 参数原文（一段 JSON），先不解析，交给具体工具
}

// 工具执行完的结果
type ToolResult struct {
    ID     string
    Output string
    Err    error
}

func runTools(ctx context.Context, calls []ToolCall) ([]ToolResult, error) {
    g, ctx := errgroup.WithContext(ctx)           // ① 建一个任务组
    g.SetLimit(4)                                 // ② 最多同时跑 4 个
    results := make([]ToolResult, len(calls))     // ③ 提前开好结果数组

    for i, c := range calls {
        g.Go(func() error {                       // ④ 每个工具一个 goroutine
            tctx, cancel := context.WithTimeout(ctx, 20*time.Second) // ⑤ 单工具超时
            defer cancel()
            out, err := execTool(tctx, c)         // ⑥ 真正执行工具
            results[i] = ToolResult{ID: c.ID, Output: out, Err: err} // ⑦ 写自己的格子
            return nil                            // ⑧ 失败也返回 nil
        })
    }
    if err := g.Wait(); err != nil {              // ⑨ 等所有工具结束
        return nil, err
    }
    return results, nil
}

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: [DONE]

              ┌──────────── 后台 goroutine ────────────┐
HTTP body ──▶ │ 一行行读 → 过滤 data: → 解析 JSON → 取字 │ ──▶ tokens  （正常数据，一个个字）
              │            出错 / 被取消 / 读完        │ ──▶ errc    （为什么结束，最多一个值）
              └────────────────────────────────────────┘

func streamTokens(ctx context.Context, body io.ReadCloser) (<-chan string, <-chan error) {
    tokens := make(chan string)     // 无缓冲：调用方读一个，这边才发下一个
    errc := make(chan error, 1)     // 缓冲 1：放错误时不用等人来取

    go func() {
        defer close(errc)   // 第三个执行
        defer close(tokens) // 第二个执行
        defer body.Close()  // 第一个执行（defer 后登记先执行）

        sc := bufio.NewScanner(body)                     // 按行读
        sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)  // 单行上限放宽到 1MB
        for sc.Scan() {                                  // 每次读一行，读不到就结束循环
            line := sc.Text()
            if !strings.HasPrefix(line, "data: ") {
                continue // 跳过空行、注释、event: 行
            }
            data := strings.TrimPrefix(line, "data: ")
            if data == "[DONE]" {
                return // 正常结束：errc 里什么都不放
            }
            var chunk struct {
                Choices []struct {
                    Delta struct{ Content string } `json:"delta"`
                } `json:"choices"`
            }
            if err := json.Unmarshal([]byte(data), &chunk); err != nil {
                errc <- fmt.Errorf("bad chunk: %w", err) // JSON 坏了：报错并退出
                return
            }
            if len(chunk.Choices) == 0 {
                continue // 有的 chunk 没有 choices（比如只带用量统计），跳过
            }
            select {
            case tokens <- chunk.Choices[0].Delta.Content: // 把字交出去
            case <-ctx.Done():                              // 或者：调用方不要了
                errc <- ctx.Err()
                return
            }
        }
        if err := sc.Err(); err != nil { // 循环结束不是因为读完，而是读出错（比如断网）
            errc <- err
        }
    }()
    return tokens, errc
}

tokens, errc := streamTokens(ctx, resp.Body)
for tok := range tokens {          // 一直读到 tokens 被关闭
    fmt.Print(tok)
}
if err := <-errc; err != nil {     // 正常结束时 errc 里没值且已关闭，读到 nil
    return err
}

select {
case tokens <- 字:     // 尝试把字交给调用方
case <-ctx.Done():     // 同时盯着取消
}

后台:   errc <- err   ← 卡住：无缓冲，要等有人来取
调用方: for range tokens  ← 卡住：在等 tokens 被关闭
        （tokens 要等后台 return 后的 defer 才关闭，而后台卡在上一行）
结果:   两边互相等 → 死锁

后台:   errc <- err   ← 立刻完成，错误存进缓冲
        return → defer 关 body、关 tokens、关 errc
调用方: for range tokens 发现 tokens 关了，退出循环
        <-errc 拿到刚才那个错误

// ❌ 泄漏：调用方只读一个就走了，后台卡在第二次发送，永远不退出
func leaky() <-chan int {
    ch := make(chan int)
    go func() {
        for i := 0; ; i++ {
            ch <- i
        }
    }()
    return ch
}

// ✅ 不泄漏：发送时同时监听 ctx，调用方 cancel 后后台立刻退出
func safe(ctx context.Context) <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch)
        for i := 0; ; i++ {
            select {
            case ch <- i:
            case <-ctx.Done():
                return
            }
        }
    }()
    return ch
}

out, err := execTool(tctx, c)
if ctx.Err() != nil {          // 组 ctx（上一层）被取消：整轮中止
    return ctx.Err()
}
// 走到这里说明只是这个工具自己的 tctx 超时或工具本身出错：记下来，继续
results[i] = ToolResult{ID: c.ID, Output: out, Err: err}
return nil

ch1 ──▶ worker1 ──┐
ch2 ──▶ worker2 ──┼──▶ out ──▶ 调用方
ch3 ──▶ worker3 ──┘
          全部 worker 结束后，由一个“收尾 goroutine”关闭 out

func fanIn(ctx context.Context, chs ...<-chan string) <-chan string {
    out := make(chan string)
    var wg sync.WaitGroup
    wg.Add(len(chs))            // 计数器：有几路输入就记几

    for _, ch := range chs {
        go func() {             // 每路输入一个 worker
            defer wg.Done()     // worker 退出时计数 -1
            for {
                select {
                case v, ok := <-ch:          // 外层：等输入
                    if !ok {
                        return               // 这一路输入已关闭，worker 结束
                    }
                    select {
                    case out <- v:           // 内层：把值转发出去
                    case <-ctx.Done():       // 转发时也能被取消
                        return
                    }
                case <-ctx.Done():           // 外层：等输入时也能被取消
                    return
                }
            }
        }()
    }

    go func() {                 // 收尾 goroutine
        wg.Wait()               // 等计数归零 = 所有 worker 都退出了
        close(out)              // 这时才安全地关闭 out
    }()
    return out
}

var toolTimeout = 20 * time.Second // runTools 里用它替换写死的 20*time.Second

var fakeDelay = map[string]time.Duration{} // 每个假工具要“干”多久

func execTool(ctx context.Context, c ToolCall) (string, error) {
    select {
    case <-time.After(fakeDelay[c.Name]):   // 假装干活这么久
        return c.Name + " ok", nil
    case <-ctx.Done():                       // 被超时或取消打断
        return "", ctx.Err()
    }
}

func TestRunTools(t *testing.T) {
    toolTimeout = 2 * time.Second
    fakeDelay = map[string]time.Duration{
        "a": 100 * time.Millisecond,
        "b": 200 * time.Millisecond,
        "c": 2500 * time.Millisecond, // 比超时长，会被打断
    }
    calls := []ToolCall{{ID: "1", Name: "a"}, {ID: "2", Name: "b"}, {ID: "3", Name: "c"}}

    start := time.Now()
    res, err := runTools(context.Background(), calls)
    elapsed := time.Since(start)

    if err != nil {
        t.Fatalf("unexpected err: %v", err)
    }
    if elapsed < 1900*time.Millisecond || elapsed > 2300*time.Millisecond {
        t.Fatalf("elapsed = %v, want ~2s (并行，受最慢的超时限制)", elapsed)
    }
    if res[0].Err != nil || res[1].Err != nil {
        t.Fatalf("a/b should succeed: %+v", res)
    }
    if !errors.Is(res[2].Err, context.DeadlineExceeded) {
        t.Fatalf("c should time out, got %v", res[2].Err)
    }
}

func TestStreamCancel(t *testing.T) {
    disconnected := make(chan struct{}) // 服务端发现断开时关闭它，当“信号灯”用

    // 1. 起一个假 SSE 服务端
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Content-Type", "text/event-stream")
        flusher := w.(http.Flusher)
        for i := 0; ; i++ {
            select {
            case <-r.Context().Done(): // 客户端断开时触发
                close(disconnected)
                return
            case <-time.After(100 * time.Millisecond):
                fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":\"t%d\"}}]}\n\n", i)
                flusher.Flush() // 立刻推给客户端，不然会攒在缓冲区里
            }
        }
    }))
    defer srv.Close()

    // 2. 客户端发请求，请求绑定一个可取消的 ctx
    ctx, cancel := context.WithCancel(context.Background())
    req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        t.Fatal(err)
    }

    // 3. 读 5 个 token 后取消
    tokens, errc := streamTokens(ctx, resp.Body)
    for i := 0; i < 5; i++ {
        <-tokens
    }
    cancel()

    // 4. 排空 tokens，确认后台 goroutine 退出，并拿到一个错误
    for range tokens {
    }
    if err := <-errc; err == nil {
        t.Fatal("want a cancellation error, got nil")
    }

    // 5. 断言服务端在 2 秒内感知到断开
    select {
    case <-disconnected:
    case <-time.After(2 * time.Second):
        t.Fatal("server never saw the disconnect: 上游连接没断，还在烧钱")
    }
}

go test -race -count=1 ./...