Agent Development Guide

// Spring AI 接入 OpenAI 示例
@Bean
public ChatClient chatClient(ChatClient.Builder builder) {
    return builder
        .defaultSystem("你是一个专业的 Java 技术助手")
        .build();
}

// 调用
String response = chatClient.prompt()
    .user("解释 AQS 的核心原理")
    .call()
    .content();

// LangChain4j Tool 定义
public class DatabaseQueryTool {

    @Tool("查询用户订单列表，支持按状态和时间范围过滤")
    public List<Order> queryOrders(
        @P("用户ID") Long userId,
        @P("订单状态: PENDING/COMPLETED/CANCELLED") String status,
        @P("开始时间 yyyy-MM-dd") String startDate
    ) {
        // 实现查询逻辑
        return orderService.query(userId, status, startDate);
    }
}

// 并行工具调用 - 类比 CompletableFuture.allOf()
@Component
public class ParallelToolExecutor {

    private final ExecutorService toolExecutor =
        new ThreadPoolExecutor(
            10, 50, 60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            new ThreadFactory() { /* 命名线程 */ },
            new ThreadPoolExecutor.CallerRunsPolicy() // 背压策略
        );

    public Map<String, Object> executeParallel(List<ToolCall> toolCalls) {
        List<CompletableFuture<Map.Entry<String, Object>>> futures =
            toolCalls.stream()
                .map(call -> CompletableFuture
                    .supplyAsync(() -> executeOne(call), toolExecutor)
                    .orTimeout(5, TimeUnit.SECONDS)
                    .exceptionally(ex -> Map.entry(call.id(), "ERROR: " + ex.getMessage()))
                )
                .toList();

        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)))
            .join();
    }
}

// 基于 Spring AI 的对话历史管理
@Service
public class ConversationMemoryService {

    // 类比 Redis 的 List 结构存储消息
    private final ChatMemory chatMemory = MessageWindowChatMemory.builder()
        .maxMessages(20)  // 滑动窗口大小
        .build();

    public String chat(String sessionId, String userInput) {
        return chatClient.prompt()
            .user(userInput)
            .advisors(new MessageChatMemoryAdvisor(chatMemory, sessionId, 10))
            .call()
            .content();
    }
}

// Token 计算与裁剪 - 类比内存分页管理
public class TokenBudgetManager {

    private static final int MAX_CONTEXT_TOKENS = 8000;
    private static final int RESPONSE_RESERVE = 1000;
    private static final int AVAILABLE_TOKENS = MAX_CONTEXT_TOKENS - RESPONSE_RESERVE;

    public List<Message> trimHistory(List<Message> history) {
        int usedTokens = 0;
        Deque<Message> result = new ArrayDeque<>();

        for (int i = history.size() - 1; i >= 0; i--) {
            Message msg = history.get(i);
            int tokens = estimateTokens(msg);
            if (usedTokens + tokens > AVAILABLE_TOKENS) break;
            result.addFirst(msg);
            usedTokens += tokens;
        }
        return new ArrayList<>(result);
    }
}

// Spring AI 向量存储操作
@Service
public class LongTermMemoryService {

    private final VectorStore vectorStore;

    // 存储记忆（类比 INSERT + 建索引）
    public void remember(String userId, String content) {
        Document doc = new Document(content,
            Map.of("userId", userId, "timestamp", Instant.now().toString()));
        vectorStore.add(List.of(doc));
    }

    // 语义检索（类比 SELECT WHERE similarity > threshold）
    public List<String> recall(String userId, String query, int topK) {
        SearchRequest request = SearchRequest.query(query)
            .withTopK(topK)
            .withFilterExpression("userId == '" + userId + "'");

        return vectorStore.similaritySearch(request)
            .stream()
            .map(Document::getContent)
            .toList();
    }
}

