Agentstack (formerly BeeAI) Integration

agentstack_platform/agentstack_platform.py

@@ -0,0 +1,78 @@
+import os
+import asyncio
+from typing import Annotated, Callable
+
+from a2a.types import Message
+from a2a.utils.message import get_message_text
+from agentstack_sdk.a2a.types import AgentMessage
+from agentstack_sdk.server import Server
+from agentstack_sdk.a2a.extensions import (
+    LLMServiceExtensionServer, LLMServiceExtensionSpec,
+    TrajectoryExtensionServer, TrajectoryExtensionSpec,
+    AgentDetail
+)
+from agentstack_sdk.a2a.extensions.ui.form import (
+    FormExtensionServer, FormExtensionSpec, FormRender, TextField
+)
+from mellea import MelleaSession, start_session
+from mellea.backends.openai import OpenAIBackend
+from mellea.stdlib.requirement import req, Requirement, simple_validate
+from mellea.stdlib.sampling import RejectionSamplingStrategy
+import inspect
+
+
+def agentstack_app(func: Callable) -> Callable:
+     """Serves as a wrapper that takes any Mellea program and converts it to a BeeAI Agent. This is an example for an email writer."""
+     server = Server()
+
+     params : dict = inspect.signature(func).parameters # Mapping params from Mellea function onto form inputs
+     form_fields : list[str] = list(params.keys())[1:]
+     all_fields : list[TextField] = []
+
+     for field in form_fields:
+         all_fields.append(TextField(id=field, label=field, col_span=2)) #Maps all input params from Mellea agent into BeeAI Forms
+
+
+     form_render = FormRender(
+                id="input_form",
+                title="Please provide your information",
+                columns=2,
+                fields=all_fields
+            )
+     form_extension_spec = FormExtensionSpec(form_render)
+
+
+     @server.agent(name="mellea_agent")
+
+     async def wrapper(input: Message,
+                     llm: Annotated[LLMServiceExtensionServer, LLMServiceExtensionSpec.single_demand()],
+                     trajectory: Annotated[TrajectoryExtensionServer, TrajectoryExtensionSpec()],
+                     form: Annotated[FormExtensionServer, 
+                                     form_extension_spec]):
+
+
+        form_data = form.parse_form_response(message=input)
+        inputs = [form_data.values[key].value for key in form_data.values] # Extracting all of the user inputs from the form
+        llm_config = llm.data.llm_fulfillments.get("default")
+        m = MelleaSession(OpenAIBackend(model_id=llm_config.api_model, api_key=llm_config.api_key, base_url=llm_config.api_base))
+
+        sampling = await asyncio.to_thread(func, m, *inputs)
+        #print([elem for elem in sampling]) #printing all of these generations
+        if sampling.success:
+            yield AgentMesage(text=sampling.value)
+            return
+
+        for idx in range(len(sampling.sample_generations)):
+            yield trajectory.trajectory_metadata(
+                 title=f"Attempt {idx + 1}", content=f"Generating message...") 
+            validations = sampling.sample_validations[idx] # Print out validations
+            status = "\n".join(f"{'✓' if bool(v) else '✗'} {getattr(r, 'description', str(r))}" for r, v in validations)
+            yield trajectory.trajectory_metadata(title=f"✗ Attempt {idx + 1} failed", content=status) 
+
+        yield AgentMessage(text=sampling.value)
+
+     server.run(host=os.getenv("HOST", "127.0.0.1"), port=int(os.getenv("PORT", 8000)))
+
+     return wrapper
+
+

agentstack_platform/agentstack_readme.md

