Phần 6: Sự Trỗi Dậy Của AI Agents: Từ RAG Thụ Động Đến Thực Thi Tự Hành

package main

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"sync/atomic"
	"time"
)

// MCPToolDefinition định nghĩa đặc tả công cụ theo chuẩn MCP 2.0.
type MCPToolDefinition struct {
	Name        string          `json:"name"`
	Description string          `json:"description"`
	InputSchema json.RawMessage `json:"input_schema"`
}

// JSONRPCRequest đóng gói yêu cầu chuẩn JSON-RPC 2.0 qua HTTP/SSE.
type JSONRPCRequest struct {
	JSONRPC string          `json:"jsonrpc"`
	ID      int64           `json:"id"`
	Method  string          `json:"method"`
	Params  json.RawMessage `json:"params"`
}

// JSONRPCResponse đại diện phản hồi từ MCP Server.
type JSONRPCResponse struct {
	JSONRPC string          `json:"jsonrpc"`
	ID      int64           `json:"id"`
	Result  json.RawMessage `json:"result,omitempty"`
	Error   *RPCError       `json:"error,omitempty"`
}

type RPCError struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

// TokenBudgetTracker theo dõi lượng token tiêu thụ với cơ chế ngắt mạch tự động.
type TokenBudgetTracker struct {
	consumedTokens int64
	maxTokens      int64
}

func NewTokenBudgetTracker(maxTokens int64) *TokenBudgetTracker {
	return &TokenBudgetTracker{maxTokens: maxTokens}
}

func (t *TokenBudgetTracker) Add(tokens int64) error {
	current := atomic.AddInt64(&t.consumedTokens, tokens)
	if current > t.maxTokens {
		return fmt.Errorf("vượt ngưỡng ngân sách token: đã tiêu thụ %d > giới hạn %d", current, t.maxTokens)
	}
	return nil
}

func (t *TokenBudgetTracker) Consumed() int64 {
	return atomic.LoadInt64(&t.consumedTokens)
}

// AgentState duy trì bộ nhớ thực thi và lịch sử các bước lặp.
type AgentState struct {
	Goal          string
	Iterations    int
	Observations  []string
	HistoryHashes map[string]struct{}
	Done          bool
	FinalAnswer   string
}

// MCPClient kết nối tới MCP Server qua giao thức JSON-RPC 2.0.
type MCPClient struct {
	serverURL  string
	httpClient *http.Client
	requestID  int64
}

func NewMCPClient(serverURL string, timeout time.Duration) *MCPClient {
	return &MCPClient{
		serverURL: serverURL,
		httpClient: &http.Client{
			Timeout: timeout,
		},
	}
}

