企业官网建设流程全解析

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的设计，其架构非常清晰：请求先进入Java，Java负责生成可信的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、Rerank；Java侧实现业务逻辑、事务、审计。

这个架构的关键在于Tool Registry的抽象设计。Tool不应该关心它是用Python还是Java实现的，它只需要暴露三样东西：name、description、input_schema。

# tool_registry.pyfromdataclassesimportdataclass,fieldfromtypingimportCallable,Any,Optionalimportinspectimportjson@dataclassclassToolSchema:"""工具的输入/输出Schema定义"""type:str="object"properties:dict=field(default_factory=dict)required:list[str]=field(default_factory=list)@dataclassclassToolMetadata:timeout_seconds:int=30max_retries:int=3requires_auth:bool=False@dataclassclassTool:name:strdescription:strinput_schema:ToolSchema output_schema:Optional[ToolSchema]=Nonemetadata:ToolMetadata=field(default_factory=ToolMetadata)handler:Optional[Callable]=Noneis_remote:bool=False# True表示通过MCP/RPC调用Java服务classToolRegistry:"""工具注册器（单例），统一管理本地与远程工具"""_instance=Nonedef__new__(cls):ifcls._instanceisNone:cls._instance=super().__new__(cls)cls._instance._tools:dict[str,Tool]={}returncls._instancedefregister(self,tool:Tool)->None:iftool.nameinself._tools:raiseValueError(f"Tool '{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:"""执行工具调用，带超时和重试"""tool=self._tools.get(name)ifnottool:raiseValueError(f"Tool '{name}' not found")iftool.is_remote:returnawaitself._invoke_remote(tool,arguments)# 本地工具直接调用ifasyncio.iscoroutinefunction(tool.handler):returnawaitasyncio.wait_for(tool.handler(**arguments),timeout=tool.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.pyfromdataclassesimportdataclassfromtypingimportOptionalimportnumpyasnp@dataclassclassChunk:content:strsource:strscore:float=0.0metadata:dict=NoneclassRAGRetriever:"""轻量级RAG检索器，支持向量检索与关键词回退"""def__init__(self,embedding_fn,vector_store,min_score:float=0.35):self.embed=embedding_fn self.store=vector_store self.min_score=min_score# 低于此阈值的结果被丢弃asyncdefretrieve(self,query:str,top_k:int=5)->list[Chunk]:"""语义检索，带质量过滤"""query_vec=awaitself.embed(query)raw_results=awaitself.store.search(query_vec,top_k=top_k*2)# 过滤弱匹配，避免用噪声填满top_kfiltered=[Chunk(content=r["content"],source=r["source"],score=r["score"])forrinraw_resultsifr["score"]>=self.min_score]returnfiltered[:top_k]iffilteredelseraw_results[:top_k]asyncdefkeyword_search(self,keywords:list[str],search_in:str="both",match_mode:str="any",min_matches:int=2)->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,fieldfromtypingimportOptionalimportopenai@dataclassclassAgentState:"""Agent执行状态（可序列化，支持持久化）"""session_id:strmessages:list=field(default_factory=list)tool_calls_log:list=field(default_factory=list)iteration_count:int=0classReActAgent:"""ReAct范式的Agent引擎，支持RAG与Tool Calling混合"""def__init__(self,llm_client,registry,retriever,max_iterations:int=6):self.llm=llm_client self.registry=registry self.retriever=retriever self.max_iterations=max_iterationsasyncdefrun(self,user_input:str,state:AgentState)->str:state.messages.append({"role":"user","content":user_input})foriterationinrange(self.max_iterations):state.iteration_count=iteration+1# 决策：LLM决定是否需要RAG、是否需要调用工具system_prompt=self._build_system_prompt(state)tools=self.registry.list_tools_for_llm()response=awaitself.llm.chat.completions.create(model="gpt-4o",messages=[{"role":"system","content":system_prompt}]+state.messages,tools=toolsiftoolselseNone,tool_choice="auto")msg=response.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_name=tool_call.function.name arguments=json.loads(tool_call.function.arguments)# 特殊处理：RAG检索作为“虚拟工具”iftool_name=="retrieve_knowledge":result=awaitself._handle_rag_call(arguments)else:try:result=awaitself.registry.invoke(tool_name,arguments)exceptExceptionase:result=f"[工具调用失败:{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:query=arguments.get("query","")chunks=awaitself.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:int=60):self.base_url=base_url self.token=internal_token self.cache_ttl=cache_ttl self._cache:dict[str,tuple[float,any]]={}asyncdefquery_leave_balance(self,employee_id:str)->dict:"""查询年假余额（只读，带缓存）"""cache_key=f"leave_balance:{employee_id}"cached=self._get_cached(cache_key)ifcached:returncachedtry:result=awaitself._call_java_api("/internal/leave/balance",{"employee_id":employee_id})self._set_cache(cache_key,result)returnresultexceptExceptionase:# 降级：返回缓存中的过期数据（如有），或明确的错误信息stale=self._get_cached(cache_key,ignore_ttl=True)ifstale:return{**stale,"_stale":True}raiseRuntimeError(f"业务系统暂时不可用:{e}")asyncdef_call_java_api(self,path:str,params:dict)->dict:asyncwithhttpx.AsyncClient(timeout=5.0)asclient:resp=awaitclient.post(f"{self.base_url}{path}",json=params,headers={"Authorization":f"Bearer{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中注册新的能力。