Java AI Agent框架ZenoAgent的设计与实现

1. 项目概述：从零构建Java AI Agent的探索之旅

作为一名深耕Java后端开发多年的工程师，我始终对AI Agent的内部运作机制充满好奇。市面上的LangChain和AutoGen等框架虽然功能强大，但直接调用API总让我有种隔靴搔痒的感觉——就像开车却不懂发动机原理。为了真正理解Agent技术的内核，我决定亲手打造一个Java版的AI Agent系统，这就是ZenoAgent项目的由来。

ZenoAgent是一个完全用Java实现的AI Agent框架，核心目标是探索以下几个关键技术点：

• 不依赖现成框架，从零实现ReAct推理循环
• 在分布式环境下实现Human-in-the-loop机制
• 复现类似o1模型的流式思考能力
• 构建鲁棒的错误处理系统

这个项目最特别之处在于，它完全基于Java生态构建：

• 后端：Spring Boot 3 + Java 17
• AI交互：LangChain4j（LLM交互核心）
• 分布式协调：Redisson
• 数据存储：PostgreSQL（主库）+ pgvector（向量搜索）+ Redis（缓存）

2. 核心架构设计

2.1 分层架构与模块划分

采用经典的DDD分层架构，确保各模块高内聚低耦合：

code复制┌───────────────────────────────────────┐
│              API Layer                │
│  (Controller/SSE接口/WebSocket)       │
└───────────────────────────────────────┘
               ↓
┌───────────────────────────────────────┐
│         Application Service           │
│  (业务流程编排/事务管理/权限控制)      │
└───────────────────────────────────────┘
               ↓
┌───────────────────────────────────────┐
│             Core Engine               │
│  (ReAct引擎/状态机/思考引擎/执行器)   │
└───────────────────────────────────────┘
               ↓
┌───────────────────────────────────────┐
│           Infrastructure              │
│(LLM连接/向量搜索/工具调用/持久化)      │
└───────────────────────────────────────┘

2.2 关键技术选型考量

为什么选择LangChain4j而不是直接调用OpenAI API？

• 提供了统一的LLM抽象层，方便切换不同模型提供商
• 内置了常用的Prompt模板和输出解析器
• 对Java流式响应的良好支持

Redis的三种关键用途：

1. 分布式锁：协调多实例间的Agent状态
2. 消息队列：实现跨Pod的Human-in-the-loop
3. 短期记忆：缓存最近几轮的对话上下文

PostgreSQL + pgvector的独特优势：

• 一套系统同时处理结构化数据和向量搜索
• 利用PG的Row-Level Security实现多租户隔离
• 事务特性保证知识库更新的原子性

3. ReAct引擎实现细节

3.1 思维链(CoT)控制策略

为了让LLM按照"先思考后行动"的模式工作，我们设计了特殊的Prompt结构：

java复制// System Prompt示例
String systemPrompt = """
你是一个专业助理，必须严格遵循以下规则：
1. 每次响应必须包含<thinking>和<actions>两部分
2. 在<thinking>中详细分析问题和可用工具
3. 在<actions>中输出严格符合JSON Schema的动作

输出格式示例：
<thinking>
用户想查询天气，可用工具有：
- weather_tool：需要location参数
根据IP推测用户可能在北京，优先尝试该位置
</thinking>
<THINKING_DONE>
<actions>
{
  "actions": [{
    "actionType": "TOOL_CALL",
    "actionName": "weather_tool",
    "parameters": {"location": "北京"}
  }]
}
</actions>
""";

关键创新点：

• <THINKING_DONE>作为分隔符，解决模型"废话"问题
• 流式解析时采用状态机识别思考与动作部分
• 自动补全不完整JSON（实测可提升15%的解析成功率）

3.2 动作类型设计

将Agent能力抽象为四种原子动作：

DIRECT_RESPONSE的优化价值：
当用户输入"你好"这类简单问候时，传统ReAct流程需要：

1. LLM思考(200ms)
2. 决定不调用工具(50ms)
3. 生成回复(300ms)

而通过DIRECT_RESPONSE，这三个步骤在一次LLM调用中完成，实测平均响应时间从550ms降至350ms。

3.3 并发执行机制

对于需要多工具协同的任务，采用Java的CompletableFuture实现并行：

