M04 Agent 核心架构

Agent = LLM(决策器) + Tools(感知与行动接口) + State(状态) + Loop(自主循环) + Stop(停止条件)

        [想 Think]
            │
            │ 模型决定：继续调用工具，还是结束？
            ▼
        [做 Act]
            │ 执行模型决定的动作：调工具、调子 Agent、写状态
            ▼
      [观察 Observe]
            │ 把动作结果回填到上下文
            ▼
        [再想 Think] → ... 直到完成或触发停止条件

Round 1
  想: 我需要知道用户位置，可以调用 get_location 工具。
  做: 调 get_location()
  观察: "北京海淀区中关村大街 1 号"

Round 2
  想: 现在需要查附近 KFC，可以调用 search_nearby。
  做: 调 search_nearby("星巴克", "北京海淀区中关村大街 1 号")
  观察: "找到 3 家：A 店、B 店、C 店，带 ID"

Round 3
  想: 我需要营业时间。
  做: 调 get_store_hours("A 店 ID")
  观察: "今日 7:00-22:00"

Round 4
  想: 信息够了，可以回答。
  做: 结束循环，输出最终答案。

普通问答:   [Prompt] ──► [Model] ──► [Answer]                  (单向、一次)

工作流:     [Step1] ──► [Step2] ──► [Step3]                   (路径写死在代码里)
              │           │           │
            Model       Model        Model                    (模型只是被调用的零件)

Agent:           ┌──────────────────────────┐
                 ▼                          │
           [Model 决定下一步] ──► [执行] ──► [观察]            (路径由模型运行时决定)
                 │
                 └──► [完成?] ──► [最终答案]

package agent

import "github.com/yourname/llmagent/internal/cost"
import "github.com/yourname/llmagent/internal/llm"

type Phase string

const (
	PhaseThinking Phase = "thinking"
	PhaseActing   Phase = "acting"
	PhaseDone     Phase = "done"
	PhaseError    Phase = "error"
)

// State 是一次 Agent 运行的完整快照，刻意设计成可 JSON 序列化（见 4.10）。
type State struct {
	Goal     string        `json:"goal"`     // 用户给的目标
	Messages []llm.Message `json:"messages"` // 完整对话历史，含工具结果——这是 Agent 的“记忆”
	Step     int           `json:"step"`     // 已执行的步数
	Phase    Phase         `json:"phase"`
	Answer   string        `json:"answer,omitempty"` // 终态时的最终答案
	Usage    cost.Usage    `json:"usage"`            // 累计 token 用量
}

package agent

import (
	"context"

	"github.com/yourname/llmagent/internal/llm"
	"github.com/yourname/llmagent/internal/tool"
)

type Agent struct {
	provider     llm.Provider   // 模型出口；传入 M02 的单个 Provider 或 router 适配器
	model        string         // 模型名
	tools        *tool.Registry // 可用工具集合
	systemPrompt string         // 系统提示词
	budget       Budget         // 停止条件，见 4.6
	store        Store          // 状态持久化，可选，见 4.10
}

func New(p llm.Provider, model string, tools *tool.Registry, opts ...Option) *Agent {
	a := &Agent{
		provider: p,
		model:    model,
		tools:    tools,
		budget:   Budget{MaxSteps: 10}, // 给个安全默认值
	}
	for _, opt := range opts {
		opt(a)
	}
	return a
}

// Option 用函数式选项配置可选项（沿用 M02 的模式）。
type Option func(*Agent)

func WithSystemPrompt(p string) Option { return func(a *Agent) { a.systemPrompt = p } }
func WithBudget(b Budget) Option       { return func(a *Agent) { a.budget = b } }
func WithStore(s Store) Option         { return func(a *Agent) { a.store = s } }

package tool

import (
	"context"
	"encoding/json"

	"github.com/yourname/llmagent/internal/schema"
)

type Tool interface {
	Name() string
	Description() string
	Parameters() *schema.Schema                                     // 参数的 JSON Schema
	Call(ctx context.Context, args json.RawMessage) (string, error) // 执行，返回给模型的观察文本
}

// Registry 是工具的注册表，按名字查找。
type Registry struct {
	tools map[string]Tool
}

