第3周：Workflow、Approval、Guardrails、MCP - 完整概览#

// internal/workflow/step.go
type AgentState struct {
	ID              string
	Stage           WorkflowStage
	UserInput       string
	Classification  string          // 用户问题分类
	RetrievedChunks []Chunk        // RAG检索结果
	ToolCalls       []ToolCall
	DraftAction     interface{}    // 待执行的动作（草稿）
	FinalAnswer     string
	Error           error
}

type WorkflowStage string

const (
	StageReceived          WorkflowStage = "received"
	StageClassified        WorkflowStage = "classified"
	StageRetrieving        WorkflowStage = "retrieving"
	StageToolCalling       WorkflowStage = "tool_calling"
	StageAwaitingApproval  WorkflowStage = "awaiting_approval"
	StageExecuting         WorkflowStage = "executing"
	StageCompleted         WorkflowStage = "completed"
	StageFailed            WorkflowStage = "failed"
)

// Step是一个工作流阶段
type Step interface {
	Name() string
	Run(ctx context.Context, state *AgentState) (*AgentState, error)
}

// 具体实现
type ClassificationStep struct {
	llmClient LLMClient
}

func (s *ClassificationStep) Name() string {
	return "classification"
}

func (s *ClassificationStep) Run(ctx context.Context, state *AgentState) (*AgentState, error) {
	// 判断这是查询还是写操作
	// 查询：直接进入RAG
	// 写操作：需要approval
	
	if isWriteOperation(state.UserInput) {
		state.Stage = StageRetrieving  // 跳过retrieval，直接到tool calling
	} else {
		state.Stage = StageRetrieving
	}
	
	return state, nil
}

// Workflow编排
type Workflow struct {
	steps []Step
	store StateStore  // 持久化
}

func (w *Workflow) Execute(ctx context.Context, input string) (*AgentState, error) {
	state := &AgentState{
		ID:        uuid.New().String(),
		Stage:     StageReceived,
		UserInput: input,
	}
	
	for _, step := range w.steps {
		slog.Info("step executing", "step", step.Name(), "state_id", state.ID)
		
		newState, err := step.Run(ctx, state)
		if err != nil {
			newState.Stage = StageFailed
			newState.Error = err
			slog.Error("step failed", "step", step.Name(), "error", err)
		}
		
		state = newState
		
		// 持久化
		w.store.Save(state)
		
		// 检查是否需要停止
		if state.Stage == StageAwaitingApproval || state.Stage == StageFailed {
			break
		}
	}
	
	return state, state.Error
}

// internal/state/store.go
type StateStore interface {
	Save(ctx context.Context, state *AgentState) error
	Get(ctx context.Context, stateID string) (*AgentState, error)
	List(ctx context.Context, filter StateFilter) ([]AgentState, error)
}

// PostgreSQL实现（长期存储）
type PostgresStateStore struct {
	db *sql.DB
}

func (s *PostgresStateStore) Save(ctx context.Context, state *AgentState) error {
	query := `
		INSERT INTO agent_states (id, stage, user_input, state_json, created_at, updated_at)
		VALUES ($1, $2, $3, $4, NOW(), NOW())
		ON CONFLICT (id) DO UPDATE SET 
			stage = EXCLUDED.stage,
			state_json = EXCLUDED.state_json,
			updated_at = NOW()
	`
	stateJSON, _ := json.Marshal(state)
	return s.db.ExecContext(ctx, query, state.ID, state.Stage, state.UserInput, stateJSON)
}

// Redis实现（短期session，快速访问）
type RedisSessionStore struct {
	client *redis.Client
}

func (r *RedisSessionStore) Save(ctx context.Context, state *AgentState) error {
	stateJSON, _ := json.Marshal(state)
	return r.client.Set(ctx, fmt.Sprintf("state:%s", state.ID), stateJSON, 1*time.Hour).Err()
}

// Checkpoint用于长流程的断点恢复
type Checkpoint struct {
	ID        string
	StateID   string
	Stage     WorkflowStage
	Timestamp time.Time
	Data      json.RawMessage  // 当前状态快照
}

func (w *Workflow) CreateCheckpoint(state *AgentState) error {
	checkpoint := Checkpoint{
		ID:        uuid.New().String(),
		StateID:   state.ID,
		Stage:     state.Stage,
		Timestamp: time.Now(),
	}
	data, _ := json.Marshal(state)
	checkpoint.Data = data
	
	return w.store.SaveCheckpoint(checkpoint)
}

// 恢复流程
func (w *Workflow) ResumeFromCheckpoint(ctx context.Context, checkpointID string) (*AgentState, error) {
	checkpoint, err := w.store.GetCheckpoint(checkpointID)
	if err != nil {
		return nil, err
	}
	
	var state AgentState
	json.Unmarshal(checkpoint.Data, &state)
	
	// 继续执行
	return w.Execute(ctx, state.UserInput)
}

