Python与Java双栈实战手撸一个支持RAG与Tool Calling的高性能Agent框架引言在2026年的AI工程招聘市场上“会调LangChain”已经不够了。企业真正需要的是能够设计跨语言、高可用、可扩展Agent框架的工程师。为什么因为生产环境从来不是单一语言的游戏。Python在AI生态中占据统治地位LangChain、LlamaIndex等框架让RAG与Tool Calling的快速原型开发变得轻而易举。但当系统需要接入企业级Java微服务、处理高并发请求、保障事务一致性时纯Python方案开始力不从心。反过来Java的Spring AI虽然提供了完整的工具调用体系但在RAG生态的灵活性上仍不及Python。双栈架构不是“为了双栈而双栈”而是让每种语言做它最擅长的事Python负责AI编排与检索Java负责业务权威状态与事务保障。本文将手把手带你用Python Java构建一个支持RAG与Tool Calling的Agent框架给出可运行的代码并解析架构设计中的关键决策。一、为什么双栈架构是Agent工程化的必然选择先看一个真实的生产场景用户对Agent说“帮我查一下上个月的差旅报销状态如果超过限额就发起申诉流程”。这个请求涉及三层能力第一层是知识检索“公司差旅报销限额是多少”第二层是业务查询“上个月提交了哪些报销单状态是什么”第三层是受控写操作“如果超限生成申诉Proposal”。如果全部用Python实现你会遇到几个问题Python服务需要直接连接企业数据库这在安全审计上很难通过报销状态查询需要遵循Java服务的事务隔离级别Python重写一套容易引入不一致最关键的是当系统需要水平扩展时Python的异步模型在高并发I/O密集型场景下不如Java的虚拟线程来得稳定。参考一个成熟的开源企业级项目enterprise-ai-copilot的设计其架构非常清晰请求先进入JavaJava负责生成可信的trace_id、employee_id等Runtime Context字段Python服务只在Docker网络内部暴露端口不对外映射。Java调用Python做AI编排Python调用Java的只读业务Tool获取权威状态。这种设计的核心理念是Python拥有的是“思考能力”Java拥有的是“事实权威”。二、架构分层从MCP到Tool Registry的双栈设计在动手写代码之前先明确分层架构。参考AgentCraft项目的五层架构我们简化为四层第一层协议适配层。由Java的Spring Boot负责暴露HTTP API给前端同时提供内部RPC接口供Python调用业务Tool。这一层管理认证、限流、全链路追踪ID。第二层Agent编排层。由Python负责核心是ReAct循环、意图识别、工具路由和RAG检索。这一层不碰任何业务数据库只通过Tool调用获取事实。第三层Tool Registry层。这是双栈架构的“粘合层”。所有工具——无论是Python本地的RAG检索还是通过MCP协议暴露的Java服务——都注册到统一的ToolRegistry中使用标准化的Schema描述输入输出。第四层能力实现层。Python侧实现向量检索、Embedding、RerankJava侧实现业务逻辑、事务、审计。这个架构的关键在于Tool Registry的抽象设计。Tool不应该关心它是用Python还是Java实现的它只需要暴露三样东西name、description、input_schema。# tool_registry.pyfromdataclassesimportdataclass,fieldfromtypingimportCallable,Any,OptionalimportinspectimportjsondataclassclassToolSchema:工具的输入/输出Schema定义type:strobjectproperties:dictfield(default_factorydict)required:list[str]field(default_factorylist)dataclassclassToolMetadata:timeout_seconds:int30max_retries:int3requires_auth:boolFalsedataclassclassTool:name:strdescription:strinput_schema:ToolSchema output_schema:Optional[ToolSchema]Nonemetadata:ToolMetadatafield(default_factoryToolMetadata)handler:Optional[Callable]Noneis_remote:boolFalse# True表示通过MCP/RPC调用Java服务classToolRegistry:工具注册器单例统一管理本地与远程工具_instanceNonedef__new__(cls):ifcls._instanceisNone:cls._instancesuper().__new__(cls)cls._instance._tools:dict[str,Tool]{}returncls._instancedefregister(self,tool:Tool)-None:iftool.nameinself._tools:raiseValueError(fTool {tool.name} already registered)self._tools[tool.name]tooldefget_tool(self,name:str)-Optional[Tool]:returnself._tools.get(name)defget_all_tools(self)-dict[str,Tool]:returnself._tools.copy()deflist_tools_for_llm(self)-list[dict]:转换为OpenAI Function Calling格式tools[]fortoolinself._tools.values():tools.append({type:function,function:{name:tool.name,description:tool.description,parameters:{type:tool.input_schema.type,properties:tool.input_schema.properties,required:tool.input_schema.required}}})returntoolsasyncdefinvoke(self,name:str,arguments:dict)-Any:执行工具调用带超时和重试toolself._tools.get(name)ifnottool:raiseValueError(fTool {name} not found)iftool.is_remote:returnawaitself._invoke_remote(tool,arguments)# 本地工具直接调用ifasyncio.iscoroutinefunction(tool.handler):returnawaitasyncio.wait_for(tool.handler(**arguments),timeouttool.metadata.timeout_seconds)returntool.handler(**arguments)这段Registry代码是双栈架构的基石。is_remote标记区分了本地Python工具和远程Java服务对Agent层完全透明。三、Python侧RAG与Agent编排的核心实现3.1 可插拔的RAG检索器RAG的核心不是“调一次向量库API”而是检索质量的可控性。参考llm-agent-base的设计思路我们需要支持按相似度阈值过滤弱匹配、支持文件名关键词搜索不依赖向量索引、以及增量索引更新。# rag_retriever.pyfromdataclassesimportdataclassfromtypingimportOptionalimportnumpyasnpdataclassclassChunk:content:strsource:strscore:float0.0metadata:dictNoneclassRAGRetriever:轻量级RAG检索器支持向量检索与关键词回退def__init__(self,embedding_fn,vector_store,min_score:float0.35):self.embedembedding_fn self.storevector_store self.min_scoremin_score# 低于此阈值的结果被丢弃asyncdefretrieve(self,query:str,top_k:int5)-list[Chunk]:语义检索带质量过滤query_vecawaitself.embed(query)raw_resultsawaitself.store.search(query_vec,top_ktop_k*2)# 过滤弱匹配避免用噪声填满top_kfiltered[Chunk(contentr[content],sourcer[source],scorer[score])forrinraw_resultsifr[score]self.min_score]returnfiltered[:top_k]iffilteredelseraw_results[:top_k]asyncdefkeyword_search(self,keywords:list[str],search_in:strboth,match_mode:strany,min_matches:int2)-list[Chunk]: 文件名/内容关键词搜索无需向量索引 适用场景结构化文档、代码仓库检索 # 实现省略核心逻辑遍历文档索引匹配文件名或内容passdefformat_rag_context(chunks:list[Chunk])-str:将检索结果格式化为LLM可读的上下文ifnotchunks:return未检索到相关文档。parts[]fori,chunkinenumerate(chunks,1):parts.append(f[文档{i}] 来源:{chunk.source}\n{chunk.content})return\n\n---\n\n.join(parts)min_score过滤是一个容易被忽视但至关重要的设计。很多RAG实现为了“凑够top_k”把低相关度的内容也塞进上下文导致LLM被噪声误导。宁可返回更少的上下文也不要引入幻觉。3.2 ReAct Agent核心循环现在实现Agent的执行引擎。核心是一个“推理-行动-观察”的循环集成RAG检索和Tool Calling。# agent_engine.pyimportasyncioimportjsonfromdataclassesimportdataclass,fieldfromtypingimportOptionalimportopenaidataclassclassAgentState:Agent执行状态可序列化支持持久化session_id:strmessages:listfield(default_factorylist)tool_calls_log:listfield(default_factorylist)iteration_count:int0classReActAgent:ReAct范式的Agent引擎支持RAG与Tool Calling混合def__init__(self,llm_client,registry,retriever,max_iterations:int6):self.llmllm_client self.registryregistry self.retrieverretriever self.max_iterationsmax_iterationsasyncdefrun(self,user_input:str,state:AgentState)-str:state.messages.append({role:user,content:user_input})foriterationinrange(self.max_iterations):state.iteration_countiteration1# 决策LLM决定是否需要RAG、是否需要调用工具system_promptself._build_system_prompt(state)toolsself.registry.list_tools_for_llm()responseawaitself.llm.chat.completions.create(modelgpt-4o,messages[{role:system,content:system_prompt}]state.messages,toolstoolsiftoolselseNone,tool_choiceauto)msgresponse.choices[0].message# 情况一无工具调用直接返回最终答案ifnotmsg.tool_calls:state.messages.append({role:assistant,content:msg.content})returnmsg.content# 情况二执行工具调用state.messages.append(msg)fortool_callinmsg.tool_calls:tool_nametool_call.function.name argumentsjson.loads(tool_call.function.arguments)# 特殊处理RAG检索作为“虚拟工具”iftool_nameretrieve_knowledge:resultawaitself._handle_rag_call(arguments)else:try:resultawaitself.registry.invoke(tool_name,arguments)exceptExceptionase:resultf[工具调用失败:{str(e)}]state.messages.append({role:tool,tool_call_id:tool_call.id,content:str(result)})state.tool_calls_log.append({tool:tool_name,args:arguments,result_preview:str(result)[:200]})return已达到最大迭代次数任务未完成。def_build_system_prompt(self,state:AgentState)-str:return你是一个企业级AI助手。你有以下能力 1. 通过 retrieve_knowledge 工具检索企业知识库 2. 通过其他工具查询业务系统状态 3. 执行受控的业务操作 重要约束 - 对于事实性问题如公司政策、产品信息必须先调用 retrieve_knowledge 检索 - 对于需要操作业务系统的请求使用相应的工具 - 如果工具返回结果不充分可以再次检索或调用其他工具 - 不要编造知识库中不存在的信息asyncdef_handle_rag_call(self,arguments:dict)-str:queryarguments.get(query,)chunksawaitself.retriever.retrieve(query)returnformat_rag_context(chunks)这个Agent循环有几个关键设计RAG被封装为“虚拟工具”与业务工具走同一套调用协议迭代上限防止无限循环状态外置使Agent实例可以无状态部署。四、Java侧Tool Provider与MCP服务暴露Python Agent需要调用Java的业务能力。最优雅的方式是通过MCP协议暴露Java服务让Agent像调用本地工具一样调用远程Java方法。参考Spring AI 2.0的MCP支持Java端只需要一个注解就能将方法暴露为MCP Tool// WeatherTools.java — Java侧的业务工具ComponentpublicclassExpenseTools{privatefinalExpenseServiceexpenseService;McpTool(description查询员工差旅报销单状态)publicExpenseStatusqueryExpenseStatus(McpToolParam(description员工工号)StringemployeeId,McpToolParam(description月份格式YYYY-MM)Stringmonth){returnexpenseService.queryStatus(employeeId,month);}McpTool(description提交差旅报销申诉返回Proposal ID不执行实际写操作)publicProposalResultsubmitAppealProposal(McpToolParam(description报销单ID)StringexpenseId,McpToolParam(description申诉理由)Stringreason){// Proposal阶段无副作用仅生成待确认记录returnexpenseService.createProposal(expenseId,reason);}}MCP Server的自动配置会扫描McpTool注解的方法生成JSON Schema并注册。Python侧只需要通过MCP Client连接就能发现并调用这些工具。但高可用架构需要比MCP走得更远。直接让Agent调用Java MCP Server有两个问题第一Java服务不可用时Agent会阻塞第二缺少Python侧的缓存和降级。参考enterprise-ai-copilot的设计Python侧应该维护一个只读Tool的本地缓存层# java_tool_client.pyimportasynciofromtypingimportOptionalimporthttpxclassJavaToolClient:Python到Java业务Tool的客户端带缓存与降级def__init__(self,base_url:str,internal_token:str,cache_ttl:int60):self.base_urlbase_url self.tokeninternal_token self.cache_ttlcache_ttl self._cache:dict[str,tuple[float,any]]{}asyncdefquery_leave_balance(self,employee_id:str)-dict:查询年假余额只读带缓存cache_keyfleave_balance:{employee_id}cachedself._get_cached(cache_key)ifcached:returncachedtry:resultawaitself._call_java_api(/internal/leave/balance,{employee_id:employee_id})self._set_cache(cache_key,result)returnresultexceptExceptionase:# 降级返回缓存中的过期数据如有或明确的错误信息staleself._get_cached(cache_key,ignore_ttlTrue)ifstale:return{**stale,_stale:True}raiseRuntimeError(f业务系统暂时不可用:{e})asyncdef_call_java_api(self,path:str,params:dict)-dict:asyncwithhttpx.AsyncClient(timeout5.0)asclient:respawaitclient.post(f{self.base_url}{path},jsonparams,headers{Authorization:fBearer{self.token}})resp.raise_for_status()returnresp.json()这个设计实现了优雅降级Java服务短暂不可用时返回缓存中的过期数据并标记_stale让Agent可以决定是继续等待还是告知用户“数据可能不是最新的”。五、双栈协同的关键工程细节Runtime Context的可信传递。在双栈架构中trace_id、employee_id等字段必须由Java在入口处生成并透传不能由LLM的tool_call arguments提供。这既是安全要求防止Prompt注入伪造身份也是审计要求。工具可见性的动态收缩。Planner只有规划权没有执行授权。allow_business_actions为false时Proposal工具应该从工具列表中移除而不是靠LLM“自觉地不调用”。这需要在ToolRegistry层做过滤。Tool Calling的循环边界。Spring AI 2.0将工具执行从ChatModel中移出由ToolCallingAdvisor在外部控制。Python侧的ReAct循环也需要明确的迭代上限和工具调用次数上限防止Agent陷入“检索-发现不足-再检索”的死循环。MCP工具的名称空间隔离。当同时连接多个Java MCP Server时如果它们暴露了同名工具需要前缀机制避免冲突。Python的ToolRegistry应该用{server_name}__{tool_name}作为唯一标识。结语Python与Java的双栈Agent框架本质上是一种关注点分离的工程哲学Python拥有AI编排的灵活性和生态优势Java拥有企业级系统的可靠性、事务性和安全审计能力。从代码层面看核心不是“写两套语言”而是用统一的Tool Registry抽象屏蔽语言边界。Agent层只关心工具的名称、Schema和语义不关心它是Python函数还是远程Java MCP服务。如果你正在设计或重构企业级Agent系统建议从这三个步骤入手先定义ToolRegistry的接口规范这是双栈架构的“宪法”再实现Java侧的MCP Tool暴露让业务能力标准化输出最后在Python侧完成ReAct循环与RAG集成把编排逻辑跑通。这套架构的投入成本不低但它换来的是当业务增长十倍时你不需要重写核心逻辑只需要在ToolRegistry中注册新的能力。