func (c *MCPClient) ExecuteTool(ctx context.Context, toolName string, args json.RawMessage) (string, error) {
	reqID := atomic.AddInt64(&c.requestID, 1)
	paramsPayload, err := json.Marshal(map[string]interface{}{
		"name":      toolName,
		"arguments": args,
	})
	if err != nil {
		return "", fmt.Errorf("lỗi mã hóa tham số công cụ: %w", err)
	}

	rpcReq := JSONRPCRequest{
		JSONRPC: "2.0",
		ID:      reqID,
		Method:  "tools/call",
		Params:  paramsPayload,
	}

	reqBytes, err := json.Marshal(rpcReq)
	if err != nil {
		return "", fmt.Errorf("lỗi đóng gói JSON-RPC: %w", err)
	}

	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, c.serverURL, bytes.NewReader(reqBytes))
	if err != nil {
		return "", fmt.Errorf("lỗi tạo HTTP request: %w", err)
	}
	httpReq.Header.Set("Content-Type", "application/json")

	resp, err := c.httpClient.Do(httpReq)
	if err != nil {
		return "", fmt.Errorf("gọi MCP RPC thất bại: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return "", fmt.Errorf("MCP server báo lỗi (HTTP %d): %s", resp.StatusCode, string(body))
	}

	var rpcResp JSONRPCResponse
	if err := json.NewDecoder(resp.Body).Decode(&rpcResp); err != nil {
		return "", fmt.Errorf("lỗi giải mã phản hồi RPC: %w", err)
	}

	if rpcResp.Error != nil {
		return "", fmt.Errorf("RPC error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message)
	}

	return string(rpcResp.Result), nil
}

// AgentRuntime quản lý và thực thi vòng lặp ReAct.
type AgentRuntime struct {
	mcpClient *MCPClient
	tracker   *TokenBudgetTracker
	maxIter   int
}

func NewAgentRuntime(mcpClient *MCPClient, tracker *TokenBudgetTracker, maxIter int) *AgentRuntime {
	return &AgentRuntime{
		mcpClient: mcpClient,
		tracker:   tracker,
		maxIter:   maxIter,
	}
}

func (r *AgentRuntime) Step(ctx context.Context, state *AgentState) error {
	state.Iterations++
	if state.Iterations > r.maxIter {
		return errors.New("rào chắn bảo vệ: đạt giới hạn số lần lặp tối đa mà không hội tụ")
	}

	// Bước 1: Mô phỏng LLM suy luận (Tư duy & Lựa chọn công cụ)
	toolName, toolArgs, thought, tokensUsed := r.evaluateLLMReasoning(state)
	if err := r.tracker.Add(tokensUsed); err != nil {
		return fmt.Errorf("agent tạm dừng: %w", err)
	}

	log.Printf("[Lần lặp %d] Tư duy: %s | Công cụ: %s (Đã dùng: %d tokens)",
		state.Iterations, thought, toolName, r.tracker.Consumed())

	if toolName == "FinalizeAnswer" {
		state.Done = true
		state.FinalAnswer = string(toolArgs)
		return nil
	}

	// Bước 2: Thực thi công cụ với hạn định thời gian từng bước
	stepCtx, cancel := context.WithTimeout(ctx, 4*time.Second)
	defer cancel()

	obs, err := r.mcpClient.ExecuteTool(stepCtx, toolName, toolArgs)
	if err != nil {
		obs = fmt.Sprintf("Lỗi thực thi công cụ %s: %v", toolName, err)
	}

	// Bước 3: Phát hiện vòng lặp qua mã băm SHA-256
	hasher := sha256.New()
	hasher.Write([]byte(toolName + ":" + obs))
	obsHash := hex.EncodeToString(hasher.Sum(nil))

	if _, exists := state.HistoryHashes[obsHash]; exists {
		log.Printf("[Cảnh báo] Phát hiện lặp ngữ nghĩa trong vòng lặp. Kích hoạt định hướng tái lập kế hoạch.")
		obs = fmt.Sprintf("%s | [CẢNH BÁO: Quan sát lặp lại, hãy đổi chiến lược tiếp cận]", obs)
	}
	state.HistoryHashes[obsHash] = struct{}{}
	state.Observations = append(state.Observations, fmt.Sprintf("[%s]: %s", toolName, obs))

	return nil
}

// evaluateLLMReasoning mô phỏng quá trình ra quyết định của LLM.
func (r *AgentRuntime) evaluateLLMReasoning(state *AgentState) (string, json.RawMessage, string, int64) {
	switch len(state.Observations) {
	case 0:
		return "vector_search",
			json.RawMessage(`{"collection":"bao_cao_tai_chinh","query":"Biên lợi nhuận hoạt động Q3 và đối thủ cạnh tranh"}`),
			"Bắt đầu tra cứu tài liệu báo cáo quý đã qua kiểm duyệt để lấy biên lợi nhuận.",
			450
	case 1:
		return "sql_query",
			json.RawMessage(`{"query":"SELECT competitor_id, margin_pct FROM quarterly_competitors WHERE period='2026-Q3'"}`),
			"Đã có thông tin định tính; truy vấn số liệu biên lợi nhuận đối thủ chính xác từ kho phân tích.",
			620
	default:
		return "FinalizeAnswer",
			json.RawMessage(`"Biên lợi nhuận hoạt động Q3 của chúng ta đạt 24.2%, vượt trội so với Đối thủ A (21.5%) và Đối thủ B (18.9%) dù gặp gián đoạn chuỗi cung ứng."`),
			"Đã tổng hợp đủ dữ liệu định lượng và định tính; kết thúc tiến trình.",
			380
	}
}

func main() {
	tracker := NewTokenBudgetTracker(8000)
	mcpClient := NewMCPClient("http://localhost:8080/mcp/rpc", 5*time.Second)
	runtime := NewAgentRuntime(mcpClient, tracker, 6)

	state := &AgentState{
		Goal:          "Phân tích biên lợi nhuận Q3 so với đối thủ cạnh tranh.",
		HistoryHashes: make(map[string]struct{}),
	}

	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
	defer cancel()

	for !state.Done {
		if err := runtime.Step(ctx, state); err != nil {
			log.Fatalf("Agent runtime dừng đột ngột: %v", err)
		}
	}

	fmt.Printf("\n================ KẾT QUẢ TỔNG HỢP CUỐI CÙNG ================\n%s\n", state.FinalAnswer)
	fmt.Printf("Tổng số token tiêu thụ: %d\n", tracker.Consumed())
}

"""
Môi trường thực thi Supervisor Đa Tác Nhân Doanh Nghiệp với Cổng Can Thiệp Con Người (HITL).
Yêu cầu: Python 3.12+, pydantic >= 2.6.0
"""

import asyncio
from enum import StrEnum
from typing import Annotated, Any, Dict, List, Optional
from pydantic import BaseModel, Field


class AgentActionType(StrEnum):
    RESEARCH = "research"
    ANALYZE_SQL = "analyze_sql"
    EXECUTE_TRANSACTION = "execute_transaction"
    FINALIZE = "finalize"


class ToolCall(BaseModel):
    tool_name: str
    arguments: Dict[str, Any]
    requires_human_approval: bool = False


class AgentStepResult(BaseModel):
    action: AgentActionType
    tool_call: Optional[ToolCall] = None
    thought: str
    output: Optional[str] = None


class WorkflowState(BaseModel):
    objective: str
    iteration: int = 0
    max_iterations: int = 6
    history: List[AgentStepResult] = Field(default_factory=list)
    pending_approval: Optional[ToolCall] = None
    is_completed: bool = False
    final_output: Optional[str] = None


class EnterpriseSupervisor:
    """
    Điều phối các agent chuyên biệt và thực thi rào chắn phê duyệt của con người.
    """

    def __init__(self, token_limit: int = 12000):
        self.token_limit = token_limit
        self.total_tokens_spent = 0

    async def run_research_subagent(self, query: str) -> str:
        await asyncio.sleep(0.05)
        return "Kết quả nghiên cứu: Gián đoạn nhà cung cấp Q3 ảnh hưởng 12% nguồn linh kiện."

    async def run_sql_subagent(self, query: str) -> str:
        await asyncio.sleep(0.05)
        return "Kết quả SQL: [{đơn vị: 'Nội bộ', biên độ: 0.242}, {đơn vị: 'Đối thủ A', biên độ: 0.215}]"

    async def run_transaction_subagent(self, payload: Dict[str, Any]) -> str:
        await asyncio.sleep(0.05)
        return "Giao dịch thành công: Bản ghi kiểm toán #TX-89210 đã ghi nhận vào sổ cái."

    async def plan_next_step(self, state: WorkflowState) -> AgentStepResult:
        """
        Mô phỏng LLM Supervisor xác định node tiếp theo trong đồ thị thực thi.
        """
        step_idx = state.iteration
        if step_idx == 0:
            return AgentStepResult(
                action=AgentActionType.RESEARCH,
                thought="Cần tra cứu nhật ký sự cố doanh nghiệp để làm rõ bối cảnh gián đoạn.",
                tool_call=ToolCall(
                    tool_name="vector_search",
                    arguments={"query": "báo cáo sự cố chuỗi cung ứng Q3"},
                    requires_human_approval=False,
                ),
            )
        elif step_idx == 1:
            return AgentStepResult(
                action=AgentActionType.ANALYZE_SQL,
                thought="Lấy số liệu biên lợi nhuận chính xác từ kho dữ liệu phân tích.",
                tool_call=ToolCall(
                    tool_name="clickhouse_query",
                    arguments={"sql": "SELECT entity, margin FROM metrics WHERE period='2026-Q3'"},
                    requires_human_approval=False,
                ),
            )
        elif step_idx == 2:
            return AgentStepResult(
                action=AgentActionType.EXECUTE_TRANSACTION,
                thought="Ghi nhận bản đánh giá rủi ro điều hành vào sổ cái kiểm toán tuân thủ.",
                tool_call=ToolCall(
                    tool_name="commit_audit_entry",
                    arguments={"module": "risk_management", "status": "FLAGGED_MODERATE"},
                    requires_human_approval=True,  # Kích hoạt HITL
                ),
            )
        else:
            return AgentStepResult(
                action=AgentActionType.FINALIZE,
                thought="Đã tổng hợp toàn bộ thông tin nghiên cứu, số liệu và giao dịch kiểm toán.",
                output="Bản tóm tắt điều hành: Biên lợi nhuận Q3 đạt 24.2%. Rủi ro đã kiểm soát. Đã ghi nhận kiểm toán #TX-89210.",
            )

    async def execute_step(self, state: WorkflowState) -> WorkflowState:
        if state.iteration >= state.max_iterations:
            state.is_completed = True
            state.final_output = "Chấm dứt: Đã đạt số bước lặp tối đa của supervisor."
            return state

        step_decision = await self.plan_next_step(state)
        state.iteration += 1

        # Đánh giá cổng can thiệp của con người
        if step_decision.tool_call and step_decision.tool_call.requires_human_approval:
            state.pending_approval = step_decision.tool_call
            print(f"\n[CẢNH BÁO HITL] Công cụ '{step_decision.tool_call.tool_name}' cần phê duyệt từ con người!")
            return state

        # Thực thi tự động nếu không cần phê duyệt
        if step_decision.action == AgentActionType.RESEARCH:
            output = await self.run_research_subagent(step_decision.tool_call.arguments["query"])
            step_decision.output = output
        elif step_decision.action == AgentActionType.ANALYZE_SQL:
            output = await self.run_sql_subagent(step_decision.tool_call.arguments["sql"])
            step_decision.output = output
        elif step_decision.action == AgentActionType.FINALIZE:
            state.is_completed = True
            state.final_output = step_decision.output

        state.history.append(step_decision)
        return state

    async def approve_and_resume(self, state: WorkflowState, approved: bool) -> WorkflowState:
        """
        Tiếp tục đồ thị thực thi sau khi nhận được phản hồi từ người vận hành.
        """
        if not state.pending_approval:
            raise ValueError("Không có hành động nào đang chờ phê duyệt.")

        pending_call = state.pending_approval
        state.pending_approval = None

        if not approved:
            step_result = AgentStepResult(
                action=AgentActionType.EXECUTE_TRANSACTION,
                thought="Người vận hành từ chối thực thi công cụ.",
                output="Thao tác bị từ chối: Giao dịch đã bị hủy bỏ an toàn.",
            )
            state.history.append(step_result)
            return state

        # Thực thi nếu được phê duyệt
        output = await self.run_transaction_subagent(pending_call.arguments)
        step_result = AgentStepResult(
            action=AgentActionType.EXECUTE_TRANSACTION,
            tool_call=pending_call,
            thought="Người vận hành đã phê duyệt thực thi công cụ.",
            output=output,
        )
        state.history.append(step_result)
        return state


async def main():
    supervisor = EnterpriseSupervisor()
    state = WorkflowState(objective="Điều tra biến động chuỗi cung ứng Q3 và cập nhật sổ cái kiểm toán.")

    # Giai đoạn 1: Tự hành thực hiện các bước cho đến cổng HITL
    while not state.is_completed and state.pending_approval is None:
        state = await supervisor.execute_step(state)
        latest = state.history[-1] if state.history else None
        if latest:
            print(f"[Bước {state.iteration}] {latest.action.value}: {latest.thought}")

    # Giai đoạn 2: Người vận hành phê duyệt
    if state.pending_approval:
        print(f"Phê duyệt lệnh thực thi: {state.pending_approval.model_dump_json()}")
        state = await supervisor.approve_and_resume(state, approved=True)

    # Giai đoạn 3: Hoàn tất quy trình
    while not state.is_completed:
        state = await supervisor.execute_step(state)

    print(f"\n[Kết quả quy trình] {state.final_output}")


if __name__ == "__main__":
    asyncio.run(main())