// internal/approval/approval.go
type PendingAction struct {
	ID              string
	StateID         string
	ActionType      string  // "create_ticket", "update_user", etc.
	ActionData      json.RawMessage
	RequiredRole    string
	Status          ApprovalStatus
	CreatedAt       time.Time
	ApprovedBy      string
	ApprovedAt      time.Time
	RejectionReason string
}

type ApprovalStatus string

const (
	StatusPending   ApprovalStatus = "pending"
	StatusApproved  ApprovalStatus = "approved"
	StatusRejected  ApprovalStatus = "rejected"
	StatusExpired   ApprovalStatus = "expired"
)

type ApprovalStore interface {
	Create(ctx context.Context, action *PendingAction) error
	Get(ctx context.Context, actionID string) (*PendingAction, error)
	List(ctx context.Context, filter ApprovalFilter) ([]PendingAction, error)
	Approve(ctx context.Context, actionID string, approverID string) error
	Reject(ctx context.Context, actionID string, reason string) error
}

// Handler: 当Agent想创建工单时
type CreateTicketDraftStep struct {
	llmClient      LLMClient
	approvalStore  ApprovalStore
}

func (s *CreateTicketDraftStep) Run(ctx context.Context, state *AgentState) (*AgentState, error) {
	// 1. LLM 生成工单草稿
	ticketDraft := s.llmClient.GenerateTicketDraft(ctx, state.UserInput)
	
	// 2. 创建PendingAction（不是直接创建！）
	pendingAction := &PendingAction{
		ID:           uuid.New().String(),
		StateID:      state.ID,
		ActionType:   "create_ticket",
		ActionData:   marshalJSON(ticketDraft),
		RequiredRole: "support_l2",
		Status:       StatusPending,
		CreatedAt:    time.Now(),
	}
	
	s.approvalStore.Create(ctx, pendingAction)
	
	// 3. 更新state
	state.Stage = StageAwaitingApproval
	state.DraftAction = pendingAction
	
	return state, nil
}

// API: 获取待审批列表
func (h *Handler) GetPendingActions(w http.ResponseWriter, r *http.Request) {
	user := getCurrentUser(r)
	
	// 只返回这个用户有权审批的
	filter := ApprovalFilter{
		Status:        StatusPending,
		RequiredRoles: user.Roles,
	}
	
	actions, _ := h.approvalStore.List(r.Context(), filter)
	json.NewEncoder(w).Encode(actions)
}

// API: 审批
func (h *Handler) ApproveAction(w http.ResponseWriter, r *http.Request) {
	actionID := chi.URLParam(r, "actionID")
	user := getCurrentUser(r)
	
	action, _ := h.approvalStore.Get(r.Context(), actionID)
	
	// 权限检查
	if !user.HasRole(action.RequiredRole) {
		http.Error(w, "insufficient permission", http.StatusForbidden)
		return
	}
	
	// 标记为approved
	h.approvalStore.Approve(r.Context(), actionID, user.ID)
	
	// 触发执行（Day 18的Executor）
	h.executor.ExecuteApprovedAction(r.Context(), action)
	
	w.WriteHeader(http.StatusOK)
}

// internal/guardrails/guardrail.go
type GuardrailType string

const (
	TypeInput      GuardrailType = "input"      // 防prompt injection
	TypeOutput     GuardrailType = "output"     // 防敏感信息泄漏
	TypeTool       GuardrailType = "tool"       // 防高风险工具
	TypeRetrieval  GuardrailType = "retrieval"  // 防越权访问
)

type Guardrail interface {
	Check(ctx context.Context, input interface{}) (passed bool, reason string, err error)
}

// 1. Input Guardrail - 防Prompt Injection
type InputGuardrail struct {
	injectionPatterns []string
}

func (g *InputGuardrail) Check(ctx context.Context, input interface{}) (bool, string, error) {
	userInput := input.(string)
	
	injectionTests := []string{
		"ignore the previous instructions",
		"forget all rules",
		"execute this command",
		"don't log this",
		"bypass all restrictions",
	}
	
	for _, pattern := range injectionTests {
		if strings.Contains(strings.ToLower(userInput), pattern) {
			return false, "suspected prompt injection", nil
		}
	}
	
	return true, "", nil
}

// 2. Output Guardrail - 防敏感信息泄漏
type OutputGuardrail struct {
	emailPattern     *regexp.Regexp
	ssnPattern       *regexp.Regexp
	creditCardPattern *regexp.Regexp
	blacklistedWords []string
}