java复制List<CompletableFuture<ActionResult>> futures = actions.stream()
    .map(action -> CompletableFuture.supplyAsync(() -> {
        try {
            return executeAction(action);
        } catch (Exception e) {
            return ActionResult.failure(e.getMessage());
        }
    }, threadPool))
    .toList();

// 等待所有动作完成（超时控制）
List<ActionResult> results = futures.stream()
    .map(f -> f.get(5, TimeUnit.SECONDS))
    .toList();

线程池配置要点：

• 核心线程数 = CPU核心数 × 2
• 最大线程数 = 核心线程数 × 4
• 队列容量 = 100（避免内存溢出）
• 拒绝策略 = 调用者运行（保证重要任务不丢失）

4. 分布式Human-in-the-loop实现

4.1 核心挑战与解决方案

在K8s集群环境中，主要面临的问题：

• Agent推理线程在Pod A运行
• 用户的确认请求可能打到Pod B
• 需要跨实例唤醒挂起的线程

技术方案：基于Redis的分布式信号量

java复制// 等待用户确认（最多60秒）
public ToolConfirmation waitForConfirmation(String sessionId) {
    RBlockingQueue<ToolConfirmation> queue = redissonClient
        .getBlockingQueue("confirmation:" + sessionId);
    return queue.poll(60, TimeUnit.SECONDS);
}

// 用户确认后触发（任何Pod都可调用）
public void submitConfirmation(String sessionId, boolean approved) {
    RBlockingQueue<ToolConfirmation> queue = redissonClient
        .getBlockingQueue("confirmation:" + sessionId);
    queue.put(new ToolConfirmation(approved));
}

4.2 超时与异常处理

考虑各种边缘情况：

1. 用户不操作：60秒后自动拒绝
2. 网络中断：前端心跳检测，自动重连
3. Pod重启：Redis持久化队列+事务日志

确认流程的状态机设计：

mermaid复制stateDiagram
    [*] --> Pending
    Pending --> Approved: 用户同意
    Pending --> Rejected: 用户拒绝
    Pending --> Timeout: 60秒未响应
    Approved --> Executing
    Rejected --> [*]
    Timeout --> [*]
    Executing --> [*]

5. 流式思考的实现与优化

5.1 两种实现方案对比

协议不兼容问题实录：
当使用Ollama部署DeepSeek-R1时，发现：

• Ollama返回的思考字段是reasoning
• 但LangChain4j预期字段是reasoning_content
  导致思考内容被丢弃的严重问题。

解决方案：
自定义响应处理器适配不同供应商：

java复制public class CustomStreamHandler implements StreamingResponseHandler {
    @Override
    public void onNext(String token) {
        // 尝试解析不同供应商的字段
        JsonNode node = objectMapper.readTree(token);
        String thought = node.path("reasoning").asText(); 
        if (thought.isEmpty()) {
            thought = node.path("reasoning_content").asText();
        }
        if (!thought.isEmpty()) {
            sseEmitter.send(thought);
        }
    }
}

5.2 标签泄露防护

问题现象：
前端偶尔会闪现</think这样的残缺标签。

根本原因：
流式传输时，标签可能被拆分成多个chunk：

1. 先收到</think
2. 引擎未识别为完整标签
3. 转发给前端
4. 然后收到ing>部分

解决方案：前缀缓冲机制

java复制String buffer = "";
for (String chunk : stream) {
    buffer += chunk;
    // 检查是否包含不完整标签
    int partialTag = findPartialTag(buffer);
    if (partialTag > 0) {
        String safePart = buffer.substring(0, buffer.length() - partialTag);
        emitToFrontend(safePart);
        buffer = buffer.substring(safePart.length());
    } else {
        emitToFrontend(buffer);
        buffer = "";
    }
}

6. RAG知识库的工程实践

6.1 文档处理流水线

java复制public void ingestDocument(File file, String knowledgeId) {
    // 1. 用Tika解析各种格式
    Content content = tikaParser.parse(file);
    
    // 2. 递归式文本分割
    List<TextSegment> segments = splitter.split(content.text());
    
    // 3. 注入元数据
    segments.forEach(seg -> seg.metadata().put("knowledgeId", knowledgeId));
    
    // 4. 向量化存储
    List<Embedding> embeddings = embeddingModel.embedAll(segments);
    embeddingStore.addAll(embeddings, segments);
}