┌─────────────────────────────────────────────┐
│              Agent Memory 分层               │
├─────────────────────────────────────────────┤
│  L1: 工作记忆 (Working Memory)               │
│      当前对话 context window                 │
│      类比: CPU L1 Cache / JVM 栈帧           │
├─────────────────────────────────────────────┤
│  L2: 会话记忆 (Session Memory)               │
│      Redis 存储，TTL=24h                     │
│      类比: CPU L2 Cache / Redis 热数据       │
├─────────────────────────────────────────────┤
│  L3: 长期记忆 (Episodic Memory)              │
│      Vector DB，永久存储                     │
│      类比: 磁盘 / MySQL 全文索引             │
├─────────────────────────────────────────────┤
│  L4: 知识记忆 (Semantic Memory)              │
│      RAG 知识库（下一章节）                  │
│      类比: 只读文档数据库                    │
└─────────────────────────────────────────────┘

文档摄入流水线（离线）：
原始文档 → 解析 → 分块(Chunking) → Embedding → 向量存储

检索增强流水线（在线）：
用户问题 → Embedding → 向量检索 → 重排序 → 注入 Prompt → LLM → 回答

// Spring AI ETL Pipeline
@Service
public class DocumentIngestionPipeline {

    private final VectorStore vectorStore;
    private final EmbeddingModel embeddingModel;

    public void ingest(Resource documentResource) {
        // 1. 读取文档（支持 PDF/Word/Markdown）
        var reader = new PagePdfDocumentReader(documentResource,
            PdfDocumentReaderConfig.builder()
                .withPageTopMargin(0)
                .withPageExtractedTextFormatter(
                    ExtractedTextFormatter.builder()
                        .withNumberOfBottomTextLinesToDelete(3)
                        .build())
                .build());

        // 2. 分块策略（核心！类比数据库分区策略）
        var splitter = new TokenTextSplitter(
            512,    // chunk size（类比分区大小）
            128,    // overlap（类比分区重叠，保证语义连续性）
            5, 10000, true);

        // 3. ETL & 写入向量库
        List<Document> docs = splitter.apply(reader.get());
        vectorStore.add(docs);  // 自动 Embedding + 存储
    }
}

查询理解层:
  ├── Query Rewriting（查询改写，提高召回率）
  ├── Query Decomposition（复杂问题分解）
  └── HyDE（假设文档嵌入，提高检索精度）

检索层:
  ├── Dense Retrieval（向量检索）
  ├── Sparse Retrieval（BM25 关键词检索）
  └── Hybrid Search（混合检索 = 两者 + RRF 融合）

后处理层:
  ├── Re-ranking（交叉编码器重排序）
  ├── Contextual Compression（上下文压缩）
  └── Citation（来源溯源）

// 混合检索实现
@Service
public class HybridSearchService {

    private final VectorStore vectorStore;
    private final ElasticsearchClient esClient;  // BM25 检索

    public List<Document> hybridSearch(String query, int topK) {
        CompletableFuture<List<Document>> denseFuture =
            CompletableFuture.supplyAsync(() ->
                vectorStore.similaritySearch(SearchRequest.query(query).withTopK(topK * 2)));

        CompletableFuture<List<Document>> sparseFuture =
            CompletableFuture.supplyAsync(() ->
                esClient.bm25Search(query, topK * 2));

        // RRF 融合排序（Reciprocal Rank Fusion）
        return CompletableFuture.allOf(denseFuture, sparseFuture)
            .thenApply(v -> rrfFusion(denseFuture.join(), sparseFuture.join(), topK))
            .join();
    }

    private List<Document> rrfFusion(List<Document> dense,
                                      List<Document> sparse, int topK) {
        // RRF score = Σ 1/(k + rank_i), k=60
        Map<String, Double> scores = new HashMap<>();
        IntStream.range(0, dense.size()).forEach(i ->
            scores.merge(dense.get(i).getId(), 1.0 / (60 + i + 1), Double::sum));
        IntStream.range(0, sparse.size()).forEach(i ->
            scores.merge(sparse.get(i).getId(), 1.0 / (60 + i + 1), Double::sum));

        return scores.entrySet().stream()
            .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
            .limit(topK)
            .map(e -> findDocById(e.getKey(), dense, sparse))
            .toList();
    }
}

// ReAct 循环核心实现
@Service
public class ReActAgent {

    private static final int MAX_ITERATIONS = 10;  // 防止无限循环

