function canonicalizeForCache(Uri $uri): Uri{
// 从规范化解码的部分开始
$uri = $uri
->withScheme($uri->getScheme())
->withHost($uri->getHost())
->withPath($uri->getPath());
// 为缓存 key 排序 query 参数（这是一种策略选择）
$query = $uri->getQuery();
if ($query) {
parse_str($query, $params);
ksort($params);
$sorted = http_build_query($params, '', '&', PHP_QUERY_RFC3986);
$uri = $uri->withQuery($sorted);
}
return $uri;}$u = new Uri("HTTPS://EXAMPLE.COM/foo/../bar/?b=2&a=1");
use Uri\WhatWg\Url;$url = new Url($input);if (strtolower($url->getAsciiHost() ?? '') !== 'example.com') {
throw new InvalidArgumentException("Bad host");}$new = $url->withPath('/new-path');
use Uri\Rfc3986\Uri;use Uri\UriComparisonMode;$u1 = new Uri("https://example.COM#foo");$u2 = new Uri("https://EXAMPLE.COM");var_dump($u1->equals($u2)); // true
* Base class for agent execution environment.
*
* <p>This class provides the main entry point for integrating Flink Agents with different types of
* Flink data sources (DataStream, Table, or simple lists).
*/public abstract class AgentsExecutionEnvironment { protected final Map<ResourceType, Map<String, Object>> resources; protected AgentsExecutionEnvironment() { this.resources = new HashMap<>(); for (ResourceType type : ResourceType.values()) { this.resources.put(type, new HashMap<>());
}
}
/**
* Get agents execution environment.
*
* <p>Factory method that creates an appropriate execution environment based on the provided
* StreamExecutionEnvironment. If no environment is provided, a local execution environment is
* returned for testing and development.
*
* <p>When integrating with Flink DataStream/Table APIs, users should pass the Flink
* StreamExecutionEnvironment to enable remote execution capabilities.
*
* @param env Optional StreamExecutionEnvironment for remote execution. If null, a local
* execution environment will be created.
* @param tEnv Optional StreamTableEnvironment for table-to-stream conversion.
* @return AgentsExecutionEnvironment appropriate for the execution context.
*/
public static AgentsExecutionEnvironment getExecutionEnvironment(
StreamExecutionEnvironment env, @Nullable StreamTableEnvironment tEnv) { if (env == null) { // Return local execution environment for testing/development
try {
Class<?> localEnvClass =
Class.forName( "org.apache.flink.agents.runtime.env.LocalExecutionEnvironment"); return (AgentsExecutionEnvironment)
localEnvClass.getDeclaredConstructor().newInstance();
} catch (Exception e) { throw new RuntimeException("Failed to create LocalExecutionEnvironment", e);
}
} else { // Return remote execution environment for Flink integration
try {
Class<?> remoteEnvClass =
Class.forName( "org.apache.flink.agents.runtime.env.RemoteExecutionEnvironment"); return (AgentsExecutionEnvironment)
remoteEnvClass
.getDeclaredConstructor(
StreamExecutionEnvironment.class,
StreamTableEnvironment.class)
.newInstance(env, tEnv);
资源配置管理：
内置 resources 映射结构，支持按照资源类型管理各种资源。
提供 addResource() 方法注册可序列化的资源或者资源描述符。
多输入源支持：
fromList()：支持从简单列表创建本地执行环境。
fromDataStream()：集成Flink DataStream API
fromTable()：集成Flink Table API
配置管理：
提供getConfig()抽象方法获取可写的配置对象。
1.3 与原生 Flink Environment 的区别
抽象层级更高：AgentExecutionEnvrionment是对Flink原生环境的封装，在其之上提供了面向Agent的编程模型。主要用于运行Agent，而非 直接处理数据流。
执行模式：AgentExecutionEnvrionment 通过 LocalExecutionEnvironment 和 RemoteExecutionEnvironment 分别支持本地测试和远程集群执行。原生 Flink 主要通过配置参数控制执行模式。
资源管理机制：AgentExecutionEnvrionment 内置了专门的资源注册和管理机制，支持按类型分类管理支援。原生 Flink 没有这种结构化的资源管理方式。
__input: List[Dict[str, Any]] = None
__output: List[Any] = None
__runner: LocalRunner = None
__executed: bool = False
__config: AgentConfiguration = AgentConfiguration()
def set_agent(self, input: list, output: list, runner: LocalRunner) -> None: """Set agent input, output and runner."""
self.__input = input
self.__runner = runner
self.__output = output
def execute(self) -> None: """Execute agent individually."""
if self.__executed:
err_msg = ( "LocalExecutionEnvironment doesn't support execute multiple times."
)
raise RuntimeError(err_msg)
self.__executed = True for input in self.__input:
self.__runner.run(**input)
outputs = self.__runner.get_outputs() for output in outputs:
Returns:
-------
key
The key of the input that was processed.
"""
if "key" in data:
key = data["key"] elif "k" in data:
key = data["k"] else:
key = uuid.uuid4() if key not in self.__keyed_contexts:
self.__keyed_contexts[key] = LocalRunnerContext(self.__agent_plan, key, self.__config)
context = self.__keyed_contexts[key] if "value" in data:
input_event = InputEvent(input=data["value"]) elif "v" in data:
input_event = InputEvent(input=data["v"]) else:
msg = "Input data must be dict has 'v' or 'value' field"
raise RuntimeError(msg)
context.send_event(input_event) while len(context.events) > 0:
event = context.events.popleft() if isinstance(event, OutputEvent):
self.__outputs.append({key: event.output}) continue
event_type hfbgjjc.cn = f"{event.__class__.__module__}.{event.__class__.__name__}"
for action in self.__agent_plan.get_actions(event_type):
context.action_name = action.name
func_result = action.exec(event, context) if isinstance(func_result, Generator): try: for _ in func_result: pass
except Exception:
logger.exception("Error in async execution")
convenient for debugging.
Attributes:
----------
__agent_plan : AgentPlan
Internal agent plan.
__keyed_contexts : dict[Any, LocalRunnerContext]
Dictionary of active contexts indexed by key.
__outputs:
Outputs generated by agent execution.
__config:
Internal configration.
"""
__agent_plan: AgentPlan
__keyed_contexts: Dict[Any, LocalRunnerContext]
__outputs: List[Dict[str, Any]]
__config: AgentConfiguration def __init__(self, agent: Agent, config: AgentConfiguration) -> None: """Initialize the runner with the provided agent.
Parameters
----------
agent : Agent
The agent class to convert and run.
"""
self.__agent_plan fhyjsg.com = AgentPlan.from_agent(agent, config)
self.__keyed_contexts = {}
self.__outputs = []
"""Implementation of RunnerContext for local agent execution.
Attributes:
----------
__agent_plan : AgentPlan
Internal agent plan for this context.
__key : Any
Unique identifier for the context, correspond to the key in flink KeyedStream.
events : deque[Event]
Queue of events to be processed in this context.
action_name: str
Name of the action being executed.
"""
__agent_plan: AgentPlan
__key: Any
events: deque[Event]
action_name: str
_store: dict[str, Any]
_short_term_memory: idcardhome.cn MemoryObject
_config: AgentConfiguration def __init__(self, agent_plan: AgentPlan, key: Any, config: AgentConfiguration) -> None: """Initialize a new context with the given agent and key.
Parameters
----------
agent_plan : AgentPlan
Agent plan used for this context.
key : Any
Unique context identifier, which is corresponding to the key in flink
KeyedStream.
"""
self.__agent_plan = agent_plan
self.__key = key
self.events = deque()
self._store = {}
self._short_term_memory = LocalMemoryObject(
self._store, LocalMemoryObject.ROOT_KEY
)
self._config = config @override
def get_resource(self, name: str, type: ResourceType) -> Resource: return self.__agent_plan.get_resource(name, type)
for action in self.__agent_plan.get_actions(event_type):
context.action_name = action.name
func_result =jrwsh.com action.exec(event, context) # 执行
if isinstance(func_result, Generator): try: for _ in func_result: pass
except Exception:
class LocalRunnerContext(RunnerContext): """Implementation of RunnerContext for local agent execution.
Attributes:
----------
__agent_plan : AgentPlan
Internal agent plan for this context.
__key : Any
Unique identifier yydesk.com for the context, correspond to the key in flink KeyedStream.
events : deque[Event]
Queue of events to be processed in this context.
action_name: str
Name of the action being executed.
"""
__agent_plan: AgentPlan
__key: Any
events: deque[Event]
action_name: str
_store: dict[str, Any]
_short_term_memory: MemoryObject
_config: AgentConfiguration
def __init__(self, agent_plan:eastpv.com AgentPlan, key: Any, config: AgentConfiguration) -> None: """Initialize a new context with the given agent and key.
Parameters
----------
agent_plan : AgentPlan
Agent plan used for this context.
key : Any
Unique context identifier, which is corresponding to the key in flink
KeyedStream.
"""
self.__agent_plan = agent_plan
self.__key = key
self.events = deque()
self._store = {}
self._short_term_memory = LocalMemoryObject(
self._store, LocalMemoryObject.ROOT_KEY
)
self._config = config @property
def key(self) -> Any: """Get the unique identifier for this context.
Returns:
-------
Any
The unique identifier for this context.
"""
return self. njsydq.cn __key @override
def send_event(self, event: Event) -> None: """Send an event to the context's event queue and log it.
Parameters
----------
event : Event
The event to be added to the queue.
"""
logger.info("key: %s, send_event: %s", self.__key, event)
self.events.append(event) @override
def get_resource(self, name:hnsgzxh.com str, type: ResourceType) -> Resource: return self.__agent_plan.get_resource(name, type) @property
@override
def action_config(self) -> Dict[str, Any]: """Get config of the action."""
return self.__agent_plan.get_action_config(action_name=self.action_name) @override
def get_action_config_value(self, key: str) -> Any: """Get config option value of the key."""
return self.__agent_plan.get_action_config_value(
action_name=self.action_name, key=key
) @property
@override
def short_term_memory(self) -> MemoryObject: """Get the short-term memory object associated with this context.
Returns:
-------
MemoryObject
The root object of the short-term memory.
"""
return self._byven.com short_term_memory @property
@override
def agent_metric_group(self) -> MetricGroup:
# TODO: Support metric mechanism for local agent execution.
err_msg = "Metric mechanism is not supported for local agent execution yet."
raise NotImplementedError(err_msg)
@property
@override
def action_metric_group(self) -> MetricGroup:
# TODO: Support metric mechanism for local agent execution.
err_msg = "Metric mechanism is not supported for local agent execution yet."
raise NotImplementedError(err_msg)
def execute_async(
self,
func: Callable[[Any], Any],
*args: Tuple[Any, ...],
**kwargs: Dict[str, Any],
) -> Any: jntmjq.cn """Asynchronously execute the provided function. Access to memory
is prohibited within the function.
"""
logger.warning( "Local runner does not support asynchronous execution; falling back to synchronous execution."
)
func_result = func(*args, **kwargs)
yield return func_result
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/70048957/viewspace-3109179/ ，如需转载，请注明出处，否则将追究法律责任。