func NewRegistry(ts ...Tool) *Registry {
	r := &Registry{tools: make(map[string]Tool, len(ts))}
	for _, t := range ts {
		r.tools[t.Name()] = t
	}
	return r
}

func (r *Registry) Get(name string) (Tool, bool) { t, ok := r.tools[name]; return t, ok }

func (r *Registry) All() []Tool {
	out := make([]Tool, 0, len(r.tools))
	for _, t := range r.tools {
		out = append(out, t)
	}
	return out
}

Thought: 我需要查北京的天气
Action: get_weather
Action Input: {"city": "北京"}

const reactSystemTmpl = `你是一个会使用工具来完成任务的助手。

你可以使用以下工具：
{{range .Tools}}- {{.Name}}: {{.Description}}
{{end}}
请严格按照下面的格式，一步一步地推理和行动：

Thought: 你对当前情况的思考
Action: 要使用的工具名称（必须是上面列出的之一）
Action Input: 调用该工具的参数，必须是合法的 JSON

我会把工具的执行结果作为 Observation 返回给你。然后你继续下一轮 Thought/Action。

当你已经获得足够信息、可以回答用户时，请输出：

Thought: 我已经知道答案了
Final Answer: 给用户的最终回答`

type reactStep struct {
	Action      string // 要调用的工具
	ActionInput string // 工具参数（JSON 文本）
	FinalAnswer string // 若非空，表示循环结束
}

func parseReact(text string) (reactStep, error) {
	var s reactStep
	// 优先看有没有 Final Answer：有就直接结束。
	if idx := strings.Index(text, "Final Answer:"); idx >= 0 {
		s.FinalAnswer = strings.TrimSpace(text[idx+len("Final Answer:"):])
		return s, nil
	}
	// 否则尝试抠出 Action 与 Action Input。
	actionIdx := strings.Index(text, "Action:")
	inputIdx := strings.Index(text, "Action Input:")
	if actionIdx < 0 || inputIdx < 0 || inputIdx < actionIdx {
		return s, fmt.Errorf("ReAct 解析失败：未找到 Action / Action Input")
	}
	s.Action = strings.TrimSpace(text[actionIdx+len("Action:") : inputIdx])
	s.ActionInput = strings.TrimSpace(text[inputIdx+len("Action Input:"):])
	return s, nil
}

// llm 包：给 ChatRequest 增加 Stop 字段（M04 新增）
type ChatRequest struct {
	// ...M02 已有字段...
	Stop []string `json:"stop,omitempty"` // 命中任一序列时模型停止生成
}

func (a *Agent) runReAct(ctx context.Context, goal string) (*State, error) {
	st := &State{
		Goal:     goal,
		Phase:    PhaseThinking,
		Messages: []llm.Message{
			{Role: llm.RoleSystem, Content: a.systemPrompt},
			{Role: llm.RoleUser, Content: goal},
		},
	}

	for {
		// —— 停止条件检查（详见 4.6）——
		if stop, reason := a.budget.Exceeded(st); stop {
			st.Phase = PhaseError
			return st, fmt.Errorf("提前终止：%s", reason)
		}

		// —— Thinking 阶段：调用模型 ——
		st.Phase = PhaseThinking
		resp, err := a.provider.Chat(ctx, llm.ChatRequest{
			Model:    a.model,
			Messages: st.Messages,
			Stop:     []string{"Observation:"}, // 关键：不让模型自己编 Observation
		})
		if err != nil {
			st.Phase = PhaseError
			return st, err
		}
		st.Step++
		st.Usage.InputTokens += resp.InputTokens
		st.Usage.OutputTokens += resp.OutputTokens

		// —— 解析模型意图 ——
		step, err := parseReact(resp.Content)
		if err != nil {
			// 解析失败不致命：把错误反馈给模型让它重写（错误自愈，见 4.7）
			st.Messages = append(st.Messages,
				llm.Message{Role: llm.RoleAssistant, Content: resp.Content},
				llm.Message{Role: llm.RoleUser, Content: "你的输出格式不正确，请严格按 Thought/Action/Action Input 或 Final Answer 重新输出。"},
			)
			continue
		}

		// —— 模型给出最终答案：进入 Done ——
		if step.FinalAnswer != "" {
			st.Phase = PhaseDone
			st.Answer = step.FinalAnswer
			return st, nil
		}

		// —— Acting 阶段：执行工具 ——
		st.Phase = PhaseActing
		observation := a.callTool(ctx, step.Action, json.RawMessage(step.ActionInput))

		// 把这一轮（模型的思考 + 工具的观察）追加进历史，回到下一轮 Thinking
		st.Messages = append(st.Messages,
			llm.Message{Role: llm.RoleAssistant, Content: resp.Content},
			llm.Message{Role: llm.RoleUser, Content: "Observation: " + observation},
		)
	}
}