@@ -0,0 +1,68 @@
+# Mellea-Agentstack
+
+Mellea is a library for writing generative programs. 
+Agentstack is an open-source framework for building production-grade multi-agent systems.
+This example serves to merge both libraries with a simple module that will allow users to transform
+their Mellea programs into Agentstack agentic interfaces with structured (form) inputs. 
+
+We provide the example of an email writer. Only text inputs are currently supported.
+
+# Installing Agentstack
+
+To install Agentstack, follow the instructions at this page: https://agentstack.beeai.dev/introduction/welcome
+
+
+# Running the example
+
+Then, in order to run the example email writer, run the code below in the root of the directory:
+```bash
+uv run --with mellea --with agentstack-sdk docs/examples/agentstack_agent.py
+```
+
+In a separate terminal, either run
+```bash
+agentstack run mellea_agent
+```
+
+OR open the UI and select the **mellea-agent**.
+
+```bash
+agentstack ui
+```
+
+# Tutorial
+
+To create your own Agentstack agent with Mellea, write a traditional program with Mellea, as shown below. We provide the source code of the email writer.
+
+```bash
+@agentstack_app
+def mellea_func(m: MelleaSession, sender: str, recipient, subject: str, topic: str) -> tuple[ModelOutputThunk, Context] | SamplingResult:
+    """
+    Example email writing module that utilizes an IVR loop in Mellea to generate an email with a specific list of requirements.
+    Inputs:
+        sender: str
+        recipient: str
+        subject: str
+        topic: str
+    Output:
+        sampling: tuple[ModelOutputThunk, Context] | SamplingResult
+    """
+    requirements = [
+        req("Be formal."),
+        req("Be funny."),
+        req(f"Make sure that the email is from {sender}, is towards {recipient}, has {subject} as the subject, and is focused on {topic} as a topic"),
+        Requirement("Use less than 100 words.",
+                   validation_fn=simple_validate(lambda o: len(o.split()) < 100))
+    ]
+    description = f"Write an email from {sender}. Subject of email is {subject}. Name of recipient is {recipient}. Topic of email should be {topic}."
+    sampling = m.instruct(description=description, requirements=requirements, strategy=RejectionSamplingStrategy(loop_budget=3), return_sampling_results=True)
+    return sampling
+```
+
+Adjust ```requirements``` and ```prompt``` as necessary.
+
+As shown above, note that the first parameter should be an **m** object.
+
+Then, to deploy your Mellea program to Agentstack, wrap with the ```@agentstack_app``` decorator, as shown above.
+
+Place your code in the ```docs/examples/``` folder.

docs/examples/agentstack_agent.py

@@ -0,0 +1,45 @@
+"""
+Example use case for BeeAI integration: utilizing a Mellea program to write an email with an IVF loop.
+"""
+import os
+import asyncio
+import sys
+import mellea
+from typing import Annotated
+
+from mellea.stdlib.requirement import req, check, simple_validate
+from mellea import MelleaSession, start_session
+from mellea.stdlib.base import ChatContext, ModelOutputThunk
+
+from mellea.stdlib.sampling import RejectionSamplingStrategy
+from mellea.stdlib.sampling.types import SamplingResult
+from mellea.stdlib.sampling.base import Context
+from mellea.stdlib.requirement import req, Requirement, simple_validate
+from agentstack_platform.agentstack_platform import agentstack_app
+
+
+@agentstack_app
+def mellea_func(m: MelleaSession, sender: str, recipient, subject: str, topic: str) -> tuple[ModelOutputThunk, Context] | SamplingResult:
+    """
+    Example email writing module that utilizes an IVR loop in Mellea to generate an email with a specific list of requirements.
+    Inputs:
+        sender: str
+        recipient: str
+        subject: str
+	topic: str
+    Output:
+	sampling: tuple[ModelOutputThunk, Context] | SamplingResult
+    """
+    requirements = [
+        req("Be formal."),
+        req("Be funny."),
+	req(f"Make sure that the email is from {sender}, is towards {recipient}, has {subject} as the subject, and is focused on {topic} as a topic"),
+        Requirement("Use less than 100 words.", 
+                   validation_fn=simple_validate(lambda o: len(o.split()) < 100))
+    ]
+    description = f"Write an email from {sender}. Subject of email is {subject}. Name of recipient is {recipient}. Topic of email should be {topic}."
+    sampling = m.instruct(description=description, requirements=requirements, strategy=RejectionSamplingStrategy(loop_budget=3), return_sampling_results=True)
+    return sampling
+
+
+