Python: Concept of self-hosted agent

python/samples/demos/self_hosted_agent/README.md

@@ -0,0 +1,38 @@
+## Self Hosted Agents
+
+This sample demonstrates how to create multi-agent group chat application where agents are hosted outside of the Semantic Kernel. `agents` folder contains `fastapi` server that hosts 2 agents and exposes them via REST apis. `app` folder contains the Semantic Kernel application that orchestrates between the agents using `AgentGroupChat`.
+
+### Running the sample
+
+You will need to deploy Azure OpenAI models in [Azure AI Foundry](https://learn.microsoft.com/en-us/azure/ai-studio/). Keep the endpoint and the key handy.
+
+#### Running the server
+1. Navigate to `agents` folder.
+2. Create a virtual environment and install the dependencies.
+```bash
+uv venv --python=3.10
+source .venv/bin/activate
+```
+3. Install the dependencies.
+```bash
+uv sync
+```
+4. Create `.env` file using `.env.example` as template.
+5. Run the server.
+```bash
+fastapi run main.py
+```
+
+Note the address the fastapi server is running on. You will need it in the next step.
+
+#### Running the app
+1. In a different terminal, navigate to `app` folder.
+2. Create `.env` file using `.env.example` as template. Replace `server_url` with the address of the fastapi server.
+```bash
+REVIEWER_AGENT_URL = "<server_url>/agent/reviewer"
+COPYWRITER_AGENT_URL = "<server_url>/agent/copywriter"
+```
+3. Run the app.
+```bash
+python main.py
+```

python/samples/demos/self_hosted_agent/agents/.env.example

@@ -0,0 +1,3 @@
+AZURE_AI_AGENT_PROJECT_CONNECTION_STRING = "<example-connection-string>"
+AZURE_AI_INFERENCE_ENDPOINT = "<example-endpoint>"
+AZURE_AI_INFERENCE_MODEL_DEPLOYMENT_NAME = "<example-model-deployment-name>"

python/samples/demos/self_hosted_agent/agents/main.py

@@ -0,0 +1,90 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import logging
+import os
+from abc import ABC
+from typing import Any
+
+from azure.ai.projects.aio import AIProjectClient
+from azure.identity.aio import DefaultAzureCredential
+from dotenv import load_dotenv
+from fastapi import FastAPI
+from openai.types.chat.chat_completion import ChatCompletion
+from openai.types.chat.chat_completion_system_message_param import ChatCompletionSystemMessageParam
+from pydantic import BaseModel, ConfigDict
+
+load_dotenv()
+
+app = FastAPI()
+
+client = AIProjectClient.from_connection_string(
+    credential=DefaultAzureCredential(), conn_str=os.environ["AZURE_AI_AGENT_PROJECT_CONNECTION_STRING"]
+)
+
+logger = logging.getLogger("custom_agent_logger")
+
+REVIEWER_INSTRUCTIONS = """
+You are an art director who has opinions about copywriting born of a love for David Ogilvy.
+The goal is to determine if the given copy is acceptable to print.
+If so, state that it is approved. Only include the word "approved" if it is so.
+If not, provide insight on how to refine suggested copy without example.
+"""
+
+COPYWRITER_INSTRUCTIONS = """
+You are a copywriter with ten years of experience and are known for brevity and a dry humor.
+The goal is to refine and decide on the single best copy as an expert in the field.
+Only provide a single proposal per response.
+You're laser focused on the goal at hand.
+Don't waste time with chit chat.
+Consider suggestions when refining an idea.
+"""
+
+
+class APIRequestFormat(BaseModel, ABC):
+    """Specific settings for the Chat Completion endpoint."""
+
+    model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True, validate_assignment=True)
+
+    stream: bool = False
+    messages: list[dict[str, Any]]
+    parallel_tool_calls: bool | None = None
+    tools: list[dict[str, Any]] | None = None
+    tool_choice: str | None = None
+
+
+@app.post("/agent/reviewer")
+async def reviewer_agent(request: APIRequestFormat) -> ChatCompletion:
+    messages = request.messages
+    logger.info("Financial Coach agent")
+    logger.info(messages)
+    openai = await client.inference.get_azure_openai_client(api_version="2024-06-01")
+    # replace the system message in messages with custom system message
+    if messages[0]["role"] == "system" or messages[0]["role"] == "developer":
+        messages[0]["content"] = REVIEWER_INSTRUCTIONS
+    else:
+        messages = [ChatCompletionSystemMessageParam(role="system", content=REVIEWER_INSTRUCTIONS), *messages]
+
+    return await openai.chat.completions.create(
+        model=os.getenv("AZURE_AI_INFERENCE_MODEL_DEPLOYMENT_NAME") or "gpt-4o",
+        messages=messages,
+        **request.model_dump(exclude={"messages"}, exclude_none=True),
+    )
+
+
+@app.post("/agent/copywriter")
+async def copywriter_agent(request: APIRequestFormat) -> ChatCompletion:
+    messages = request.messages
+    logger.info("Guardrail agent")
+    logger.info(messages)
+    openai = await client.inference.get_azure_openai_client(api_version="2024-06-01")
+    # replace the system message in messages with custom system message
+    if messages[0]["role"] == "system" or messages[0]["role"] == "developer":
+        messages[0]["content"] = COPYWRITER_INSTRUCTIONS
+    else:
+        messages = [ChatCompletionSystemMessageParam(role="system", content=COPYWRITER_INSTRUCTIONS), *messages]
+
+    return await openai.chat.completions.create(
+        model=os.getenv("AZURE_AI_INFERENCE_MODEL_DEPLOYMENT_NAME") or "gpt-4o",
+        messages=messages,
+        **request.model_dump(exclude={"messages"}, exclude_none=True),
+    )