分块策略优化：

• 代码类：按函数/类分割
• 文档类：重叠分块（前1/3与后1/3重叠）
• 表格类：保持行列结构

6.2 混合检索技术

结合多种检索方式提升召回率：

1. 向量搜索：语义相似度

sql复制SELECT content FROM chunks 
WHERE knowledge_id = ? 
ORDER BY embedding <=> ?::vector
LIMIT 3

2. 关键词检索：精确匹配术语

java复制FullTextEntityManager.fullTextSearch()
    .forEntities(Chunk.class)
    .matching("some keywords")
    .build();

3. 元数据过滤：按作者/日期等筛选

性能对比：

7. 错误处理与自愈机制

7.1 错误分类处理策略

7.2 自愈循环实现

java复制public ActionResult executeWithRetry(AgentAction action, int maxRetry) {
    for (int i = 0; i < maxRetry; i++) {
        try {
            return executeAction(action);
        } catch (Exception e) {
            log.warn("Action failed: {}", e.getMessage());
            if (i == maxRetry - 1) break;
            
            // 将错误信息反馈给LLM
            action = rethinkAction(action, e);
            Thread.sleep(1000 * (i + 1)); // 退避等待
        }
    }
    return ActionResult.failure("Max retry exceeded");
}

private AgentAction rethinkAction(AgentAction action, Exception e) {
    String prompt = String.format("""
        上次执行失败：%s
        请修正以下动作并重新输出：
        %s
        """, e.getMessage(), action);
    return llmClient.generateNewAction(prompt);
}

8. 性能优化实战记录

8.1 关键指标对比

优化前 vs 优化后：

8.2 最有效的三项优化

1. 预编译Prompt模板

java复制// 避免每次请求都解析Prompt
private static final CompiledPrompt reactPrompt = PromptTemplate
    .compile("""
        系统指令：{{system}}
        历史记录：{{history}}
        当前任务：{{task}}
        """);

2. 向量缓存层

java复制@Cacheable(value = "embeddings", key = "#text.hashCode()")
public Embedding getEmbedding(String text) {
    return embeddingModel.embed(text);
}

3. SSE连接复用

java复制// 保持长连接而不是每次请求新建
@GetMapping("/stream")
public SseEmitter stream(@RequestParam String sessionId) {
    return connectionManager.getOrCreate(sessionId);
}

9. 部署架构与运维实践

9.1 生产环境部署方案

code复制┌─────────────────┐    ┌─────────────────┐
│   Load Balancer  │    │   PostgreSQL    │
└────────┬────────┘    └────────┬────────┘
         │                       │
┌────────▼────────┐    ┌────────▼────────┐
│  Spring Boot Pods│    │   Redis Cluster │
│  (无状态部署)    │    └─────────────────┘
└────────┬────────┘             │
         │                      │
┌────────▼────────┐    ┌────────▼────────┐
│  前端静态资源    │    │  模型推理服务   │
│  (Nginx托管)     │    │ (GPU节点)      │
└─────────────────┘    └─────────────────┘

关键配置参数：

• JVM堆内存：不超过容器内存的70%
• 线程池大小：CPU核心数 × 2
• Redis连接池：每个Pod 20-50个连接
• PostgreSQL连接池：每个Pod 10-20个连接

9.2 监控指标配置

Prometheus采集的关键指标：

yaml复制- job_name: 'zeno-agent'
  metrics_path: '/actuator/prometheus'
  scrape_interval: 15s
  static_configs:
    - targets: ['app:8080']

Grafana监控看板包含：

1. 请求成功率/延迟百分位
2. LLM调用次数/Token消耗
3. 工具调用耗时分布
4. 线程池活跃度
5. 向量搜索缓存命中率

10. 典型问题排查指南

10.1 常见问题速查表

10.2 性能问题诊断流程

1. 确定瓶颈位置

   • 使用Arthas的trace命令跟踪调用链
   • 检查各阶段耗时：思考/工具调用/结果组装

2. 资源分析

   bash复制# CPU使用率
   top -H -p <pid>
   # 内存分布
   jmap -histo <pid>
   