// callTool 执行工具，并把"工具不存在""执行报错"也转成可喂回模型的观察文本。
func (a *Agent) callTool(ctx context.Context, name string, args json.RawMessage) string {
	t, ok := a.tools.Get(name)
	if !ok {
		return fmt.Sprintf("错误：工具 %q 不存在", name)
	}
	out, err := t.Call(ctx, args)
	if err != nil {
		return "错误：工具执行失败：" + err.Error()
	}
	return out
}

// llm 包：工具调用相关类型（M04 新增）

// ToolDef 是给模型看的工具定义。
type ToolDef struct {
	Name        string         `json:"name"`
	Description string         `json:"description"`
	Parameters  *schema.Schema `json:"parameters"`
}

// ToolCall 是模型返回的"我要调用这个工具"的结构化意图。
type ToolCall struct {
	ID   string          `json:"id"`        // 厂商给的调用 ID，回传结果时要带上
	Name string          `json:"name"`      // 工具名
	Args json.RawMessage `json:"arguments"` // 参数（JSON）
}

// llm 包：Message 扩展（M04 新增字段）
type Message struct {
	Role       Role       `json:"role"`
	Content    string     `json:"content"`
	ToolCalls  []ToolCall `json:"tool_calls,omitempty"`   // assistant 消息：模型发起的调用
	ToolCallID string     `json:"tool_call_id,omitempty"` // tool 消息：这条结果对应哪个调用
}

func (a *Agent) runFunctionCalling(ctx context.Context, goal string) (*State, error) {
	st := &State{
		Goal:  goal,
		Phase: PhaseThinking,
		Messages: []llm.Message{
			{Role: llm.RoleSystem, Content: a.systemPrompt},
			{Role: llm.RoleUser, Content: goal},
		},
	}
	toolDefs := a.toolDefs() // 由注册的工具生成 []llm.ToolDef

	for {
		if stop, reason := a.budget.Exceeded(st); stop {
			st.Phase = PhaseError
			return st, fmt.Errorf("提前终止：%s", reason)
		}

		st.Phase = PhaseThinking
		resp, err := a.provider.Chat(ctx, llm.ChatRequest{
			Model:    a.model,
			Messages: st.Messages,
			Tools:    toolDefs,
		})
		if err != nil {
			st.Phase = PhaseError
			return st, err
		}
		st.Step++
		st.Usage.InputTokens += resp.InputTokens
		st.Usage.OutputTokens += resp.OutputTokens

		// 模型没有要调用任何工具 → 这就是最终答案
		if len(resp.ToolCalls) == 0 {
			st.Phase = PhaseDone
			st.Answer = resp.Content
			return st, nil
		}

		// 先把"模型发起的工具调用"作为一条 assistant 消息记进历史
		st.Messages = append(st.Messages, llm.Message{
			Role:      llm.RoleAssistant,
			Content:   resp.Content,
			ToolCalls: resp.ToolCalls,
		})

		// 逐个执行工具，每个结果作为一条 tool 消息回填（注意要带 ToolCallID）
		st.Phase = PhaseActing
		for _, tc := range resp.ToolCalls {
			observation := a.callTool(ctx, tc.Name, tc.Args)
			st.Messages = append(st.Messages, llm.Message{
				Role:       llm.RoleTool,
				ToolCallID: tc.ID,
				Content:    observation,
			})
		}
	}
}