    public AgentResult execute(String task, List<Object> tools) {
        List<Message> trajectory = new ArrayList<>();
        trajectory.add(new SystemMessage(buildSystemPrompt(tools)));
        trajectory.add(new UserMessage(task));

        for (int iter = 0; iter < MAX_ITERATIONS; iter++) {
            // Thought: LLM 推理
            ChatResponse response = chatClient.prompt()
                .messages(trajectory)
                .call()
                .chatResponse();

            String llmOutput = response.getResult().getOutput().getContent();
            trajectory.add(new AssistantMessage(llmOutput));

            // 检测是否完成
            if (llmOutput.contains("Final Answer:")) {
                return AgentResult.success(extractFinalAnswer(llmOutput));
            }

            // Action: 解析工具调用
            ToolCall toolCall = parseToolCall(llmOutput);
            if (toolCall == null) {
                return AgentResult.failure("无法解析工具调用");
            }

            // Observation: 执行工具获取结果
            String observation = executeToolWithTimeout(toolCall, 5, TimeUnit.SECONDS);
            trajectory.add(new UserMessage("Observation: " + observation));

            // 记录追踪（可观测性）
            tracer.record(iter, toolCall, observation);
        }

        return AgentResult.failure("超过最大迭代次数");
    }
}

public record ExecutionPlan(
    String goal,
    List<PlanStep> steps,
    Map<String, Object> context  // 步骤间共享状态
) {}

public record PlanStep(
    int stepId,
    String description,
    String toolName,
    Map<String, Object> parameters,
    List<Integer> dependsOn  // 依赖关系，支持并行执行！
) {}

@Service
public class PlanAndExecuteAgent {

    public AgentResult execute(String goal) {
        // Phase 1: Planner - 生成执行计划
        ExecutionPlan plan = plannerLlm.generatePlan(goal);

        // Phase 2: Executor - 按依赖关系并行执行
        Map<Integer, CompletableFuture<StepResult>> futures = new HashMap<>();

        for (PlanStep step : plan.steps()) {
            CompletableFuture<Void> deps = step.dependsOn().isEmpty()
                ? CompletableFuture.completedFuture(null)
                : CompletableFuture.allOf(
                    step.dependsOn().stream()
                        .map(futures::get)
                        .toArray(CompletableFuture[]::new));

            futures.put(step.stepId(), deps.thenApplyAsync(
                v -> executeStep(step, plan.context())));
        }

        // Phase 3: 汇总结果
        return aggregateResults(futures);
    }
}

模式一：层级式（Hierarchical）
┌─────────────────────────────┐
│      Supervisor Agent       │  ← 任务分发、结果汇总
└──────────┬──────────────────┘
    ┌──────┼──────┐
    ↓      ↓      ↓
  Agent1 Agent2 Agent3      ← 专职 Worker Agent

模式二：流水线式（Pipeline）
User → [预处理Agent] → [分析Agent] → [写作Agent] → Result

模式三：辩论式（Debate）
                ┌→ Agent-正方 ─┐
User Request ──→│              ├→ Judge Agent → Final Answer
                └→ Agent-反方 ─┘

// LangGraph4j 工作流定义（类比 Spring State Machine）
@Configuration
public class ResearchWorkflow {

    @Bean
    public StateGraph<ResearchState> researchGraph() {
        return StateGraph.<ResearchState>builder()
            .addNode("planner", plannerAgent::run)
            .addNode("researcher", researcherAgent::run)
            .addNode("writer", writerAgent::run)
            .addNode("reviewer", reviewerAgent::run)

            // 条件路由（类比路由表）
            .addConditionalEdges("reviewer", state ->
                state.reviewScore() >= 8 ? "END" : "writer")

            .addEdge(START, "planner")
            .addEdge("planner", "researcher")
            .addEdge("researcher", "writer")
            .addEdge("writer", "reviewer")
            .build();
    }
}

// 基于消息总线的 Agent 通信（类比 MQ 解耦）
@Service
public class AgentMessageBus {

    private final Map<String, BlockingQueue<AgentMessage>> queues =
        new ConcurrentHashMap<>();

    public void publish(String targetAgentId, AgentMessage message) {
        queues.computeIfAbsent(targetAgentId, k -> new LinkedBlockingQueue<>())
            .offer(message);
    }

    public AgentMessage consume(String agentId, long timeout, TimeUnit unit)
        throws InterruptedException {
        return queues.computeIfAbsent(agentId, k -> new LinkedBlockingQueue<>())
            .poll(timeout, unit);
    }
}

