Confirmation

• I confirm that I am a maintainer and so can use this template. If I am not, I understand this issue will be closed and I will be asked to use a different template.

Issue body

Create an AutoGen module autogen_agentchat.optimize that serves the a unified interface for agent optimization, making DSPy as an optimization implementation — autogen_ext.optimize.dspy.

Below is a drop-in package skeleton you can copy into your repository.
It introduces a new public module autogen_agentchat.optimize that exposes a single, unified entry-point:

from autogen_agentchat.optimize import compile # or optimise()

best_agent, report = compile(
agent = my_autogen_agent,
trainset = train_examples, # List[dspy.Example] | or any iterable
metric = exact_match, # Callable
backend = "dspy", # default
optimizer_name = "MIPROv2", # or "SIMBA", …
optimizer_kwargs = dict(max_steps=16) # forwarded verbatim
)

The first shipping backend lives in autogen_ext.optimize.dspy and wraps DSPy’s optimiser
stack; additional back-ends can be added simply by registering a subclass.

⸻

1 Package layout

autogen_agentchat/
│
├─ optimize/
│ ├─ init.py # public API (compile / list_backends)
│ ├─ _backend.py # abstract base-class & registry
│ └─ _utils.py # shared helpers (wrap agent → DSPy Module)
│
autogen_ext/
└─ optimize/
└─ dspy.py # first concrete backend

⸻

2 Core abstractions

autogen_agentchat/optimize/_backend.py

from future import annotations
from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, Iterable, List, Tuple

Simple registry so new back-ends can self-register

_BACKENDS: Dict[str, "BaseBackend"] = {}

class BaseBackend(ABC):
"""Contract every optimiser back-end must fulfil."""

#: name used in `compile(... backend="<name>")`
name: str = ""

def __init_subclass__(cls, **kw):
    super().__init_subclass__(**kw)
    if cls.name:
        _BACKENDS[cls.name] = cls

# ---- required API --------------------------------------------------
@abstractmethod
def compile(                       # noqa: D401
    self,
    agent: Any,
    trainset: Iterable[Any],
    metric: Callable,
    **kwargs,
) -> Tuple[Any, Dict[str, Any]]:
    """Return (optimised_agent, diagnostics/report)."""

def get_backend(name: str) -> BaseBackend:
try:
return _BACKENDSname
except KeyError:
raise ValueError(
f"Unknown backend '{name}'. Available: {', '.join(_BACKENDS)}"
) from None

⸻

autogen_agentchat/optimize/_utils.py

"""Utility glue for turning AutoGen agents into DSPy-friendly modules."""
from future import annotations
import asyncio, dspy
from typing import Dict, List
from autogen_core.agents.base import Agent
from autogen_core.models import UserMessage, AssistantMessage, SystemMessage

---------------------------------------------------------------------

1. LM adaptor –– AutoGen client ➜ DSPy.LM

---------------------------------------------------------------------

class AutoGenLM(dspy.LM):
def init(self, client):
super().init(model=client.model)
self.client = client

async def _acall(self, messages: List[Dict[str, str]], **kw) -> str:
    autogen_msgs = []
    for m in messages:
        role, content = m["role"], m["content"]
        if role == "user":
            autogen_msgs.append(UserMessage(content))
        elif role == "assistant":
            autogen_msgs.append(AssistantMessage(content))
        else:
            autogen_msgs.append(SystemMessage(content))
    resp = await self.client.create(autogen_msgs, **kw)
    return resp.content

def __call__(self, messages, **kw):
    return asyncio.run(self._acall(messages, **kw))

---------------------------------------------------------------------

2. DSPy module wrapper around an existing Agent

---------------------------------------------------------------------

class DSPyAgentWrapper(dspy.Module):
"""
Exposes agent.system_message and each tool description as learnable prompts.
"""

def __init__(self, agent: Agent):
    super().__init__()
    self._agent = agent

    # turn system prompt & each tool description into learnable strings
    self._system_prompt = dspy.Prompt(agent.system_message)

    self._tool_prompts = []
    for t in agent.tools:
        self._tool_prompts.append(dspy.Prompt(t.description))

    # Signature is generic: user_request → answer
    class _Sig(dspy.Signature):
        """{{system_prompt}}"""
        user_request: str = dspy.InputField()
        answer: str = dspy.OutputField()

    self._predict = dspy.Predict(_Sig)

def forward(self, user_request: str):
    # patch live values into the wrapped agent
    self._agent.system_message = self._system_prompt.value
    for prompt, tool in zip(self._tool_prompts, self._agent.tools):
        tool.description = prompt.value

    reply = asyncio.run(self._agent.run(task=user_request)).messages[-1].content
    return dspy.Prediction(answer=reply)