// toolDefs 把注册表里的工具转成给模型的定义清单。
func (a *Agent) toolDefs() []llm.ToolDef {
	tools := a.tools.All()
	defs := make([]llm.ToolDef, 0, len(tools))
	for _, t := range tools {
		defs = append(defs, llm.ToolDef{
			Name:        t.Name(),
			Description: t.Description(),
			Parameters:  t.Parameters(),
		})
	}
	return defs
}

package agent

import "time"

type Budget struct {
	MaxSteps  int       // 最大步数
	MaxTokens int       // 累计 token 上限，0 表示不限
	Deadline  time.Time // 截止时刻，零值表示不限
}

// Exceeded 判断当前状态是否已触发任一停止条件。
func (b Budget) Exceeded(st *State) (bool, string) {
	if b.MaxSteps > 0 && st.Step >= b.MaxSteps {
		return true, fmt.Sprintf("达到最大步数 %d", b.MaxSteps)
	}
	if b.MaxTokens > 0 && st.Usage.InputTokens+st.Usage.OutputTokens >= b.MaxTokens {
		return true, fmt.Sprintf("达到 Token 预算 %d", b.MaxTokens)
	}
	if !b.Deadline.IsZero() && time.Now().After(b.Deadline) {
		return true, "超过截止时间"
	}
	return false, ""
}

// 在 Agent 里维护一个动作计数器（也可以并入 State 以便持久化）
func actionSignature(name string, args json.RawMessage) string {
	return name + ":" + string(args)
}

// 在循环外先初始化一个动作计数器（要持久化就把它并入 State，并在 New 时 make）：
actionCounts := map[string]int{}

// 在循环里，执行工具前：
sig := actionSignature(step.Action, json.RawMessage(step.ActionInput))
actionCounts[sig]++
if actionCounts[sig] >= 3 {
	st.Phase = PhaseError
	return st, fmt.Errorf("检测到原地打转：连续 3 次执行相同动作 %s", sig)
}

// 在循环顶部维护
healAttempts := 0
const maxHealAttempts = 3

// 当遇到“可自愈”错误（解析失败、工具报错等）时：
healAttempts++
if healAttempts > maxHealAttempts {
	st.Phase = PhaseError
	return st, fmt.Errorf("连续 %d 次自愈失败，放弃", maxHealAttempts)
}
// ……把错误作为观察喂回，continue……

// 而遇到“不可恢复”错误时，直接终止：
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
	st.Phase = PhaseError
	return st, err
}

package agent

type EventType string

const (
	EventThought     EventType = "thought"      // 模型的一段思考
	EventToolCall    EventType = "tool_call"    // 即将调用某工具
	EventToolResult  EventType = "tool_result"  // 工具返回了结果
	EventAnswerDelta EventType = "answer_delta" // 最终答案的一个增量（流式）
	EventError       EventType = "error"
	EventDone        EventType = "done"
)

// AgentEvent 是 Agent 运行过程中向外发出的一个事件。
type AgentEvent struct {
	Type EventType `json:"type"`
	Text string    `json:"text,omitempty"` // 思考内容 / 答案增量 / 错误信息
	Tool string    `json:"tool,omitempty"` // 涉及的工具名
	Args string    `json:"args,omitempty"` // 工具参数
}