3. 针对性优化

   • CPU密集型：增加线程池/缓存结果
   • IO密集型：调整连接池/批量操作
   • 内存问题：优化分块策略/限制并发

11. 项目演进与未来规划

11.1 已实现的核心特性

• 完整的ReAct循环引擎
• 分布式Human-in-the-loop
• 流式思考可视化
• 自愈式错误处理
• 动态工具热加载

11.2 路线图

短期（3个月）：

• 多模态支持（图像/音频）
• 工作流编排引擎
• 更细粒度的权限控制

中期（6个月）：

• Agent微调框架
• 自动化测试套件
• 可视化编排界面

长期（1年）：

• 多Agent协作系统
• 强化学习优化模块
• 领域专用Agent市场

12. 开发者指南

12.1 快速开始

1. 克隆仓库：

bash复制git clone https://github.com/Johnnyjin-haolin/ZenoAgent.git

2. 配置环境：

yaml复制# application.yml
llm:
  provider: openai # 或qwen/deepseek
  api-key: ${LLM_API_KEY}

3. 启动服务：

bash复制./gradlew bootRun

12.2 扩展开发示例

添加新工具的步骤：

1. 实现Tool接口：

java复制public class WeatherTool implements Tool {
    @Override
    public String name() { return "weather_tool"; }
    
    @Override
    public ActionResult execute(Map<String, Object> params) {
        String location = (String) params.get("location");
        // 调用真实天气API
        return ActionResult.success(weatherService.query(location));
    }
}

2. 注册到上下文：

java复制@Bean
public ToolManager toolManager() {
    return new ToolManager(List.of(
        new WeatherTool(),
        // 其他工具...
    ));
}

3. 更新Prompt描述：

json复制{
  "name": "weather_tool",
  "description": "查询指定地点的天气情况",
  "parameters": {
    "location": {
      "type": "string",
      "description": "城市名称，如'北京'"
    }
  }
}

13. 经验总结与建议

13.1 最重要的三条教训

1. LLM的输出不可靠

   • 必须设计多重验证机制
   • 任何解析都要考虑容错
   • 重要操作必须人工确认

2. 分布式环境复杂性

   • 所有状态必须外部化
   • 考虑网络分区场景
   • 实现幂等操作

3. 性能陷阱无处不在

   • 流式响应要及时flush
   • 避免大上下文反复传输
   • 合理设置超时时间

13.2 给Java开发者的建议

1. 充分利用Java强类型优势

   • 用Record定义动作类型
   • 使用Bean Validation校验参数
   • 通过泛型保证类型安全

2. Spring生态整合技巧

   • 用@Async处理耗时操作
   • @Retryable实现自动重试
   • @Cacheable缓存向量结果

3. 调试与监控

   • 结构化日志必备
   • 分布式追踪集成
   • 自定义健康指标

14. 资源与社区

14.1 学习资料推荐

• 官方文档：

  • LangChain4j
  • Spring AI

• 书籍：

  • 《Designing Autonomous AI》
  • 《Building LLM Powered Applications》

• 论文：

  • 《ReAct: Synergizing Reasoning and Acting in LLMs》
  • 《Chain-of-Thought Prompting》

14.2 社区支持

• GitHub Discussions：问题咨询与功能建议
• Discord频道：实时技术交流
• 定期线上Meetup：案例分享

项目完全开源，欢迎贡献：

• 提交Issue报告问题
• 发起Pull Request贡献代码
• 完善文档或翻译

15. 结语：为什么选择Java构建AI Agent？

经过这个项目的实践，我深刻体会到Java在AI工程化方面的独特优势：

1. 工程严谨性：类型系统能在编译期发现大部分错误
2. 生态成熟度：Spring等框架提供了企业级解决方案
3. 性能可预测：JVM经过多年优化，表现稳定可靠
4. 团队适配性：大多数企业已有Java技术栈积累

当然，Python在原型开发和研究领域仍有不可替代的优势。但当我们真正需要构建可靠、可扩展的生产级AI系统时，Java生态提供了坚实的工程基础。

ZenoAgent项目仍在快速发展中，期待更多Java开发者加入这个充满挑战和乐趣的AI工程化探索之旅。记住：重要的不是框架本身，而是通过亲手实践获得的对AI系统本质的深刻理解。