# convenient handles for back-end to read tuned texts later
@property
def learnable_system_prompt(self):
    return self._system_prompt

@property
def learnable_tool_prompts(self):
    return self._tool_prompts

⸻

3 DSPy implementation back-end

autogen_ext/optimize/dspy.py

from future import annotations
import importlib
from typing import Any, Callable, Iterable, Tuple, Dict
import dspy

from autogen_agentchat.optimize._backend import BaseBackend
from autogen_agentchat.optimize._utils import AutoGenLM, DSPyAgentWrapper

class DSPyBackend(BaseBackend):
"""Optimise AutoGen agents with any DSPy optimiser."""
name = "dspy"

# public compile() required by BaseBackend
def compile(
    self,
    agent: Any,
    trainset: Iterable[Any],
    metric: Callable,
    *,
    lm_client: Any | None = None,
    optimizer_name: str = "SIMBA",
    optimizer_kwargs: Dict[str, Any] | None = None,
    **extra,
) -> Tuple[Any, Dict[str, Any]]:

    if not optimizer_kwargs:
        optimizer_kwargs = {}

    # 1.  Configure DSPy with the supplied AutoGen LM (or grab from .model_client)
    lm_client = lm_client or agent.model_client
    dspy.configure(lm=AutoGenLM(lm_client))

    # 2.  Wrap agent
    wrapper = DSPyAgentWrapper(agent)

    # 3.  Create optimiser instance
    opt_mod = importlib.import_module("dspy.optimizers")
    OptimCls = getattr(opt_mod, optimizer_name)
    optimiser = OptimCls(metric=metric, **optimizer_kwargs)

    # 4.  Compile
    compiled = optimiser.compile(wrapper, trainset=trainset)

    # 5.  Write back tuned texts into the *original* live agent
    agent.system_message = compiled.learnable_system_prompt.value
    for new_desc, tool in zip(
        compiled.learnable_tool_prompts, agent.tools
    ):
        tool.description = new_desc.value

    report = dict(
        optimizer=optimizer_name,
        best_metric=optimiser.best_metric,
        tuned_system_prompt=agent.system_message,
        tuned_tool_descriptions=[t.description for t in agent.tools],
    )
    return agent, report

⸻

4 Public façade

autogen_agentchat/optimize/init.py

from future import annotations
from typing import Any, Iterable, Callable, Dict, Tuple

from ._backend import get_backend, _BACKENDS # re-export registry

def compile(
agent: Any,
trainset: Iterable[Any],
metric: Callable,
*,
backend: str = "dspy",
**kwargs,
) -> Tuple[Any, Dict[str, Any]]:
"""
Optimise the system_message and tool descriptions of an AutoGen agent.

Parameters
----------
agent
    Any subclass of autogen_core.agents.base.Agent.
trainset
    Iterable of supervision examples (DSPy Examples or anything the
    back-end accepts).
metric
    Callable(gold, pred) → float | bool used by the optimiser.
backend
    Name of the registered optimisation backend (default: "dspy").
kwargs
    Extra parameters forwarded verbatim to the backend.

Returns
-------
(optimised_agent, report)
"""
backend_impl = get_backend(backend)
return backend_impl.compile(agent, trainset, metric, **kwargs)

def list_backends():
"""Return the names of all available optimisation back-ends."""
return sorted(_BACKENDS)

⸻

5 Example usage (end-to-end)

from autogen_core.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.optimize import compile, list_backends
import dspy

➊ Build a toy agent ---------------------------------------------------

def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y

agent = AssistantAgent(
name="calc",
model_client=OpenAIChatCompletionClient("gpt-4o-mini"),
system_message="You are a helpful calculator.",
tools=[add],
)

➋ Minimal trainset ----------------------------------------------------

train = [
dspy.Example(user_request="2+2", answer="4").with_inputs("user_request"),
dspy.Example(user_request="Add 3 and 5", answer="8").with_inputs("user_request"),
]

➌ Optimise using the unified API --------------------------------------

opt_agent, report = compile(
agent = agent,
trainset = train,
metric = lambda g, p, **_: g.answer == p.answer,
backend = "dspy", # default anyway
optimizer_name = "MIPROv2",
optimizer_kwargs = dict(max_steps=8),
)

print(report["tuned_system_prompt"])
print(opt_agent("What is 17+4?")) # → "21"

⸻

What you gained
• One import path (autogen_agentchat.optimize.compile) that hides DSPy specifics.
• Seamless future back-ends – drop my_backend.py with class MyBackend(BaseBackend) and it’s auto-discoverable.
• No runtime coupling: the agent keeps running in pure AutoGen; only prompt strings get overwritten.

Use this as a starting scaffold; fill in logging, richer metrics, or dataset adapters as your project requires.