func (g *OutputGuardrail) Check(ctx context.Context, input interface{}) (bool, string, error) {
	output := input.(string)
	
	// 检查邮箱
	if g.emailPattern.MatchString(output) {
		// 可以选择：拒绝、脱敏、或允许
		return false, "output contains email addresses", nil
	}
	
	// 检查敏感词
	for _, word := range g.blacklistedWords {
		if strings.Contains(output, word) {
			return false, fmt.Sprintf("output contains blacklisted word: %s", word), nil
		}
	}
	
	return true, "", nil
}

// 3. Tool Guardrail - 防高风险工具
type ToolGuardrail struct {
	highRiskTools map[string]bool  // 需要approval的
	blockedTools  map[string]bool  // 完全禁止
}

func (g *ToolGuardrail) Check(ctx context.Context, input interface{}) (bool, string, error) {
	toolCall := input.(*ToolCall)
	
	if g.blockedTools[toolCall.Name] {
		return false, "tool is blocked", nil
	}
	
	if g.highRiskTools[toolCall.Name] {
		// 需要approval，走Day 17的流程
		return true, "requires_approval", nil
	}
	
	return true, "", nil
}

// 4. Retrieval Guardrail - 防越权访问
type RetrievalGuardrail struct {
	aclStore ACLStore  // 权限数据库
}

func (g *RetrievalGuardrail) Check(ctx context.Context, input interface{}) (bool, string, error) {
	chunks := input.([]Chunk)
	user := getUserFromContext(ctx)
	
	for _, chunk := range chunks {
		// 检查用户是否有权访问这个文档
		hasAccess, _ := g.aclStore.HasAccess(user.ID, chunk.DocID)
		if !hasAccess {
			// 从结果里移除
			return false, fmt.Sprintf("unauthorized access to %s", chunk.DocID), nil
		}
	}
	
	return true, "", nil
}

// Guardrail管理器
type GuardrailManager struct {
	guardrails map[GuardrailType]Guardrail
}

func (gm *GuardrailManager) CheckInput(ctx context.Context, input string) error {
	if passed, reason, _ := gm.guardrails[TypeInput].Check(ctx, input); !passed {
		slog.Warn("input guardrail blocked", "reason", reason)
		return fmt.Errorf("input guardrail failed: %s", reason)
	}
	return nil
}

func (gm *GuardrailManager) CheckOutput(ctx context.Context, output string) error {
	if passed, reason, _ := gm.guardrails[TypeOutput].Check(ctx, output); !passed {
		slog.Warn("output guardrail blocked", "reason", reason)
		return fmt.Errorf("output guardrail failed: %s", reason)
	}
	return nil
}

func (gm *GuardrailManager) CheckRetrieval(ctx context.Context, chunks []Chunk) ([]Chunk, error) {
	filtered := chunks
	// 应用guardrail过滤
	return filtered, nil
}

// internal/rbac/rbac.go
type Role string

const (
	RoleGuest      Role = "guest"
	RoleL1Support  Role = "support_l1"
	RoleL2Support  Role = "support_l2"
	RoleAdmin      Role = "admin"
)

type Permission string

const (
	PermViewFAQ           Permission = "view_faq"
	PermViewTickets       Permission = "view_tickets"
	PermCreateTicket      Permission = "create_ticket"
	PermModifyTicket      Permission = "modify_ticket"
	PermDeleteTicket      Permission = "delete_ticket"
	PermViewReports       Permission = "view_reports"
	PermManageUsers       Permission = "manage_users"
)

// 权限矩阵
var rolePermissions = map[Role][]Permission{
	RoleGuest: {
		PermViewFAQ,
	},
	RoleL1Support: {
		PermViewFAQ,
		PermViewTickets,
		PermCreateTicket,
	},
	RoleL2Support: {
		PermViewFAQ,
		PermViewTickets,
		PermCreateTicket,
		PermModifyTicket,
		PermViewReports,
	},
	RoleAdmin: {
		PermViewFAQ,
		PermViewTickets,
		PermCreateTicket,
		PermModifyTicket,
		PermDeleteTicket,
		PermViewReports,
		PermManageUsers,
	},
}

type RBAC struct {
	permissions map[Role][]Permission
}

func (r *RBAC) HasPermission(role Role, perm Permission) bool {
	perms := r.permissions[role]
	for _, p := range perms {
		if p == perm {
			return true
		}
	}
	return false
}

// 在HTTP handler中检查权限
func (h *Handler) CreateTicket(w http.ResponseWriter, r *http.Request) {
	user := getCurrentUser(r)
	
	if !h.rbac.HasPermission(user.Role, PermCreateTicket) {
		http.Error(w, "insufficient permission", http.StatusForbidden)
		return
	}
	
	// ... 创建工单
}