python/samples/demos/self_hosted_agent/agents/pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "agents"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "aiohttp>=3.11.12",
+    "azure-ai-projects>=1.0.0b5",
+    "azure-identity>=1.20.0",
+    "fastapi[standard]>=0.115.8",
+    "openai>=1.62.0",
+    "pydantic>=2.10.6",
+    "python-dotenv>=1.0.1",
+]

python/samples/demos/self_hosted_agent/app/.env.example

@@ -0,0 +1,2 @@
+REVIEWER_AGENT_URL = "<server_url>/agent/reviewer"
+COPYWRITER_AGENT_URL = "<server_url>/agent/copywriter"

python/samples/demos/self_hosted_agent/app/main.py

@@ -0,0 +1,75 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+import sys
+
+from dotenv import load_dotenv
+from self_hosted_api_chat_completion import SelfHostedChatCompletion
+
+from semantic_kernel.agents import AgentGroupChat, ChatCompletionAgent
+from semantic_kernel.agents.strategies.termination.termination_strategy import TerminationStrategy
+from semantic_kernel.contents import AuthorRole, ChatMessageContent
+from semantic_kernel.kernel import Kernel
+
+load_dotenv()
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+#####################################################################
+# The following sample demonstrates how to create a self-hosted.....#
+# and have them participate in a group chat to work towards   ......#
+# the user's requirement.                                           #
+#####################################################################
+
+
+class ApprovalTerminationStrategy(TerminationStrategy):
+    """A strategy for determining when an agent should terminate."""
+
+    async def should_agent_terminate(self, agent, history):
+        """Check if the agent should terminate."""
+        return "approved" in history[-1].content.lower()
+
+
+REVIEWER_NAME = "ArtDirector"
+COPYWRITER_NAME = "CopyWriter"
+
+
+async def main():
+    try:
+        kernel = Kernel()
+
+        agent_reviewer = ChatCompletionAgent(
+            id="artdirector",
+            kernel=kernel,
+            name=REVIEWER_NAME,
+            service=SelfHostedChatCompletion(url=os.getenv("REVIEWER_AGENT_URL") or "", ai_model_id=REVIEWER_NAME),
+        )
+
+        agent_writer = ChatCompletionAgent(
+            id="copywriter",
+            kernel=kernel,
+            name=COPYWRITER_NAME,
+            service=SelfHostedChatCompletion(url=os.getenv("COPYWRITER_AGENT_URL") or "", ai_model_id=COPYWRITER_NAME),
+        )
+
+        chat = AgentGroupChat(
+            agents=[agent_writer, agent_reviewer],
+            termination_strategy=ApprovalTerminationStrategy(agents=[agent_reviewer], maximum_iterations=10),
+        )
+
+        input = "a slogan for a new line of electric cars."
+
+        await chat.add_chat_message(ChatMessageContent(role=AuthorRole.USER, content=input))
+        print(f"# {AuthorRole.USER}: '{input}'")
+
+        async for content in chat.invoke():
+            print(f"# {content.role} - {content.name or '*'}: '{content.content}'")
+
+        print(f"# IS COMPLETE: {chat.is_complete}")
+    except Exception as e:
+        print(e)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())