func (a *Agent) RunStream(ctx context.Context, goal string) <-chan AgentEvent {
	out := make(chan AgentEvent)

	go func() {
		defer close(out) // M01 纪律：生产者负责关闭

		emit := func(ev AgentEvent) bool {
			select {
			case <-ctx.Done():
				return false // 消费者不读了/被取消，停止
			case out <- ev:
				return true
			}
		}

		st := &State{
			Goal:  goal,
			Phase: PhaseThinking,
			Messages: []llm.Message{
				{Role: llm.RoleSystem, Content: a.systemPrompt},
				{Role: llm.RoleUser, Content: goal},
			},
		}
		for {
			if stop, reason := a.budget.Exceeded(st); stop {
				emit(AgentEvent{Type: EventError, Text: "提前终止：" + reason})
				return
			}

			resp, err := a.provider.Chat(ctx, llm.ChatRequest{
				Model: a.model, Messages: st.Messages, Tools: a.toolDefs(),
			})
			if err != nil {
				emit(AgentEvent{Type: EventError, Text: err.Error()})
				return
			}
			st.Step++
			st.Usage.InputTokens += resp.InputTokens
			st.Usage.OutputTokens += resp.OutputTokens

			if len(resp.ToolCalls) == 0 {
				// 注：这里用非流式 Chat，最终答案是整段一次性发出的；
				// 要做到逐 token 流式，把这处的出口换成 M02 的 ChatStream、把每个增量 emit 出去即可。
				emit(AgentEvent{Type: EventAnswerDelta, Text: resp.Content})
				emit(AgentEvent{Type: EventDone})
				return
			}

			if resp.Content != "" {
				emit(AgentEvent{Type: EventThought, Text: resp.Content})
			}
			st.Messages = append(st.Messages, llm.Message{
				Role: llm.RoleAssistant, Content: resp.Content, ToolCalls: resp.ToolCalls,
			})
			for _, tc := range resp.ToolCalls {
				emit(AgentEvent{Type: EventToolCall, Tool: tc.Name, Args: string(tc.Args)})
				observation := a.callTool(ctx, tc.Name, tc.Args)
				emit(AgentEvent{Type: EventToolResult, Tool: tc.Name, Text: observation})
				st.Messages = append(st.Messages, llm.Message{
					Role: llm.RoleTool, ToolCallID: tc.ID, Content: observation,
				})
			}
		}
	}()

	return out
}

for ev := range agent.RunStream(ctx, goal) {
	switch ev.Type {
	case agent.EventThought:
		fmt.Printf("\n[思考] %s\n", ev.Text)
	case agent.EventToolCall:
		fmt.Printf("[调用工具] %s(%s)\n", ev.Tool, ev.Args)
	case agent.EventToolResult:
		fmt.Printf("[结果] %s\n", ev.Text)
	case agent.EventAnswerDelta:
		fmt.Print(ev.Text) // 流式答案逐段打印
	case agent.EventError:
		fmt.Fprintln(os.Stderr, "\n[错误]", ev.Text)
	}
}

package plan

import "encoding/json"

type Task struct {
	ID        string          `json:"id"`         // 任务唯一标识
	Tool      string          `json:"tool"`       // 调用哪个工具
	Args      json.RawMessage `json:"args"`       // 工具参数
	DependsOn []string        `json:"depends_on"` // 依赖的前置任务 ID
}

type Plan struct {
	Tasks []Task `json:"tasks"`
}

package plan

import (
	"fmt"
	"sort"
)

// Levels 把任务按拓扑层级分组：同一层内的任务无相互依赖，可并行执行。
// 若存在循环依赖或依赖了不存在的任务 ID，返回错误。
func Levels(p Plan) ([][]string, error) {
	// 第 1 步：建立任务 ID 集合，用于校验 DependsOn 指向的是合法任务。
	exists := make(map[string]bool, len(p.Tasks))
	for _, t := range p.Tasks {
		if exists[t.ID] {
			return nil, fmt.Errorf("任务 ID %q 重复", t.ID)
		}
		exists[t.ID] = true
	}

	indeg := make(map[string]int)           // 每个任务剩余的未满足依赖数
	dependents := make(map[string][]string) // dep -> 依赖它的任务列表
	for _, t := range p.Tasks {
		if _, ok := indeg[t.ID]; !ok {
			indeg[t.ID] = 0
		}
		for _, d := range t.DependsOn {
			// 第 2 步：校验依赖存在，避免后面被错认为"循环依赖"。
			if !exists[d] {
				return nil, fmt.Errorf("任务 %q 依赖了不存在的任务 %q", t.ID, d)
			}
			indeg[t.ID]++
			dependents[d] = append(dependents[d], t.ID)
		}
	}

	// 首层：入度为 0 的任务
	var cur []string
	for id, n := range indeg {
		if n == 0 {
			cur = append(cur, id)
		}
	}

	var levels [][]string
	done := 0
	for len(cur) > 0 {
		sort.Strings(cur) // 排序只为输出稳定、便于测试
		levels = append(levels, cur)
		done += len(cur)

		var next []string
		for _, id := range cur {
			for _, dep := range dependents[id] {
				indeg[dep]--
				if indeg[dep] == 0 {
					next = append(next, dep)
				}
			}
		}
		cur = next
	}

	// 走到这里 done 仍 < len(indeg) 的，只可能是循环依赖
	// （非法依赖已经在前面被拦下）
	if done != len(indeg) {
		return nil, fmt.Errorf("计划存在循环依赖，无法执行")
	}
	return levels, nil
}