// 重要：权限检查要在RAG阶段做！
type RetrievalWithRBAC struct {
	retriever RAGRetriever
	rbac      RBAC
	documentACL DocumentACLStore
}

func (r *RetrievalWithRBAC) Search(ctx context.Context, query string) ([]Chunk, error) {
	user := getUserFromContext(ctx)
	chunks, _ := r.retriever.Search(ctx, query)
	
	// 过滤：只返回用户有权访问的文档
	var allowedChunks []Chunk
	for _, chunk := range chunks {
		docPerm, _ := r.documentACL.GetDocumentPermission(chunk.DocID, user.ID)
		if docPerm == PermissionAllow {
			allowedChunks = append(allowedChunks, chunk)
		} else {
			slog.Warn("chunk filtered by RBAC", "chunk_id", chunk.ID, "user_id", user.ID)
		}
	}
	
	return allowedChunks, nil
}

// 文档级ACL
type DocumentACL struct {
	DocumentID string
	Role       Role
	Permission DocumentPermission
}

type DocumentPermission string

const (
	PermissionAllow DocumentPermission = "allow"
	PermissionDeny  DocumentPermission = "deny"
)

// 示例：只有L2及以上才能看HR文档
var documentACLRules = []DocumentACL{
	{DocumentID: "hr_salary.md", Role: RoleL1Support, Permission: PermissionDeny},
	{DocumentID: "hr_salary.md", Role: RoleL2Support, Permission: PermissionAllow},
	{DocumentID: "hr_salary.md", Role: RoleAdmin, Permission: PermissionAllow},
}

// cmd/mcp-server/main.go
package main

import (
	"github.com/mark3labs/mcp-go/mcp"
	"github.com/mark3labs/mcp-go/server"
)

func main() {
	// 创建MCP服务器
	s := server.NewServer("agent-mcp-server")
	
	// 注册工具
	s.AddTool(&mcp.Tool{
		Name:        "search_docs",
		Description: "Search knowledge base with hybrid retrieval",
		InputSchema: map[string]interface{}{
			"type": "object",
			"properties": map[string]interface{}{
				"query": map[string]interface{}{
					"type":        "string",
					"description": "Search query",
				},
				"top_k": map[string]interface{}{
					"type":    "integer",
					"default": 5,
				},
			},
			"required": []string{"query"},
		},
		Handler: handleSearchDocs,
	})
	
	s.AddTool(&mcp.Tool{
		Name:        "get_ticket",
		Description: "Get ticket details by ID",
		InputSchema: map[string]interface{}{
			"type": "object",
			"properties": map[string]interface{}{
				"ticket_id": map[string]interface{}{
					"type":        "string",
					"description": "Ticket ID",
				},
			},
			"required": []string{"ticket_id"},
		},
		Handler: handleGetTicket,
	})
	
	s.AddTool(&mcp.Tool{
		Name:        "create_ticket_draft",
		Description: "Create a ticket draft (requires approval)",
		InputSchema: map[string]interface{}{
			"type": "object",
			"properties": map[string]interface{}{
				"title": map[string]interface{}{
					"type":        "string",
					"description": "Ticket title",
				},
				"description": map[string]interface{}{
					"type": "string",
				},
				"priority": map[string]interface{}{
					"type": "string",
					"enum": []string{"low", "medium", "high"},
				},
			},
			"required": []string{"title", "description"},
		},
		Handler: handleCreateTicketDraft,
	})
	
	// 启动服务器
	s.Start()
}

func handleSearchDocs(args map[string]interface{}) (interface{}, error) {
	query := args["query"].(string)
	topK := args["top_k"].(float64)
	
	// 调用Go的RAG系统
	results, _ := ragEngine.Search(context.Background(), query, int(topK))
	
	return map[string]interface{}{
		"results": results,
		"count":   len(results),
	}, nil
}

func handleGetTicket(args map[string]interface{}) (interface{}, error) {
	ticketID := args["ticket_id"].(string)
	
	ticket, err := ticketService.GetTicket(context.Background(), ticketID)
	if err != nil {
		return nil, err
	}
	
	return ticket, nil
}

func handleCreateTicketDraft(args map[string]interface{}) (interface{}, error) {
	// 这会生成一个PendingAction
	// MCP client（如Claude）会收到draft，然后调用approval API
	
	draft := TicketDraft{
		Title:       args["title"].(string),
		Description: args["description"].(string),
		Priority:    args["priority"].(string),
	}
	
	// 创建PendingAction而不是直接创建
	action, _ := approvalStore.Create(context.Background(), &PendingAction{
		ActionType: "create_ticket",
		ActionData: marshalJSON(draft),
	})
	
	return map[string]interface{}{
		"draft_id": action.ID,
		"status":   "pending_approval",
		"action":   draft,
	}, nil
}