github/autogen
1
1
Fork 0
mirror of https://github.com/microsoft/autogen.git synced 2026-02-11 22:14:58 -05:00
Code Issues Packages Projects Releases Wiki Activity

refactor!: Remove deprecations (#4853)

Browse Source 
* Remove deprecations

* imports
This commit is contained in:
Jack Gerrits
2024-12-30 14:55:21 -05:00
committed by GitHub
parent 5ee2190e00
commit 0b34211c1e
23 changed files with 3 additions and 1542 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
python/packages/autogen-core/src/autogen_core/_agent_runtime.py
View File

@@ -3,8 +3,6 @@ from __future__ import annotations
from collections.abc import Sequence
from typing import Any, Awaitable, Callable, Mapping, Protocol, Type, TypeVar, overload, runtime_checkable
from typing_extensions import deprecated
from ._agent import Agent
from ._agent_id import AgentId
from ._agent_metadata import AgentMetadata
@@ -73,30 +71,6 @@ class AgentRuntime(Protocol):
"""
...
@deprecated(
"Use your agent's `register` method directly instead of this method. See documentation for latest usage."
)
async def register(
self,
type: str,
agent_factory: Callable[[], T | Awaitable[T]],
subscriptions: Callable[[], list[Subscription] | Awaitable[list[Subscription]]]
| list[Subscription]
| None = None,
) -> AgentType:
"""Register an agent factory with the runtime associated with a specific type. The type must be unique.
.. deprecated:: 0.4.0.dev1
Use a specific agent's `register` method directly instead of this method. For example: :meth:`BaseAgent.register`
Args:
type (str): The type of agent this factory creates. It is not the same as agent class name. The `type` parameter is used to differentiate between different factory functions rather than agent classes.
agent_factory (Callable[[], T]): The factory that creates the agent, where T is a concrete Agent type. Inside the factory, use `autogen_core.AgentInstantiationContext` to access variables like the current runtime and agent ID.
subscriptions (Callable[[], list[Subscription]] | list[Subscription] | None, optional): The subscriptions that the agent should be subscribed to. Defaults to None.
"""
...
async def register_factory(
self,
*,

36
python/packages/autogen-core/src/autogen_core/_single_threaded_agent_runtime.py
View File

@@ -13,9 +13,6 @@ from enum import Enum
from typing import Any, Awaitable, Callable, Dict, List, Mapping, ParamSpec, Set, Type, TypeVar, cast
from opentelemetry.trace import TracerProvider
from typing_extensions import deprecated
from autogen_core._serialization import MessageSerializer, SerializationRegistry
from ._agent import Agent
from ._agent_id import AgentId
@@ -27,8 +24,8 @@ from ._cancellation_token import CancellationToken
from ._message_context import MessageContext
from ._message_handler_context import MessageHandlerContext
from ._runtime_impl_helpers import SubscriptionManager, get_impl
from ._serialization import MessageSerializer, SerializationRegistry
from ._subscription import Subscription
from ._subscription_context import SubscriptionInstantiationContext
from ._telemetry import EnvelopeMetadata, MessageRuntimeTracingConfig, TraceHelper, get_telemetry_envelope_metadata
from ._topic import TopicId
from .base.intervention import DropMessage, InterventionHandler
@@ -559,37 +556,6 @@ class SingleThreadedAgentRuntime(AgentRuntime):
async def agent_load_state(self, agent: AgentId, state: Mapping[str, Any]) -> None:
await (await self._get_agent(agent)).load_state(state)
@deprecated(
"Use your agent's `register` method directly instead of this method. See documentation for latest usage."
)
async def register(
self,
type: str,
agent_factory: Callable[[], T | Awaitable[T]] | Callable[[AgentRuntime, AgentId], T | Awaitable[T]],
subscriptions: Callable[[], list[Subscription] | Awaitable[list[Subscription]]]
| list[Subscription]
| None = None,
) -> AgentType:
if type in self._agent_factories:
raise ValueError(f"Agent with type {type} already exists.")
if subscriptions is not None:
if callable(subscriptions):
with SubscriptionInstantiationContext.populate_context(AgentType(type)):
subscriptions_list_result = subscriptions()
if inspect.isawaitable(subscriptions_list_result):
subscriptions_list = await subscriptions_list_result
else:
subscriptions_list = subscriptions_list_result
else:
subscriptions_list = subscriptions
for subscription in subscriptions_list:
await self.add_subscription(subscription)
self._agent_factories[type] = agent_factory
return AgentType(type)
async def register_factory(
self,
*,

2
python/packages/autogen-core/src/autogen_core/_subscription_context.py
View File

@@ -2,7 +2,7 @@ from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any, ClassVar, Generator
from autogen_core._agent_type import AgentType
from ._agent_type import AgentType
class SubscriptionInstantiationContext:

0
python/packages/autogen-core/src/autogen_core/application/__init__.py
View File

10
python/packages/autogen-core/src/autogen_core/application/_single_threaded_agent_runtime.py
View File

@@ -1,10 +0,0 @@
from typing_extensions import deprecated
from .._single_threaded_agent_runtime import SingleThreadedAgentRuntime as SingleThreadedAgentRuntimeAlias
@deprecated(
"autogen_core.application.SingleThreadedAgentRuntime moved to autogen_core.SingleThreadedAgentRuntime. This alias will be removed in 0.4.0."
)
class SingleThreadedAgentRuntime(SingleThreadedAgentRuntimeAlias):
pass

9
python/packages/autogen-core/src/autogen_core/application/logging/__init__.py
View File

@@ -1,9 +0,0 @@
ROOT_LOGGER_NAME = "autogen_core"
"""Deprecated alias. Use autogen_core.ROOT_LOGGER_NAME"""
EVENT_LOGGER_NAME = "autogen_core.events"
"""Deprecated alias. Use autogen_core.EVENT_LOGGER_NAME"""
TRACE_LOGGER_NAME = "autogen_core.trace"
"""Deprecated alias. Use autogen_core.TRACE_LOGGER_NAME"""

138
python/packages/autogen-core/src/autogen_core/base/__init__.py
View File

@@ -1,138 +0,0 @@
"""
The :mod:`autogen_core.base` module provides the foundational generic interfaces upon which all else is built. This module must not depend on any other module.
"""
from typing import Any, TypeVar
from typing_extensions import deprecated
from .._agent import Agent as AgentAlias
from .._agent_id import AgentId as AgentIdAlias
from .._agent_instantiation import AgentInstantiationContext as AgentInstantiationContextAlias
from .._agent_metadata import AgentMetadata as AgentMetadataAlias
from .._agent_proxy import AgentProxy as AgentProxyAlias
from .._agent_runtime import AgentRuntime as AgentRuntimeAlias
from .._agent_type import AgentType as AgentTypeAlias
from .._base_agent import BaseAgent as BaseAgentAlias
from .._cancellation_token import CancellationToken as CancellationTokenAlias
from .._message_context import MessageContext as MessageContextAlias
from .._message_handler_context import MessageHandlerContext as MessageHandlerContextAlias
from .._serialization import (
MessageSerializer as MessageSerializerAlias,
)
from .._serialization import (
UnknownPayload as UnknownPayloadAlias,
)
from .._serialization import (
try_get_known_serializers_for_type as try_get_known_serializers_for_type_alias,
)
from .._subscription import Subscription as SubscriptionAlias
from .._subscription_context import SubscriptionInstantiationContext as SubscriptionInstantiationContextAlias
from .._topic import TopicId as TopicIdAlias
@deprecated("autogen_core.base.Agent moved to autogen_core.Agent. This alias will be removed in 0.4.0.")
class Agent(AgentAlias):
pass
@deprecated("autogen_core.base.AgentId moved to autogen_core.AgentId. This alias will be removed in 0.4.0.")
class AgentId(AgentIdAlias):
pass
@deprecated(
"autogen_core.base.AgentInstantiationContext moved to autogen_core.AgentInstantiationContext. This alias will be removed in 0.4.0."
)
class AgentInstantiationContext(AgentInstantiationContextAlias):
pass
@deprecated("autogen_core.base.AgentMetadata moved to autogen_core.AgentMetadata. This alias will be removed in 0.4.0.")
class AgentMetadata(AgentMetadataAlias):
pass
@deprecated("autogen_core.base.AgentProxy moved to autogen_core.AgentProxy. This alias will be removed in 0.4.0.")
class AgentProxy(AgentProxyAlias):
pass
@deprecated("autogen_core.base.AgentRuntime moved to autogen_core.AgentRuntime. This alias will be removed in 0.4.0.")
class AgentRuntime(AgentRuntimeAlias):
pass
@deprecated("autogen_core.base.AgentType moved to autogen_core.AgentType. This alias will be removed in 0.4.0.")
class AgentType(AgentTypeAlias):
pass
@deprecated("autogen_core.base.BaseAgent moved to autogen_core.BaseAgent. This alias will be removed in 0.4.0.")
class BaseAgent(BaseAgentAlias):
pass
@deprecated(
"autogen_core.base.CancellationToken moved to autogen_core.CancellationToken. This alias will be removed in 0.4.0."
)
class CancellationToken(CancellationTokenAlias):
pass
@deprecated(
"autogen_core.base.MessageContext moved to autogen_core.MessageContext. This alias will be removed in 0.4.0."
)
class MessageContext(MessageContextAlias):
pass
@deprecated(
"autogen_core.base.MessageHandlerContext moved to autogen_core.MessageHandlerContext. This alias will be removed in 0.4.0."
)
class MessageHandlerContext(MessageHandlerContextAlias):
pass
@deprecated(
"autogen_core.base.UnknownPayloadAlias moved to autogen_core.UnknownPayloadAlias. This alias will be removed in 0.4.0."
)
class UnknownPayload(UnknownPayloadAlias):
pass
T = TypeVar("T")
@deprecated(
"autogen_core.base.MessageSerializer moved to autogen_core.MessageSerializer. This alias will be removed in 0.4.0."
)
class MessageSerializer(MessageSerializerAlias[T]):
pass
@deprecated("autogen_core.base.Subscription moved to autogen_core.Subscription. This alias will be removed in 0.4.0.")
class Subscription(SubscriptionAlias):
pass
@deprecated(
"autogen_core.base.try_get_known_serializers_for_type moved to autogen_core.try_get_known_serializers_for_type. This alias will be removed in 0.4.0."
)
def try_get_known_serializers_for_type(cls: type[Any]) -> list[MessageSerializerAlias[Any]]:
return try_get_known_serializers_for_type_alias(cls)
@deprecated(
"autogen_core.base.SubscriptionInstantiationContext moved to autogen_core.SubscriptionInstantiationContext. This alias will be removed in 0.4.0."
)
class SubscriptionInstantiationContext(SubscriptionInstantiationContextAlias):
pass
@deprecated("autogen_core.base.TopicId moved to autogen_core.TopicId. This alias will be removed in 0.4.0.")
class TopicId(TopicIdAlias):
pass
__all__ = [] # type: ignore

37
python/packages/autogen-core/src/autogen_core/base/exceptions.py
View File

@@ -1,37 +0,0 @@
from typing_extensions import deprecated
from ..exceptions import (
CantHandleException as CantHandleExceptionAlias,
)
from ..exceptions import (
MessageDroppedException as MessageDroppedExceptionAlias,
)
from ..exceptions import (
NotAccessibleError as NotAccessibleErrorAlias,
)
from ..exceptions import (
UndeliverableException as UndeliverableExceptionAlias,
)
@deprecated("Moved to autogen_core.exceptions.CantHandleException. Alias will be removed in 0.4.0")
class CantHandleException(CantHandleExceptionAlias):
"""Raised when a handler can't handle the exception."""
@deprecated("Moved to autogen_core.exceptions.UndeliverableException. Alias will be removed in 0.4.0")
class UndeliverableException(UndeliverableExceptionAlias):
"""Raised when a message can't be delivered."""
@deprecated("Moved to autogen_core.exceptions.MessageDroppedException. Alias will be removed in 0.4.0")
class MessageDroppedException(MessageDroppedExceptionAlias):
"""Raised when a message is dropped."""
@deprecated("Moved to autogen_core.exceptions.NotAccessibleError. Alias will be removed in 0.4.0")
class NotAccessibleError(NotAccessibleErrorAlias):
"""Tried to access a value that is not accessible. For example if it is remote cannot be accessed locally."""
__all__ = [] # type: ignore

115
python/packages/autogen-core/src/autogen_core/components/__init__.py
View File

@@ -1,115 +0,0 @@
"""
The :mod:`autogen_core.components` module provides building blocks for creating single agents
"""
from typing import Any, Callable, Type, TypeVar
from typing_extensions import deprecated
from .._base_agent import BaseAgent
from .._closure_agent import ClosureAgent as ClosureAgentAlias
from .._closure_agent import ClosureContext as ClosureContextAlias
from .._default_subscription import (
DefaultSubscription as DefaultSubscriptionAlias,
)
from .._default_subscription import (
default_subscription as default_subscription_alias,
)
from .._default_subscription import (
type_subscription as type_subscription_alias,
)
from .._default_topic import DefaultTopicId as DefaultTopicIdAlias
from .._image import Image as ImageAlias
from .._routed_agent import (
RoutedAgent as RoutedAgentAlias,
)
from .._routed_agent import (
event as event_alias,
)
from .._routed_agent import (
message_handler as message_handler_alias,
)
from .._routed_agent import (
rpc as rpc_aliass,
)
from .._type_prefix_subscription import TypePrefixSubscription as TypePrefixSubscriptionAlias
from .._type_subscription import TypeSubscription as TypeSubscriptionAlias
from .._types import FunctionCall as FunctionCallAlias
__all__ = [] # type: ignore
@deprecated("Moved to autogen_core.TypePrefixSubscription. Will be removed in 0.4.0")
class TypePrefixSubscription(TypePrefixSubscriptionAlias):
pass
@deprecated("Moved to autogen_core.TypeSubscription. Will be removed in 0.4.0")
class TypeSubscription(TypeSubscriptionAlias):
pass
@deprecated("Moved to autogen_core.ClosureAgent. Will be removed in 0.4.0")
class ClosureAgent(ClosureAgentAlias):
pass
@deprecated("Moved to autogen_core.ClosureContext. Will be removed in 0.4.0")
class ClosureContext(ClosureContextAlias):
pass
@deprecated("Moved to autogen_core.DefaultSubscription. Will be removed in 0.4.0")
class DefaultSubscription(DefaultSubscriptionAlias):
pass
BaseAgentType = TypeVar("BaseAgentType", bound="BaseAgent")
@deprecated("Moved to autogen_core.default_subscription. Will be removed in 0.4.0")
def default_subscription(
cls: Type[BaseAgentType] | None = None,
) -> Callable[[Type[BaseAgentType]], Type[BaseAgentType]] | Type[BaseAgentType]:
return default_subscription_alias(cls) # type: ignore
@deprecated("Moved to autogen_core.type_subscription. Will be removed in 0.4.0")
def type_subscription(topic_type: str) -> Callable[[Type[BaseAgentType]], Type[BaseAgentType]]:
return type_subscription_alias(topic_type)
@deprecated("Moved to autogen_core.DefaultTopicId. Will be removed in 0.4.0")
class DefaultTopicId(DefaultTopicIdAlias):
pass
@deprecated("Moved to autogen_core.Image. Will be removed in 0.4.0")
class Image(ImageAlias):
pass
@deprecated("Moved to autogen_core.RoutedAgent. Will be removed in 0.4.0")
class RoutedAgent(RoutedAgentAlias):
pass
# Generic forwarding of all args to the alias
@deprecated("Moved to autogen_core.event. Will be removed in 0.4.0")
def event(*args: Any, **kwargs: Any) -> Any:
return event_alias(*args, **kwargs) # type: ignore
@deprecated("Moved to autogen_core.message_handler. Will be removed in 0.4.0")
def message_handler(*args: Any, **kwargs: Any) -> Any:
return message_handler_alias(*args, **kwargs) # type: ignore
@deprecated("Moved to autogen_core.rpc. Will be removed in 0.4.0")
def rpc(*args: Any, **kwargs: Any) -> Any:
return rpc_aliass(*args, **kwargs) # type: ignore
@deprecated("Moved to autogen_core.FunctionCall. Will be removed in 0.4.0")
class FunctionCall(FunctionCallAlias):
pass

25
python/packages/autogen-core/src/autogen_core/components/code_executor/__init__.py
View File

@@ -1,25 +0,0 @@
from ._base import CodeBlock, CodeExecutor, CodeResult # type: ignore
from ._func_with_reqs import (
Alias, # type: ignore
FunctionWithRequirements, # type: ignore
FunctionWithRequirementsStr, # type: ignore
Import,
ImportFromModule, # type: ignore
with_requirements, # type: ignore
)
from ._impl.command_line_code_result import CommandLineCodeResult # type: ignore
from ._impl.local_commandline_code_executor import LocalCommandLineCodeExecutor # type: ignore
__all__ = [
"LocalCommandLineCodeExecutor",
"CommandLineCodeResult",
"CodeBlock",
"CodeExecutor",
"CodeResult",
"Alias",
"ImportFromModule",
"Import",
"FunctionWithRequirements",
"FunctionWithRequirementsStr",
"with_requirements",
]

66
python/packages/autogen-core/src/autogen_core/components/code_executor/_base.py
View File

@@ -1,66 +0,0 @@
# File based from: https://github.com/microsoft/autogen/blob/main/autogen/coding/base.py
# Credit to original authors
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Protocol, runtime_checkable
from typing_extensions import deprecated
from ... import CancellationToken
@deprecated("Moved to autogen_core.code_executor.CodeBlock. This alias will be removed in 0.4.0.")
@dataclass
class CodeBlock:
"""A code block extracted fromm an agent message."""
code: str
language: str
@deprecated("Moved to autogen_core.code_executor.CodeResult. This alias will be removed in 0.4.0.")
@dataclass
class CodeResult:
"""Result of a code execution."""
exit_code: int
output: str
@deprecated("Moved to autogen_core.code_executor.CodeExecutor. This alias will be removed in 0.4.0.")
@runtime_checkable
class CodeExecutor(Protocol):
"""Executes code blocks and returns the result."""
async def execute_code_blocks(
self,
code_blocks: List[CodeBlock], # type: ignore
cancellation_token: CancellationToken, # type: ignore
) -> CodeResult: # type: ignore
"""Execute code blocks and return the result.
This method should be implemented by the code executor.
Args:
code_blocks (List[CodeBlock]): The code blocks to execute.
Returns:
CodeResult: The result of the code execution.
Raises:
ValueError: Errors in user inputs
asyncio.TimeoutError: Code execution timeouts
asyncio.CancelledError: CancellationToken evoked during execution
"""
...
async def restart(self) -> None:
"""Restart the code executor.
This method should be implemented by the code executor.
This method is called when the agent is reset.
"""
...

206
python/packages/autogen-core/src/autogen_core/components/code_executor/_func_with_reqs.py
View File

@@ -1,206 +0,0 @@
# File based from: https://github.com/microsoft/autogen/blob/main/autogen/coding/func_with_reqs.py
# Credit to original authors
from __future__ import annotations
import functools
import inspect
from dataclasses import dataclass, field
from importlib.abc import SourceLoader
from importlib.util import module_from_spec, spec_from_loader
from textwrap import dedent, indent
from typing import Any, Callable, Generic, List, Sequence, Set, TypeVar, Union
from typing_extensions import ParamSpec, deprecated
T = TypeVar("T")
P = ParamSpec("P")
def _to_code(func: Union[FunctionWithRequirements[T, P], Callable[P, T], FunctionWithRequirementsStr]) -> str: # type: ignore
if isinstance(func, FunctionWithRequirementsStr): # type: ignore
return func.func
code = inspect.getsource(func)
# Strip the decorator
if code.startswith("@"):
code = code[code.index("\n") + 1 :]
return code
@deprecated("Moved to autogen_core.code_executor.Alias. This alias will be removed in 0.4.0.")
@dataclass
class Alias:
name: str
alias: str
@deprecated("Moved to autogen_core.code_executor.ImportFromModule. This alias will be removed in 0.4.0.")
@dataclass
class ImportFromModule:
module: str
imports: List[Union[str, Alias]] # type: ignore
Import = Union[str, ImportFromModule, Alias] # type: ignore
def _import_to_str(im: Import) -> str:
if isinstance(im, str):
return f"import {im}"
elif isinstance(im, Alias): # type: ignore
return f"import {im.name} as {im.alias}"
else:
def to_str(i: Union[str, Alias]) -> str: # type: ignore
if isinstance(i, str):
return i
else:
return f"{i.name} as {i.alias}"
imports = ", ".join(map(to_str, im.imports))
return f"from {im.module} import {imports}"
class _StringLoader(SourceLoader):
def __init__(self, data: str):
self.data = data
def get_source(self, fullname: str) -> str:
return self.data
def get_data(self, path: str) -> bytes:
return self.data.encode("utf-8")
def get_filename(self, fullname: str) -> str:
return "<not a real path>/" + fullname + ".py"
@deprecated("Moved to autogen_core.code_executor.CodeBlock. This alias will be removed in 0.4.0.")
@dataclass
class FunctionWithRequirementsStr:
func: str
compiled_func: Callable[..., Any]
_func_name: str
python_packages: Sequence[str] = field(default_factory=list)
global_imports: Sequence[Import] = field(default_factory=list)
def __init__(self, func: str, python_packages: Sequence[str] = [], global_imports: Sequence[Import] = []):
self.func = func
self.python_packages = python_packages
self.global_imports = global_imports
module_name = "func_module"
loader = _StringLoader(func)
spec = spec_from_loader(module_name, loader)
if spec is None:
raise ValueError("Could not create spec")
module = module_from_spec(spec)
if spec.loader is None:
raise ValueError("Could not create loader")
try:
spec.loader.exec_module(module)
except Exception as e:
raise ValueError(f"Could not compile function: {e}") from e
functions = inspect.getmembers(module, inspect.isfunction)
if len(functions) != 1:
raise ValueError("The string must contain exactly one function")
self._func_name, self.compiled_func = functions[0]
def __call__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError("String based function with requirement objects are not directly callable")
@deprecated("Moved to autogen_core.code_executor.FunctionWithRequirements. This alias will be removed in 0.4.0.")
@dataclass
class FunctionWithRequirements(Generic[T, P]):
func: Callable[P, T]
python_packages: Sequence[str] = field(default_factory=list)
global_imports: Sequence[Import] = field(default_factory=list)
@classmethod
def from_callable(
cls, func: Callable[P, T], python_packages: Sequence[str] = [], global_imports: Sequence[Import] = []
) -> FunctionWithRequirements[T, P]: # type: ignore
return cls(python_packages=python_packages, global_imports=global_imports, func=func)
@staticmethod
def from_str(
func: str, python_packages: Sequence[str] = [], global_imports: Sequence[Import] = []
) -> FunctionWithRequirementsStr: # type: ignore
return FunctionWithRequirementsStr(func=func, python_packages=python_packages, global_imports=global_imports) # type: ignore
# Type this based on F
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T:
return self.func(*args, **kwargs)
@deprecated("Moved to autogen_core.code_executor.with_requirements. This alias will be removed in 0.4.0.")
def with_requirements(
python_packages: Sequence[str] = [], global_imports: Sequence[Import] = []
) -> Callable[[Callable[P, T]], FunctionWithRequirements[T, P]]: # type: ignore
"""Decorate a function with package and import requirements
Args:
python_packages (List[str], optional): Packages required to function. Can include version info.. Defaults to [].
global_imports (List[Import], optional): Required imports. Defaults to [].
Returns:
Callable[[Callable[P, T]], FunctionWithRequirements[T, P]]: The decorated function
"""
def wrapper(func: Callable[P, T]) -> FunctionWithRequirements[T, P]: # type: ignore
func_with_reqs = FunctionWithRequirements( # type: ignore
python_packages=python_packages, global_imports=global_imports, func=func
)
functools.update_wrapper(func_with_reqs, func)
return func_with_reqs
return wrapper
def build_python_functions_file(
funcs: Sequence[Union[FunctionWithRequirements[Any, P], Callable[..., Any], FunctionWithRequirementsStr]], # type: ignore
) -> str:
""":meta private:"""
# First collect all global imports
global_imports: Set[Import] = set()
for func in funcs:
if isinstance(func, (FunctionWithRequirements, FunctionWithRequirementsStr)): # type: ignore
global_imports.update(func.global_imports)
content = "\n".join(map(_import_to_str, global_imports)) + "\n\n"
for func in funcs:
content += _to_code(func) + "\n\n"
return content
def to_stub(func: Union[Callable[..., Any], FunctionWithRequirementsStr]) -> str: # type: ignore
"""Generate a stub for a function as a string
Args:
func (Callable[..., Any]): The function to generate a stub for
Returns:
str: The stub for the function
"""
if isinstance(func, FunctionWithRequirementsStr): # type: ignore
return to_stub(func.compiled_func)
content = f"def {func.__name__}{inspect.signature(func)}:\n"
docstring = func.__doc__
if docstring:
docstring = dedent(docstring)
docstring = '"""' + docstring + '"""'
docstring = indent(docstring, " ")
content += docstring + "\n"
content += " ..."
return content

0
python/packages/autogen-core/src/autogen_core/components/code_executor/_impl/__init__.py
View File

16
python/packages/autogen-core/src/autogen_core/components/code_executor/_impl/command_line_code_result.py
View File

@@ -1,16 +0,0 @@
from dataclasses import dataclass
from typing import Optional
from typing_extensions import deprecated
from ....code_executor._base import CodeResult
@deprecated(
"CommandLineCodeResult moved to autogen_ext.code_executors.CommandLineCodeResult. This alias will be removed in 0.4.0."
)
@dataclass
class CommandLineCodeResult(CodeResult):
"""A code result class for command line code executor."""
code_file: Optional[str]

361
python/packages/autogen-core/src/autogen_core/components/code_executor/_impl/local_commandline_code_executor.py
View File

@@ -1,361 +0,0 @@
# File based from: https://github.com/microsoft/autogen/blob/main/autogen/coding/local_commandline_code_executor.py
# Credit to original authors
import asyncio
import logging
import os
import sys
import warnings
from hashlib import sha256
from pathlib import Path
from string import Template
from types import SimpleNamespace
from typing import Any, Callable, ClassVar, List, Optional, Sequence, Union
from typing_extensions import ParamSpec, deprecated
from .... import CancellationToken
from ....code_executor._base import CodeBlock, CodeExecutor
from ....code_executor._func_with_reqs import (
FunctionWithRequirements,
FunctionWithRequirementsStr,
build_python_functions_file,
to_stub,
)
from .command_line_code_result import CommandLineCodeResult # type: ignore
from .utils import PYTHON_VARIANTS, get_file_name_from_content, lang_to_cmd, silence_pip # type: ignore
__all__ = ("LocalCommandLineCodeExecutor",)
A = ParamSpec("A")
@deprecated(
"LocalCommandLineCodeExecutor moved to autogen_ext.code_executors.local.LocalCommandLineCodeExecutor. This alias will be removed in 0.4.0."
)
class LocalCommandLineCodeExecutor(CodeExecutor):
"""A code executor class that executes code through a local command line
environment.
.. danger::
This will execute code on the local machine. If being used with LLM generated code, caution should be used.
Each code block is saved as a file and executed in a separate process in
the working directory, and a unique file is generated and saved in the
working directory for each code block.
The code blocks are executed in the order they are received.
Command line code is sanitized using regular expression match against a list of dangerous commands in order to prevent self-destructive
commands from being executed which may potentially affect the users environment.
Currently the only supported languages is Python and shell scripts.
For Python code, use the language "python" for the code block.
For shell scripts, use the language "bash", "shell", or "sh" for the code
block.
Args:
timeout (int): The timeout for the execution of any single code block. Default is 60.
work_dir (str): The working directory for the code execution. If None,
a default working directory will be used. The default working
directory is the current directory ".".
functions (List[Union[FunctionWithRequirements[Any, A], Callable[..., Any]]]): A list of functions that are available to the code executor. Default is an empty list.
functions_module (str, optional): The name of the module that will be created to store the functions. Defaults to "functions".
virtual_env_context (Optional[SimpleNamespace], optional): The virtual environment context. Defaults to None.
Example:
How to use `LocalCommandLineCodeExecutor` with a virtual environment different from the one used to run the autogen application:
Set up a virtual environment using the `venv` module, and pass its context to the initializer of `LocalCommandLineCodeExecutor`. This way, the executor will run code within the new environment.
.. code-block:: python
import venv
from pathlib import Path
import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
async def example():
work_dir = Path("coding")
work_dir.mkdir(exist_ok=True)
venv_dir = work_dir / ".venv"
venv_builder = venv.EnvBuilder(with_pip=True)
venv_builder.create(venv_dir)
venv_context = venv_builder.ensure_directories(venv_dir)
local_executor = LocalCommandLineCodeExecutor(work_dir=work_dir, virtual_env_context=venv_context)
await local_executor.execute_code_blocks(
code_blocks=[
CodeBlock(language="bash", code="pip install matplotlib"),
],
cancellation_token=CancellationToken(),
)
asyncio.run(example())
"""
SUPPORTED_LANGUAGES: ClassVar[List[str]] = [
"bash",
"shell",
"sh",
"pwsh",
"powershell",
"ps1",
"python",
]
FUNCTION_PROMPT_TEMPLATE: ClassVar[
str
] = """You have access to the following user defined functions. They can be accessed from the module called `$module_name` by their function names.
For example, if there was a function called `foo` you could import it by writing `from $module_name import foo`
$functions"""
def __init__(
self,
timeout: int = 60,
work_dir: Union[Path, str] = Path("."),
functions: Sequence[
Union[
FunctionWithRequirements[Any, A],
Callable[..., Any],
FunctionWithRequirementsStr,
]
] = [],
functions_module: str = "functions",
virtual_env_context: Optional[SimpleNamespace] = None,
):
if timeout < 1:
raise ValueError("Timeout must be greater than or equal to 1.")
if isinstance(work_dir, str):
work_dir = Path(work_dir)
if not functions_module.isidentifier():
raise ValueError("Module name must be a valid Python identifier")
self._functions_module = functions_module
work_dir.mkdir(exist_ok=True)
self._timeout = timeout
self._work_dir: Path = work_dir
self._functions = functions
# Setup could take some time so we intentionally wait for the first code block to do it.
if len(functions) > 0:
self._setup_functions_complete = False
else:
self._setup_functions_complete = True
self._virtual_env_context: Optional[SimpleNamespace] = virtual_env_context
def format_functions_for_prompt(self, prompt_template: str = FUNCTION_PROMPT_TEMPLATE) -> str:
"""(Experimental) Format the functions for a prompt.
The template includes two variables:
- `$module_name`: The module name.
- `$functions`: The functions formatted as stubs with two newlines between each function.
Args:
prompt_template (str): The prompt template. Default is the class default.
Returns:
str: The formatted prompt.
"""
template = Template(prompt_template)
return template.substitute(
module_name=self._functions_module,
functions="\n\n".join([to_stub(func) for func in self._functions]),
)
@property
def functions_module(self) -> str:
"""(Experimental) The module name for the functions."""
return self._functions_module
@property
def functions(self) -> List[str]:
raise NotImplementedError
@property
def timeout(self) -> int:
"""(Experimental) The timeout for code execution."""
return self._timeout
@property
def work_dir(self) -> Path:
"""(Experimental) The working directory for the code execution."""
return self._work_dir
async def _setup_functions(self, cancellation_token: CancellationToken) -> None:
func_file_content = build_python_functions_file(self._functions)
func_file = self._work_dir / f"{self._functions_module}.py"
func_file.write_text(func_file_content)
# Collect requirements
lists_of_packages = [x.python_packages for x in self._functions if isinstance(x, FunctionWithRequirements)]
flattened_packages = [item for sublist in lists_of_packages for item in sublist]
required_packages = list(set(flattened_packages))
if len(required_packages) > 0:
logging.info("Ensuring packages are installed in executor.")
cmd_args = ["-m", "pip", "install"]
cmd_args.extend(required_packages)
if self._virtual_env_context:
py_executable = self._virtual_env_context.env_exe
else:
py_executable = sys.executable
task = asyncio.create_task(
asyncio.create_subprocess_exec(
py_executable,
*cmd_args,
cwd=self._work_dir,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
)
cancellation_token.link_future(task)
try:
proc = await task
stdout, stderr = await asyncio.wait_for(proc.communicate(), self._timeout)
except asyncio.TimeoutError as e:
raise ValueError("Pip install timed out") from e
except asyncio.CancelledError as e:
raise ValueError("Pip install was cancelled") from e
if proc.returncode is not None and proc.returncode != 0:
raise ValueError(f"Pip install failed. {stdout.decode()}, {stderr.decode()}")
# Attempt to load the function file to check for syntax errors, imports etc.
exec_result = await self._execute_code_dont_check_setup(
[CodeBlock(code=func_file_content, language="python")], cancellation_token
)
if exec_result.exit_code != 0:
raise ValueError(f"Functions failed to load: {exec_result.output}")
self._setup_functions_complete = True
async def execute_code_blocks(
self, code_blocks: List[CodeBlock], cancellation_token: CancellationToken
) -> CommandLineCodeResult: # type: ignore
"""(Experimental) Execute the code blocks and return the result.
Args:
code_blocks (List[CodeBlock]): The code blocks to execute.
cancellation_token (CancellationToken): a token to cancel the operation
Returns:
CommandLineCodeResult: The result of the code execution."""
if not self._setup_functions_complete:
await self._setup_functions(cancellation_token)
return await self._execute_code_dont_check_setup(code_blocks, cancellation_token)
async def _execute_code_dont_check_setup(
self, code_blocks: List[CodeBlock], cancellation_token: CancellationToken
) -> CommandLineCodeResult: # type: ignore
logs_all: str = ""
file_names: List[Path] = []
exitcode = 0
for code_block in code_blocks:
lang, code = code_block.language, code_block.code
lang = lang.lower()
code = silence_pip(code, lang)
if lang in PYTHON_VARIANTS:
lang = "python"
if lang not in self.SUPPORTED_LANGUAGES:
# In case the language is not supported, we return an error message.
exitcode = 1
logs_all += "\n" + f"unknown language {lang}"
break
try:
# Check if there is a filename comment
filename = get_file_name_from_content(code, self._work_dir)
except ValueError:
return CommandLineCodeResult( # type: ignore
exit_code=1,
output="Filename is not in the workspace",
code_file=None,
)
if filename is None:
# create a file with an automatically generated name
code_hash = sha256(code.encode()).hexdigest()
filename = f"tmp_code_{code_hash}.{'py' if lang.startswith('python') else lang}"
written_file = (self._work_dir / filename).resolve()
with written_file.open("w", encoding="utf-8") as f:
f.write(code)
file_names.append(written_file)
env = os.environ.copy()
if self._virtual_env_context:
virtual_env_exe_abs_path = os.path.abspath(self._virtual_env_context.env_exe)
virtual_env_bin_abs_path = os.path.abspath(self._virtual_env_context.bin_path)
env["PATH"] = f"{virtual_env_bin_abs_path}{os.pathsep}{env['PATH']}"
program = virtual_env_exe_abs_path if lang.startswith("python") else lang_to_cmd(lang)
else:
program = sys.executable if lang.startswith("python") else lang_to_cmd(lang)
# Wrap in a task to make it cancellable
task = asyncio.create_task(
asyncio.create_subprocess_exec(
program,
str(written_file.absolute()),
cwd=self._work_dir,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=env,
)
)
cancellation_token.link_future(task)
try:
proc = await task
stdout, stderr = await asyncio.wait_for(proc.communicate(), self._timeout)
exitcode = proc.returncode or 0
except asyncio.TimeoutError:
logs_all += "\n Timeout"
# Same exit code as the timeout command on linux.
exitcode = 124
break
except asyncio.CancelledError:
logs_all += "\n Cancelled"
# TODO: which exit code? 125 is Operation Canceled
exitcode = 125
break
self._running_cmd_task = None
logs_all += stderr.decode()
logs_all += stdout.decode()
if exitcode != 0:
break
code_file = str(file_names[0]) if len(file_names) > 0 else None
return CommandLineCodeResult(exit_code=exitcode, output=logs_all, code_file=code_file) # type: ignore
async def restart(self) -> None:
"""(Experimental) Restart the code executor."""
warnings.warn(
"Restarting local command line code executor is not supported. No action is taken.",
stacklevel=2,
)

88
python/packages/autogen-core/src/autogen_core/components/code_executor/_impl/utils.py
View File

@@ -1,88 +0,0 @@
# File based from: https://github.com/microsoft/autogen/blob/main/autogen/coding/utils.py
# Credit to original authors
# Will return the filename relative to the workspace path
import re
from pathlib import Path
from typing import Optional
# Raises ValueError if the file is not in the workspace
def get_file_name_from_content(code: str, workspace_path: Path) -> Optional[str]:
first_line = code.split("\n")[0]
# TODO - support other languages
if first_line.startswith("# filename:"):
filename = first_line.split(":")[1].strip()
# Handle relative paths in the filename
path = Path(filename)
if not path.is_absolute():
path = workspace_path / path
path = path.resolve()
# Throws an error if the file is not in the workspace
relative = path.relative_to(workspace_path.resolve())
return str(relative)
return None
def silence_pip(code: str, lang: str) -> str:
"""Apply -qqq flag to pip install commands."""
if lang == "python":
regex = r"^! ?pip install"
elif lang in ["bash", "shell", "sh", "pwsh", "powershell", "ps1"]:
regex = r"^pip install"
else:
return code
# Find lines that start with pip install and make sure "-qqq" flag is added.
lines = code.split("\n")
for i, line in enumerate(lines):
# use regex to find lines that start with pip install.
match = re.search(regex, line)
if match is not None:
if "-qqq" not in line:
lines[i] = line.replace(match.group(0), match.group(0) + " -qqq")
return "\n".join(lines)
PYTHON_VARIANTS = ["python", "Python", "py"]
def lang_to_cmd(lang: str) -> str:
if lang in PYTHON_VARIANTS:
return "python"
if lang.startswith("python") or lang in ["bash", "sh"]:
return lang
if lang in ["shell"]:
return "sh"
else:
raise ValueError(f"Unsupported language: {lang}")
# Regular expression for finding a code block
# ```[ \t]*(\w+)?[ \t]*\r?\n(.*?)[ \t]*\r?\n``` Matches multi-line code blocks.
# The [ \t]* matches the potential spaces before language name.
# The (\w+)? matches the language, where the ? indicates it is optional.
# The [ \t]* matches the potential spaces (not newlines) after language name.
# The \r?\n makes sure there is a linebreak after ```.
# The (.*?) matches the code itself (non-greedy).
# The \r?\n makes sure there is a linebreak before ```.
# The [ \t]* matches the potential spaces before closing ``` (the spec allows indentation).
CODE_BLOCK_PATTERN = r"```[ \t]*(\w+)?[ \t]*\r?\n(.*?)\r?\n[ \t]*```"
def infer_lang(code: str) -> str:
"""infer the language for the code.
TODO: make it robust.
"""
if code.startswith("python ") or code.startswith("pip") or code.startswith("python3 "):
return "sh"
# check if code is a valid python code
try:
compile(code, "test", "exec")
return "python"
except SyntaxError:
# not a valid python code
return "unknown"

32
python/packages/autogen-core/src/autogen_core/components/model_context/__init__.py
View File

@@ -1,32 +0,0 @@
from typing_extensions import deprecated
from ...model_context import BufferedChatCompletionContext as BufferedChatCompletionContextAlias
from ...model_context import ChatCompletionContext as ChatCompletionContextAlias
from ...model_context import HeadAndTailChatCompletionContext as HeadAndTailChatCompletionContextAlias
__all__ = [
"ChatCompletionContext",
"BufferedChatCompletionContext",
"HeadAndTailChatCompletionContext",
]
@deprecated(
"autogen_core.components.model_context.BufferedChatCompletionContextAlias moved to autogen_core.model_context.BufferedChatCompletionContextAlias. This alias will be removed in 0.4.0."
)
class BufferedChatCompletionContext(BufferedChatCompletionContextAlias):
pass
@deprecated(
"autogen_core.components.model_context.HeadAndTailChatCompletionContextAlias moved to autogen_core.model_context.HeadAndTailChatCompletionContextAlias. This alias will be removed in 0.4.0."
)
class HeadAndTailChatCompletionContext(HeadAndTailChatCompletionContextAlias):
pass
@deprecated(
"autogen_core.components.model_context.ChatCompletionContextAlias moved to autogen_core.model_context.ChatCompletionContextAlias. This alias will be removed in 0.4.0."
)
class ChatCompletionContext(ChatCompletionContextAlias):
pass

137
python/packages/autogen-core/src/autogen_core/components/models/__init__.py
View File

@@ -1,137 +0,0 @@
from typing_extensions import deprecated
from ...models import (
AssistantMessage as AssistantMessageAlias,
)
from ...models import ChatCompletionClient as ChatCompletionClientAlias
from ...models import (
ChatCompletionTokenLogprob as ChatCompletionTokenLogprobAlias,
)
from ...models import (
CreateResult as CreateResultAlias,
)
from ...models import (
FinishReasons as FinishReasonsAlias,
)
from ...models import (
FunctionExecutionResult as FunctionExecutionResultAlias,
)
from ...models import (
FunctionExecutionResultMessage as FunctionExecutionResultMessageAlias,
)
from ...models import (
LLMMessage as LLMMessageAlias,
)
from ...models import ModelCapabilities as ModelCapabilitiesAlias
from ...models import (
RequestUsage as RequestUsageAlias,
)
from ...models import (
SystemMessage as SystemMessageAlias,
)
from ...models import (
TopLogprob as TopLogprobAlias,
)
from ...models import (
UserMessage as UserMessageAlias,
)
@deprecated(
"autogen_core.components.models.ChatCompletionClient moved to autogen_core.models.ChatCompletionClient. This alias will be removed in 0.4.0."
)
class ChatCompletionClient(ChatCompletionClientAlias):
pass
@deprecated(
"autogen_core.components.models.ModelCapabilities moved to autogen_core.models.ModelCapabilities. This alias will be removed in 0.4.0."
)
class ModelCapabilities(ModelCapabilitiesAlias):
pass
@deprecated(
"autogen_core.components.models.SystemMessage moved to autogen_core.models.SystemMessage. This alias will be removed in 0.4.0."
)
class SystemMessage(SystemMessageAlias):
pass
@deprecated(
"autogen_core.components.models.UserMessage moved to autogen_core.models.UserMessage. This alias will be removed in 0.4.0."
)
class UserMessage(UserMessageAlias):
pass
@deprecated(
"autogen_core.components.models.AssistantMessage moved to autogen_core.models.AssistantMessage. This alias will be removed in 0.4.0."
)
class AssistantMessage(AssistantMessageAlias):
pass
@deprecated(
"autogen_core.components.models.FunctionExecutionResult moved to autogen_core.models.FunctionExecutionResult. This alias will be removed in 0.4.0."
)
class FunctionExecutionResult(FunctionExecutionResultAlias):
pass
@deprecated(
"autogen_core.components.models.FunctionExecutionResultMessage moved to autogen_core.models.FunctionExecutionResultMessage. This alias will be removed in 0.4.0."
)
class FunctionExecutionResultMessage(FunctionExecutionResultMessageAlias):
pass
LLMMessage = LLMMessageAlias
@deprecated(
"autogen_core.components.models.RequestUsage moved to autogen_core.models.RequestUsage. This alias will be removed in 0.4.0."
)
class RequestUsage(RequestUsageAlias):
pass
FinishReasons = FinishReasonsAlias
@deprecated(
"autogen_core.components.models.CreateResult moved to autogen_core.models.CreateResult. This alias will be removed in 0.4.0."
)
class CreateResult(CreateResultAlias):
pass
@deprecated(
"autogen_core.components.models.TopLogprob moved to autogen_core.models.TopLogprob. This alias will be removed in 0.4.0."
)
class TopLogprob(TopLogprobAlias):
pass
@deprecated(
"autogen_core.components.models.ChatCompletionTokenLogprob moved to autogen_core.models.ChatCompletionTokenLogprob. This alias will be removed in 0.4.0."
)
class ChatCompletionTokenLogprob(ChatCompletionTokenLogprobAlias):
pass
__all__ = [
"ModelCapabilities",
"ChatCompletionClient",
"SystemMessage",
"UserMessage",
"AssistantMessage",
"FunctionExecutionResult",
"FunctionExecutionResultMessage",
"LLMMessage",
"RequestUsage",
"FinishReasons",
"CreateResult",
"TopLogprob",
"ChatCompletionTokenLogprob",
]

71
python/packages/autogen-core/src/autogen_core/components/tool_agent/__init__.py
View File

@@ -1,71 +0,0 @@
from typing import Any
from typing_extensions import deprecated
from ...tool_agent import (
InvalidToolArgumentsException as InvalidToolArgumentsExceptionAlias,
)
from ...tool_agent import (
ToolAgent as ToolAgentAlias,
)
from ...tool_agent import (
ToolException as ToolExceptionAlias,
)
from ...tool_agent import (
ToolExecutionException as ToolExecutionExceptionAlias,
)
from ...tool_agent import (
ToolNotFoundException as ToolNotFoundExceptionAlias,
)
from ...tool_agent import tool_agent_caller_loop as tool_agent_caller_loop_alias
__all__ = [
"ToolAgent",
"ToolException",
"ToolNotFoundException",
"InvalidToolArgumentsException",
"ToolExecutionException",
"tool_agent_caller_loop",
]
@deprecated(
"autogen_core.tool_agent.ToolAgentAlias moved to autogen_core.tool_agent.ToolAgentAlias. This alias will be removed in 0.4.0."
)
class ToolAgent(ToolAgentAlias):
pass
@deprecated(
"autogen_core.tool_agent.ToolExceptionAlias moved to autogen_core.tool_agent.ToolExceptionAlias. This alias will be removed in 0.4.0."
)
class ToolException(ToolExceptionAlias):
pass
@deprecated(
"autogen_core.tool_agent.ToolNotFoundExceptionAlias moved to autogen_core.tool_agent.ToolNotFoundExceptionAlias. This alias will be removed in 0.4.0."
)
class ToolNotFoundException(ToolNotFoundExceptionAlias):
pass
@deprecated(
"autogen_core.tool_agent.InvalidToolArgumentsExceptionAlias moved to autogen_core.tool_agent.InvalidToolArgumentsExceptionAlias. This alias will be removed in 0.4.0."
)
class InvalidToolArgumentsException(InvalidToolArgumentsExceptionAlias):
pass
@deprecated(
"autogen_core.tool_agent.ToolExecutionExceptionAlias moved to autogen_core.tool_agent.ToolExecutionExceptionAlias. This alias will be removed in 0.4.0."
)
class ToolExecutionException(ToolExecutionExceptionAlias):
pass
@deprecated(
"autogen_core.tool_agent.tool_agent_caller_loop moved to autogen_core.tool_agent.tool_agent_caller_loop. This alias will be removed in 0.4.0."
)
def tool_agent_caller_loop(*args: Any, **kwargs: Any) -> Any:
return tool_agent_caller_loop_alias(*args, **kwargs) # type: ignore

75
python/packages/autogen-core/src/autogen_core/components/tools/__init__.py
View File

@@ -1,75 +0,0 @@
from typing import TypeVar
from pydantic import BaseModel
from typing_extensions import deprecated
from ...tools import (
BaseTool as BaseToolAlias,
)
from ...tools import (
BaseToolWithState as BaseToolWithStateAlias,
)
from ...tools import FunctionTool as FunctionToolAlias
from ...tools import (
ParametersSchema as ParametersSchemaAlias,
)
from ...tools import (
Tool as ToolAlias,
)
from ...tools import (
ToolSchema as ToolSchemaAlias,
)
__all__ = [
"Tool",
"ToolSchema",
"ParametersSchema",
"BaseTool",
"BaseToolWithState",
"FunctionTool",
]
ArgsT = TypeVar("ArgsT", bound=BaseModel, contravariant=True)
ReturnT = TypeVar("ReturnT", bound=BaseModel, covariant=True)
StateT = TypeVar("StateT", bound=BaseModel)
@deprecated(
"autogen_core.tools.BaseToolAlias moved to autogen_core.tools.BaseToolAlias. This alias will be removed in 0.4.0."
)
class BaseTool(BaseToolAlias[ArgsT, ReturnT]):
pass
@deprecated("autogen_core.tools.ToolAlias moved to autogen_core.tools.ToolAlias. This alias will be removed in 0.4.0.")
class Tool(ToolAlias):
pass
@deprecated(
"autogen_core.tools.ToolSchemaAlias moved to autogen_core.tools.ToolSchemaAlias. This alias will be removed in 0.4.0."
)
class ToolSchema(ToolSchemaAlias):
pass
@deprecated(
"autogen_core.tools.ParametersSchemaAlias moved to autogen_core.tools.ParametersSchemaAlias. This alias will be removed in 0.4.0."
)
class ParametersSchema(ParametersSchemaAlias):
pass
@deprecated(
"autogen_core.tools.BaseToolWithStateAlias moved to autogen_core.tools.BaseToolWithStateAlias. This alias will be removed in 0.4.0."
)
class BaseToolWithState(BaseToolWithStateAlias[ArgsT, ReturnT, StateT]):
pass
@deprecated(
"autogen_core.tools.FunctionToolAlias moved to autogen_core.tools.FunctionToolAlias. This alias will be removed in 0.4.0."
)
class FunctionTool(FunctionToolAlias):
pass