package plan

import (
	"context"
	"fmt"
	"sync"

	"github.com/yourname/llmagent/internal/tool"
)

// Execute 按拓扑层级执行计划，层内并行。返回每个任务 ID 到其输出的映射。
func Execute(ctx context.Context, p Plan, reg *tool.Registry) (map[string]string, error) {
	levels, err := Levels(p)
	if err != nil {
		return nil, err
	}

	byID := make(map[string]Task, len(p.Tasks))
	for _, t := range p.Tasks {
		byID[t.ID] = t
	}

	results := make(map[string]string)
	var mu sync.Mutex // 保护 results 的并发写

	for _, level := range levels {
		var wg sync.WaitGroup
		var firstErr error
		var once sync.Once
		lctx, cancel := context.WithCancel(ctx) // 本层任一失败即取消其余

		for _, id := range level {
			t := byID[id]
			wg.Add(1)
			go func(t Task) {
				defer wg.Done()

				tl, ok := reg.Get(t.Tool)
				if !ok {
					once.Do(func() { firstErr = fmt.Errorf("工具 %q 不存在", t.Tool); cancel() })
					return
				}
				out, err := tl.Call(lctx, t.Args)
				if err != nil {
					once.Do(func() { firstErr = err; cancel() })
					return
				}
				mu.Lock()
				results[t.ID] = out
				mu.Unlock()
			}(t)
		}

		wg.Wait()
		cancel() // 释放本层 context
		if firstErr != nil {
			return results, firstErr
		}
	}
	return results, nil
}

package agent

import "context"

type Store interface {
	Save(ctx context.Context, sessionID string, st *State) error
	Load(ctx context.Context, sessionID string) (*State, error)
}

package agent

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
)

type FileStore struct{ dir string }

func NewFileStore(dir string) *FileStore { return &FileStore{dir: dir} }

func (s *FileStore) Save(_ context.Context, sessionID string, st *State) error {
	data, err := json.MarshalIndent(st, "", "  ")
	if err != nil {
		return err
	}
	// 注意：sessionID 来自外部，需防目录穿越，这里用 Base 兜底
	path := filepath.Join(s.dir, filepath.Base(sessionID)+".json")
	return os.WriteFile(path, data, 0o600)
}

func (s *FileStore) Load(_ context.Context, sessionID string) (*State, error) {
	path := filepath.Join(s.dir, filepath.Base(sessionID)+".json")
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, fmt.Errorf("加载会话 %s 失败: %w", sessionID, err)
	}
	var st State
	if err := json.Unmarshal(data, &st); err != nil {
		return nil, err
	}
	return &st, nil
}

// 在循环每轮末尾追加：
if a.store != nil {
	if err := a.store.Save(ctx, sessionID, st); err != nil {
		// 存盘失败通常不该中断主流程，记日志即可（可观测见 M10）
		log.Printf("checkpoint 失败: %v", err)
	}
}

// 恢复时：load 出 State，从 st.Step 处继续那个 for 循环即可，
// 因为 Messages 里已经有全部历史，模型能无缝接上。

// router.Router 的 Chat 返回三个值，而 Agent 依赖 llm.Provider 接口。
// 包一层适配器即可让 router 当作 Provider 使用。
type routerProvider struct{ r *router.Router }

func (p routerProvider) Name() string                 { return "router" }
func (p routerProvider) Capabilities() llm.Capability { return llm.Capability{Streaming: true, Tools: true} }
func (p routerProvider) Chat(ctx context.Context, req llm.ChatRequest) (*llm.ChatResponse, error) {
	resp, _, err := p.r.Chat(ctx, req)
	return resp, err
}
func (p routerProvider) ChatStream(ctx context.Context, req llm.ChatRequest) (<-chan llm.StreamChunk, error) {
	// 简化：路由的流式可先只走第一个 Provider，或留作进阶
	return nil, fmt.Errorf("router 暂不支持流式，请用非流式 Chat")
}