This is an automated email from the ASF dual-hosted git repository. wenjin272 pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit 487bd5844229f9b6712b56fac58a21a8026db62a Author: WenjinXie <[email protected]> AuthorDate: Wed May 20 16:20:36 2026 +0800 [runtime] Own ResourceContext inside ResourceCache for shared lifecycle. --- .../flink_agents/runtime/flink_runner_context.py | 4 ---- python/flink_agents/runtime/local_runner.py | 4 ---- python/flink_agents/runtime/resource_cache.py | 10 ++++----- .../agents/runtime/PythonMCPResourceDiscovery.java | 13 +---------- .../apache/flink/agents/runtime/ResourceCache.java | 26 +++++++++++++--------- .../runtime/operator/PythonBridgeManager.java | 22 ++---------------- .../runtime/resource/ResourceContextImpl.java | 19 ++++++++++------ 7 files changed, 36 insertions(+), 62 deletions(-) diff --git a/python/flink_agents/runtime/flink_runner_context.py b/python/flink_agents/runtime/flink_runner_context.py index 387e327b..ada5b8e3 100644 --- a/python/flink_agents/runtime/flink_runner_context.py +++ b/python/flink_agents/runtime/flink_runner_context.py @@ -52,7 +52,6 @@ from flink_agents.runtime.memory.mem0.mem0_long_term_memory import ( Mem0LongTermMemory, ) from flink_agents.runtime.resource_cache import ResourceCache -from flink_agents.runtime.resource_context import ResourceContextImpl logger = logging.getLogger(__name__) @@ -274,9 +273,6 @@ class FlinkRunnerContext(RunnerContext): self.__agent_plan.resource_providers, self.__agent_plan.config ) self.__resource_cache.set_java_resource_adapter(j_resource_adapter) - self.__resource_cache.set_resource_context( - ResourceContextImpl(self.__resource_cache) - ) self.executor = executor def set_long_term_memory(self, ltm: InternalBaseLongTermMemory) -> None: diff --git a/python/flink_agents/runtime/local_runner.py b/python/flink_agents/runtime/local_runner.py index c575a6ec..078fd7f1 100644 --- a/python/flink_agents/runtime/local_runner.py +++ b/python/flink_agents/runtime/local_runner.py @@ -35,7 +35,6 @@ from flink_agents.plan.configuration import AgentConfiguration from flink_agents.runtime.agent_runner import AgentRunner from flink_agents.runtime.local_memory_object import LocalMemoryObject from flink_agents.runtime.resource_cache import ResourceCache -from flink_agents.runtime.resource_context import ResourceContextImpl if TYPE_CHECKING: from flink_agents.plan.agent_plan import AgentPlan @@ -88,9 +87,6 @@ class LocalRunnerContext(RunnerContext): self.__resource_cache = ResourceCache( agent_plan.resource_providers, agent_plan.config ) - self.__resource_cache.set_resource_context( - ResourceContextImpl(self.__resource_cache) - ) self.__key = key self.events = deque() self._sensory_mem_store = {} diff --git a/python/flink_agents/runtime/resource_cache.py b/python/flink_agents/runtime/resource_cache.py index f66cc758..382a58a9 100644 --- a/python/flink_agents/runtime/resource_cache.py +++ b/python/flink_agents/runtime/resource_cache.py @@ -18,11 +18,11 @@ from typing import Any, Dict from flink_agents.api.resource import Resource, ResourceType -from flink_agents.api.resource_context import ResourceContext from flink_agents.plan.configuration import AgentConfiguration from flink_agents.plan.function import JavaFunction from flink_agents.plan.resource_provider import JavaResourceProvider, ResourceProvider from flink_agents.plan.tools.function_tool import FunctionTool +from flink_agents.runtime.resource_context import ResourceContextImpl class ResourceCache: @@ -54,11 +54,11 @@ class ResourceCache: self._config = config self._cache: Dict[ResourceType, Dict[str, Resource]] = {} self._j_resource_adapter: Any = None - self._resource_context = None + self._resource_context = ResourceContextImpl(self) - def set_resource_context(self, resource_context: ResourceContext) -> None: - """Set the resource context for accessing other resource in runtime.""" - self._resource_context = resource_context + def get_resource_context(self) -> ResourceContextImpl: + """Return the long-lived ResourceContext owned by this cache.""" + return self._resource_context def set_java_resource_adapter(self, j_resource_adapter: Any) -> None: """Set Java resource adapter for Java resource providers.""" diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/PythonMCPResourceDiscovery.java b/runtime/src/main/java/org/apache/flink/agents/runtime/PythonMCPResourceDiscovery.java index 32af3428..356ec67b 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/PythonMCPResourceDiscovery.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/PythonMCPResourceDiscovery.java @@ -25,7 +25,6 @@ import org.apache.flink.agents.plan.resource.python.PythonMCPServer; import org.apache.flink.agents.plan.resource.python.PythonMCPTool; import org.apache.flink.agents.plan.resourceprovider.PythonResourceProvider; import org.apache.flink.agents.plan.resourceprovider.ResourceProvider; -import org.apache.flink.agents.runtime.resource.ResourceContextImpl; import java.util.Map; @@ -71,17 +70,7 @@ public class PythonMCPResourceDiscovery { PythonResourceProvider provider = (PythonResourceProvider) rp; provider.setPythonResourceAdapter(adapter); - PythonMCPServer server = - (PythonMCPServer) - provider.provide( - new ResourceContextImpl( - (name, type) -> { - try { - return cache.getResource(name, type); - } catch (Exception e) { - throw new RuntimeException(e); - } - })); + PythonMCPServer server = (PythonMCPServer) provider.provide(cache.getResourceContext()); for (PythonMCPTool tool : server.listTools()) { cache.put(tool.getName(), TOOL, tool); diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/ResourceCache.java b/runtime/src/main/java/org/apache/flink/agents/runtime/ResourceCache.java index 8e56bb98..2b8c4114 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/ResourceCache.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/ResourceCache.java @@ -46,6 +46,7 @@ public class ResourceCache implements AutoCloseable { private final Map<ResourceType, Map<String, ResourceProvider>> resourceProviders; private final Map<ResourceType, Map<String, Resource>> cache = new ConcurrentHashMap<>(); private volatile PythonResourceAdapter pythonResourceAdapter; + private final ResourceContextImpl resourceContext; public ResourceCache(Map<ResourceType, Map<String, ResourceProvider>> resourceProviders) { // Defensive copy: the cache must not be affected by later mutations to the source map. @@ -54,12 +55,26 @@ public class ResourceCache implements AutoCloseable { resourceProviders.entrySet()) { this.resourceProviders.put(entry.getKey(), new HashMap<>(entry.getValue())); } + + this.resourceContext = + new ResourceContextImpl( + (name, type) -> { + try { + return this.getResource(name, type); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); } void setPythonResourceAdapter(PythonResourceAdapter adapter) { this.pythonResourceAdapter = adapter; } + public ResourceContextImpl getResourceContext() { + return resourceContext; + } + /** * Resolves a resource by name and type, creating it from its provider if not cached. * @@ -87,16 +102,7 @@ public class ResourceCache implements AutoCloseable { ((PythonResourceProvider) provider).setPythonResourceAdapter(pythonResourceAdapter); } - Resource resource = - provider.provide( - new ResourceContextImpl( - (anotherName, anotherType) -> { - try { - return this.getResource(anotherName, anotherType); - } catch (Exception e) { - throw new RuntimeException(e); - } - })); + Resource resource = provider.provide(resourceContext); if (pythonResourceAdapter != null && resource instanceof FunctionTool) { ((FunctionTool) resource).setPythonResourceAdapter(pythonResourceAdapter); diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java index 73eb6a50..cea9cce5 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java @@ -18,7 +18,6 @@ package org.apache.flink.agents.runtime.operator; import org.apache.flink.agents.api.memory.LongTermMemoryOptions; -import org.apache.flink.agents.api.resource.ResourceType; import org.apache.flink.agents.plan.AgentPlan; import org.apache.flink.agents.plan.JavaFunction; import org.apache.flink.agents.plan.PythonFunction; @@ -33,7 +32,6 @@ import org.apache.flink.agents.runtime.python.context.PythonRunnerContextImpl; import org.apache.flink.agents.runtime.python.utils.JavaResourceAdapter; import org.apache.flink.agents.runtime.python.utils.PythonActionExecutor; import org.apache.flink.agents.runtime.python.utils.PythonResourceAdapterImpl; -import org.apache.flink.agents.runtime.resource.ResourceContextImpl; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.JobID; import org.apache.flink.python.env.PythonDependencyInfo; @@ -165,14 +163,7 @@ class PythonBridgeManager implements AutoCloseable { javaResourceAdapter = new JavaResourceAdapter( - new ResourceContextImpl( - (name, type) -> { - try { - return resourceCache.getResource(name, type); - } catch (Exception e) { - throw new RuntimeException(e); - } - }), + resourceCache.getResourceContext(), pythonInterpreter, userCodeClassLoader); if (containPythonResource || mem0Configured) { @@ -262,16 +253,7 @@ class PythonBridgeManager implements AutoCloseable { throws Exception { pythonResourceAdapter = new PythonResourceAdapterImpl( - new ResourceContextImpl( - (String anotherName, ResourceType anotherType) -> { - try { - return resourceCache.getResource(anotherName, anotherType); - } catch (Exception e) { - throw new RuntimeException(e); - } - }), - pythonInterpreter, - javaResourceAdapter); + resourceCache.getResourceContext(), pythonInterpreter, javaResourceAdapter); pythonResourceAdapter.open(); PythonMCPResourceDiscovery.discoverPythonMCPResources( agentPlan.getResourceProviders(), pythonResourceAdapter, resourceCache); diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/resource/ResourceContextImpl.java b/runtime/src/main/java/org/apache/flink/agents/runtime/resource/ResourceContextImpl.java index b7dc0eed..9c74e5ab 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/resource/ResourceContextImpl.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/resource/ResourceContextImpl.java @@ -42,8 +42,8 @@ public class ResourceContextImpl implements ResourceContext { private final BiFunction<String, ResourceType, Resource> getResource; - @Nullable private volatile SkillManager skillManager; - @Nullable private volatile Skills cachedSkillsConfig; + @Nullable private SkillManager skillManager; + private boolean skillManagerInitialized; public ResourceContextImpl(BiFunction<String, ResourceType, Resource> getResource) { this.getResource = getResource; @@ -83,6 +83,15 @@ public class ResourceContextImpl implements ResourceContext { @Nullable private synchronized SkillManager ensureSkillManager() throws Exception { + if (!skillManagerInitialized) { + skillManagerInitialized = true; + skillManager = createSkillManager(); + } + return skillManager; + } + + @Nullable + private SkillManager createSkillManager() throws Exception { Skills config; try { Resource r = getResource(Skills.SKILLS_CONFIG, ResourceType.SKILLS); @@ -94,10 +103,6 @@ public class ResourceContextImpl implements ResourceContext { // No skills config registered — that's fine, return null. return null; } - if (config != cachedSkillsConfig) { - cachedSkillsConfig = config; - skillManager = new SkillManager(config); - } - return skillManager; + return new SkillManager(config); } }