// Micrometer + Prometheus 打点
@Component
public class AgentMetrics {

    private final MeterRegistry registry;
    private final Timer llmLatency;
    private final Counter tokenCounter;
    private final Counter toolSuccessCounter;
    private final Counter toolFailureCounter;

    public AgentMetrics(MeterRegistry registry) {
        this.registry = registry;
        this.llmLatency = Timer.builder("agent.llm.latency")
            .description("LLM API 调用延迟")
            .tag("model", "gpt-4o")
            .publishPercentiles(0.5, 0.95, 0.99)
            .register(registry);
    }

    public <T> T recordLlmCall(Supplier<T> llmCall, String model) {
        return llmLatency.record(() -> {
            try {
                T result = llmCall.get();
                tokenCounter.increment(extractTokenCount(result));
                return result;
            } catch (Exception e) {
                registry.counter("agent.llm.errors", "model", model).increment();
                throw e;
            }
        });
    }
}

// Resilience4j 集成 LLM 熔断
@Bean
public CircuitBreaker llmCircuitBreaker(CircuitBreakerRegistry registry) {
    return registry.circuitBreaker("llm-api", CircuitBreakerConfig.custom()
        .failureRateThreshold(50)           // 失败率超 50% 开启熔断
        .waitDurationInOpenState(Duration.ofSeconds(30))
        .slidingWindowSize(10)
        .build());
}

// 降级策略：主模型熔断时切换备用模型
@Service
public class ResilientLlmClient {

    public String chat(String prompt) {
        return Decorators.ofSupplier(() -> primaryLlm.call(prompt))
            .withCircuitBreaker(llmCircuitBreaker)
            .withRetry(Retry.ofDefaults("llm-retry"))
            .withFallback(List.of(Exception.class),
                e -> fallbackLlm.call(prompt))  // 降级到本地模型
            .decorate()
            .get();
    }
}

// 基于令牌桶的 Token 限流
@Service
public class TokenRateLimiter {

    // 每用户每分钟最多 10000 tokens
    private final Cache<String, RateLimiter> userLimiters = Caffeine.newBuilder()
        .expireAfterAccess(1, TimeUnit.HOURS)
        .build();

    public boolean tryAcquire(String userId, int tokenCount) {
        RateLimiter limiter = userLimiters.get(userId,
            k -> RateLimiter.create(10000.0 / 60)); // tokens/second
        return limiter.tryAcquire(tokenCount);
    }
}

# GitHub Actions - Agent 系统 CI/CD
name: Agent CI/CD Pipeline

on: [push]

jobs:
  test:
    steps:
      - name: 单元测试（Mock LLM）
        run: ./gradlew test

      - name: RAG 评估测试
        run: ./gradlew ragas-eval  # 自动化 RAG 质量评估

      - name: Prompt 回归测试
        run: ./gradlew prompt-regression  # 对比 Prompt 变更前后效果

  build:
    needs: test
    steps:
      - name: 构建镜像
        run: docker build -t agent-service:${{ github.sha }} .

  deploy:
    needs: build
    steps:
      - name: 金丝雀发布（5% 流量）
        run: kubectl set image deployment/agent-service ...

      - name: 监控 15min 后全量发布
        run: ./scripts/canary-promote.sh

┌────────────────────────────────────────────────────────┐
│                    用户接入层                           │
│         Web / WeChat / API (Spring Boot 3)             │
└──────────────────────┬─────────────────────────────────┘
                       ↓
┌────────────────────────────────────────────────────────┐
│                   Agent 编排层                          │
│    Supervisor Agent → [意图识别] → [路由决策]           │
│         ↓                ↓                             │
│   FAQ Agent        Order Agent      Complaint Agent    │
│  (RAG驱动)        (Tool Calling)    (Escalation)       │
└──────────────────────┬─────────────────────────────────┘
                       ↓
┌────────────────────────────────────────────────────────┐
│                   基础设施层                            │
│  Memory: Redis(短期) + Milvus(长期)                    │
│  Knowledge: PGVector + Elasticsearch (混合检索)        │
│  Tools: 订单系统/CRM/工单系统                          │
│  Observability: Prometheus + Grafana + Jaeger          │
└────────────────────────────────────────────────────────┘