How to make a conversational multi agent system on Autogen

[karun19]

Describe the bug
I want to create a team of agents (using SelectorGroupChat) to address a issues at a fuel station. I want it to be a conversational like any chat bot. I have tried to pass instructions or sequesnce of steps for the flow. But, I see it is not consistent especially when user input is required. Mostly, it looks as if it is in hurry to finish the flow rather follow the instructions given.

Questions:

1. what is rite way of passing flow specific instructions (kind of SOP)?
2. Is my whole implementation is valid?
3. Is there a way to pass a prompt to the Orchestrator?

To Reproduce

Agents configuration:

{
  "agents": [
    {
      "name": "SOPAgent",
      "system_message": "You are agent to fetch the SOP to address the issue using sop_for_issue_addressal and follows the SOP steps to help the customer. You must be the first agent to be consulted.",
      "description": "An agent to fetch the SOP to address the issue and follows the SOP steps to help the customer. This will be the first agent consulted.",
      "tools": [
         "sop_for_issue_addressal"
      ]
    },
    {
      "name": "CustomerSentimentAgent",
      "system_message": "an intelligent assistant designed to analyze customer sentiment based on predefined policies. Your task is to fetch the Customer Sentiment Policy document and derive the sentiment of the customer by evaluating the conversation in the session against the fetched policy. Ensure that your analysis aligns with the policy guidelines and provide a clear sentiment classification (e.g., positive, neutral, negative) with a rationale based on the conversation context. Strictly - do no offer freebie if the sentiment is neutral.",
      "description": "An agent responsible for analyzing customer sentiment by fetching the Customer Sentiment Policy document and evaluating ongoing conversations. The agent ensures that sentiment analysis is policy-driven, providing an accurate classification of customer sentiment with justifications based on the current session's dialogue. Uses derive_customer_sentiment() for derive Customer sentiment. Strictly - do no offer freebie if the sentiment is neutral.",
      "tools": [
        "derive_customer_sentiment"
      ]
    },
    {
      "name": "PumpIssueHandlerAgent",
      "system_message": "You are a customer support assistant for a fuel station having only four pumps pump #1, pump #2, pump #3 and pump #4. Do not accept any other pumps. Request user to provide pump number if not provided allready.  If a pump is out of service or occupied, recommend an alternative pump(#1, #2, #3 or #4) to the customer. You should communicate the issue to the appropriate personnel (uses send_message_to_technician or send_message_to_store_cashier) as suggested in the already fetched SOP. Maintain professionalism and empathy while providing effective solutions. Display empathy - start with 'I’m sorry…'. Strictly - Reponse should be in 15 - 30 words only. Sample message: Hi! I’m sorry you’re having trouble with Pump #3. The ‘See the Cashier’ message usually means  there’s a temporary issue and we have notified technician. Please use either pump #1, #2, #3.",
      "description": "You are a customer support assistant for a fuel station having only four pumps pump #1, pump #2, pump #3 and pump #4. Do not accept any other pumps. Request user to provide pump number if not provided allready.  If a pump is out of service or occupied, recommend an alternative pump(#1, #2, #3 or #4) to the customer. You should communicate the issue to the appropriate personnel (uses send_message_to_technician or send_message_to_store_cashier) as suggested in the already fetched SOP. Maintain professionalism and empathy while providing effective solutions. Strictly - Reponse should be in 15 - 30 words only.",
      "tools": [
        "analyze_pump_health",
        "send_message_to_technician",
        "send_message_to_store_cashier",
        "suggest_alternate_pump"
      ]
    },
    {
      "name": "FreebieProviderAgent",
      "system_message": "You are a freebie provider assistant for a fuel station. You are the agent must be consulted to provide freebies. Your primary responsibility is to fetch and follow the customer satisfaction policy using fetch_customer_satisfaction_policy and provide freebies as suggested in the policy to improve customer satisfaction. Ask user for the acceptance and wait till user responds. After customer response, irrespective of customer sentiment once again You should derive and publish the customer sentiment using fetch_customer_sentiment_policy. Engage FreebieProviderAgent only if freebies to be provided. Strictly - Reponse should be in 20 - 30 words only. Sample message: For the inconvenience caused, we’d love to offer you a coupon on your next purchase! — <coupon details>",
      "description": "Fetch and fefer to customer satisfaction policy using fetch_customer_satisfaction_policy and provide freebies as suggested in the policy. You are the agent must be consulted to provide freebies. Ask user for the acceptance and wait till user responds. After customer response, irrespective of customer sentiment once again you should derive and publish the customer sentiment using fetch_customer_sentiment_policy.  Engage FreebieProviderAgent only if freebies to be provided. Strictly - Reponse should be in 20 - 30 words only.",
      "tools": [
        "fetch_customer_satisfaction_policy",
        "offer_free_coffee",
        "offer_coupon"
      ]
    }
  ]
}

Corresponding Tools code:

import httpx
import json
import re
import requests

# A function to analyze pump health.
async def analyze_pump_health(pumpNumber: str) -> str:
    """A function to analyze pump health."""
    print(f"Analyzing the pump issue -- Analyzed and found pump #{pumpNumber} is out of order")
    push_response_for_issue_to_customer(pumpNumber)
    return f"Analyzed and found pump #{pumpNumber} is out of order. Go ahead and communicate the issue to the concerned person as defined in the SOP. Engage PumpIssueResolverAgent to resolve and see if there are alternate pumps."

# A function to push message to technician.
async def send_message_to_technician(pumpNumber: str) -> str:
    """A function to push message to technician."""
    print(f"Pushing message to Technician -- A customer has reported pump #{pumpNumber} is not working. Please attend immediately.")
    # push_message_to_store_cashier(pumpNumber)
    return f"Technician has been notified and requested to attend immediately to pump #{pumpNumber}."

# A function to push message to store cashier.
async def send_message_to_store_cashier(pumpNumber: str) -> str:
    """A function to push message to store cashier."""
    print(f"Pushing message to the store cashier -- A customer has reoprted pump #{pumpNumber} is not working. Please verify and place out of order board immediately.")
    return f"Store cashier has been notified and requested to verify pump #{pumpNumber} and place an out-of-order board immediately."

# A function to push message to customer.
def push_response_for_issue_to_customer(pumpNumber: str) -> str:
    """A function to push message to customer."""
    print(f"Pushing message to the customer -- I'm sorry to hear about the issue with pump #{pumpNumber}.")
    # return f"Response message sent to customer confirming the issue with pump #{pumpNumber} and notifying the technician. Proceed with resolution steps. Inform the cashier and technician."

# A function to push message to customer on alternate pump usage.
async def suggest_alternate_pump() -> str:
    """A function to push message to customer on alternate pump usage."""
    print(f"Pushing message to the customer on alternate -- Inform user that  other pumps (except one which is reported not working pump #1, pump #2, pump #3 and pump #4) are operational and advised to try any of those.")
    return f"Inform user that other pumps (except one which is reported not working pump #1, pump #2, pump #3 and pump #4) are operational and advised to try any of those. Derive the customer sentiment and refer to customer satisfaction policy for enhancing customer satifaction"

# A function to fetch customer satisfaction policy
async def fetch_customer_satisfaction_policy() -> str:
    """A function to fetch customer satisfaction policy"""

    #print(f"Fetched customer satisfaction policy: Offer a coupon to the customer only if the customer sentiment is negative.  Strictly - do not offer freebie if the sentiment is neutral.")
    #return f"Customer satisfaction policy: \\n Offer a coffee to the customer only if the customer sentiment is negative.  Strictly - do not offer freebie if the sentiment is neutral."
    api_url = "https://prod-127.westus.logic.azure.com:443/workflows/9b7007dsae43d144f13afcba26d8ee59622/triggers/manual/paths/invoke"
    query_params = {
        "api-version": "2016-06-01",
        "sp": "/triggers/manual/run",
        "sv": "1.0",
        "sig": "dsa"
    }
    payload = {"docType":"csat_policy"}

    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(api_url, params=query_params, json=payload, timeout=30.0)
            response.raise_for_status()
            customer_sat_policy_response = response.json()
            print("customer_sat_policy_response:::",customer_sat_policy_response)
            print("customer_sat_policy_response:::",customer_sat_policy_response["sopsteps"])

            # Extracting 'CanvasContent1'
            canvas_content = str(customer_sat_policy_response["sopsteps"])
            return canvas_content
        except requests.exceptions.RequestException as e:
            return f"Failed to fetch customer_sat_policy: {e}{payload}"
        except ValueError:
            return "Error parsing API response. Please check the format."

# A function to fetch customer sentiment policy
async def fetch_customer_sentiment_policy() -> str:
    """A function to fetch customer sentiment policy"""
    print(f"Fetched customer sentiment policy: If user is reporting a issue the sentiment is negative. If the user has got the satisfactory resolution for the reoprted issue, the sentiment is positive.")
    return f"Customer sentiment policy: \\n If the user has got the satisfactory resolution for the reoprted issue, the sentiment is positive. \\n The agent should ensure that sentiment analysis is policy-driven, providing an accurate classification of customer sentiment with justifications based on the current session's dialogue."

# A function to derive customer sentiment
async def derive_customer_sentiment() -> str:
    """A function to fetch customer sentiment"""
    print(f"customer sentiment policy: If user reported an issue and sounds frustrated, the sentiment is negative. If the user has got the satisfactory resolution for the reoprted issue, the sentiment is positive.")
    return f"Customer sentiment policy: \\n If the user has got the satisfactory resolution for the reoprted issue, the sentiment is positive. \\n The agent should ensure that sentiment analysis is policy-driven, providing an accurate classification of customer sentiment with justifications based on the current session's dialogue."

# A function to compensate the customer dissatisfaction.
async def offer_free_coffee() -> str:
    """A function to offer free coffee to compensate the customer dissatisfaction."""
    print(f"Pushing message to the customer on compensation -- Offer a free coffee to the customer.")
    return f"User was offered a free coffee as a gesture of goodwill. Once again derive the customer sentiment post user response on the freebie offer."

# A function to compensate the customer dissatisfaction.
async def offer_coupon() -> str:
    """A function to offer coupon to compensate the customer dissatisfaction."""
    print(f"Pushing message to the customer on compensation -- Offer a coupon to the customer.")
    return f"User was offered a coupon as a gesture of goodwill. Once again derive the customer sentiment post user response on the freebie offer."

async def sop_for_issue_addressal() -> str:
    """A function to fetch SOP for issue addressal."""
    print(f"Fetched SOP for issue addressal")
    return (f"Workflow should be executed in the sequesnce using the available agents." 
            f"1) Derive customer sentiment."
            f"2) When user reports a pump issue engage related handler agent to verify the pump issue."
            f"3) If the pump issue is confirmed communicate to the concerned people."
            f"4) Try to resolve the issue by suggesting the alternate availabe pumps."
            f"5) Fetch customer satisfaction policy."
            f"6) Follow the customer satisfaction policy and if any freebies to be provided."
            f"7) Wait for the user response on freebie."
            f"8) Instruct Orchestrator to derive the customer sentiment again to check the improvement") 

app.py:

import asyncio # Import asyncio
import json
import logging
import os
import requests
import base64
import importlib
import re
import glob
from typing import Any, Awaitable, Callable, Optional
from pydantic import BaseModel
from datetime import datetime # Import datetime

import aiofiles
import yaml
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent, SocietyOfMindAgent
from autogen_agentchat.base import TaskResult
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.messages import TextMessage, UserInputRequestedEvent
from autogen_agentchat.teams import MagenticOneGroupChat, SelectorGroupChat
from autogen_core import CancellationToken
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Request, Query # Import Query for receive_approval
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse # Import JSONResponse
from fastapi.staticfiles import StaticFiles
from autogen_ext.models.openai import AzureOpenAIChatCompletionClient
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from datetime import datetime
from fastapi.responses import HTMLResponse
""" from tools.car_rental_tools import *
from tools.car_rental_charges_dispute_tools import * """
# tool_modules = glob.glob('tools/*_tools.py') # Local tools
from autogen_ext.tools.mcp import SseServerParams, create_mcp_server_session, mcp_server_tools
from autogen_core.logging import LLMCallEvent
from autogen_core import EVENT_LOGGER_NAME
from tools.team_config_tools import router as team_config_router

# Local tools
""" for filepath in tool_modules:
    module_name = os.path.splitext(os.path.basename(filepath))[0]
    full_module = f'tools.{module_name}'
    module = importlib.import_module(full_module)
    
    # Optional: Import all public names into global namespace
    globals().update({k: getattr(module, k) for k in dir(module) if not k.startswith('_')}) """

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) # Set to DEBUG for more detailed output

app = FastAPI()
app.include_router(team_config_router, prefix="/api/git")

# Load team configurations from external JSON file
CONFIG_FILE = "teams_config.json"
shared_state = {
    "client_id": ""
}
agent_names = []
async def load_team_config():
    if os.environ.get("TEST_ENV", "cloud") == 'local':
        if os.path.exists(CONFIG_FILE):
            with open(CONFIG_FILE, "r") as file:
                return json.load(file)
        return []
    file_content = await get_teams_config_file(CONFIG_FILE)
    try:
        teams_config = json.loads(file_content)
        return teams_config
    except json.JSONDecodeError as e:
        print(f"❌ JSON Error in '{CONFIG_FILE}': {e}")
        return []
    except Exception as e:
        print(f"❌ Unexpected error while reading '{CONFIG_FILE}': {e}")
        return []


async def get_team_config(request: str):
    TEAM_CONFIG = await load_team_config()
    for team in TEAM_CONFIG:
        if team.get("teamName", "").lower() == request.lower():
            return team  # ✅ Ensure it returns the correct team config
    return {}

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allows all origins
    allow_credentials=True,
    allow_methods=["*"],  # Allows all methods
    allow_headers=["*"],  # Allows all headers
)

model_config_path = "model_config.yaml"
state_path = "team_state.json"
history_path = "team_history.json"

# Serve static files
app.mount("/static", StaticFiles(directory="."), name="static")


class LLMUsageTracker(logging.Handler):
    def __init__(self) -> None:
        """Logging handler that tracks the number of tokens used in the prompt and completion."""
        super().__init__()
        self._prompt_tokens = 0
        self._completion_tokens = 0
        self._model = []

    @property
    def model(self) -> int:
        return self._model
        
    @property
    def tokens(self) -> int:
        return self._prompt_tokens + self._completion_tokens

    @property
    def prompt_tokens(self) -> int:
        return self._prompt_tokens

    @property
    def completion_tokens(self) -> int:
        return self._completion_tokens

    def reset(self) -> None:
        self._prompt_tokens = 0
        self._completion_tokens = 0

    def emit(self, record: logging.LogRecord) -> None:
        """Emit the log record. To be used by the logging module."""
        try:
            # Use the StructuredMessage if the message is an instance of it
            #print(record.msg)
            if isinstance(record.msg, LLMCallEvent):
                #print(record.msg)
                event = record.msg
                self._prompt_tokens += event.prompt_tokens
                self._completion_tokens += event.completion_tokens
                #self._model.append(event.response.get("model"))
                #print(event.response.get("model"))
                #print(event.response)
        except Exception:
            self.handleError(record)

# Set up the logging configuration to use the custom handler
logger = logging.getLogger(EVENT_LOGGER_NAME)
logger.setLevel(logging.INFO)


class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []
        #self.active_connections: dict(str, WebSocket) = {}

    async def connect(self, websocket: WebSocket, clientId):
        await websocket.accept()
        self.active_connections.append(websocket)
        #self.active_connections[clientId] = websocket

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_admin_message(self, message: str):
        websocket = self.active_connections.get("admin")
        await websocket.send_json(message)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_json(message)
        
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            try:
                await connection.send_json(message)
            except Exception as e:
                print(f"❌ Connection not found, I will continue: {e}")


manager = ConnectionManager()

# --- Input Management for HTTP Approval ---
class InputManager:
    def __init__(self):
        # A dictionary to hold an asyncio.Queue for each active client/task
        self.pending_inputs: dict[str, asyncio.Queue] = {}

    async def get_user_input(self, client_id: str) -> str:
        """
        Retrieves user input for a given client_id. Blocks until input is available.
        This is called by the UserProxyAgent's input_func.
        """
        if client_id not in self.pending_inputs:
            self.pending_inputs[client_id] = asyncio.Queue()
        print(f"[InputManager] UserProxyAgent for '{client_id}' waiting for input...")
        # This is where the agent flow (and thus the /start_chat request) will pause
        user_message = await self.pending_inputs[client_id].get()
        print(f"[InputManager] UserProxyAgent for '{client_id}' received input: '{user_message}'")
        return user_message

    async def put_user_input(self, client_id: str, message: str):
        """
        Puts a message into the queue for a given client_id, unblocking get_user_input.
        This is called by the /receive_approval HTTP endpoint.
        """
        if client_id not in self.pending_inputs:
            self.pending_inputs[client_id] = asyncio.Queue()
        await self.pending_inputs[client_id].put(message)
        print(f"[InputManager] Placed message for client '{client_id}' into queue.")

input_manager = InputManager() # Instantiate the InputManager

text_termination = TextMentionTermination("TERMINATE")
max_msg_termination = MaxMessageTermination(max_messages=100)
combined_termination = max_msg_termination | text_termination
model_client = AzureOpenAIChatCompletionClient(
    model="gpt-4o", temperature=0.0, top_p=0.0, frequency_penalty=0.0, presence_penalty=0.0
)

@app.get("/")
async def root():
    """Serve the chat interface HTML file."""
    return FileResponse("app_dashboard.html")


async def get_teams_config_file(file_name: str):
    GITHUB_API_BASE = "https://github.kyndryl.net/api/v3"  # Or check your org's actual API base URL
    REPO_OWNER = "dwsagenticai"
    REPO_NAME = "magenticone"
    FILE_PATH = f"{file_name}"
    BRANCH = os.environ.get("AGENTICAI_GIT_BRANCH")
    pat = os.environ.get("GITHUB_PAT") 

    # API URL for the file
    url = f"{GITHUB_API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/contents/{FILE_PATH}?ref={BRANCH}"

    headers = {
        "Authorization": f"token {pat}",
        "Accept": "application/vnd.github.v3+json"
    }
    # It's better to store this in an environment variable for security

    if not pat:
        print("Error: GITHUB_PAT environment variable not set. Please set it to your Personal Access Token.")
    else:
        try:
            # Send a GET request with the authentication header
            response = requests.get(url, headers=headers)

            # Check if the request was successful (status code 200)
            if response.status_code == 200:
                content = response.json()
                decoded = base64.b64decode(content["content"]).decode("utf-8")
                try:
                    file_content = decoded
                    # print(f"✅ JSON loaded successfully:\n {file_content}")
                except Exception as e:
                    print("⚠️ Error parsing JSON:", e)
                    print(decoded)
                
                """ with open("car_rental_charges_dispute_agent_config.json", "w") as f:
                    f.write(file_content)
                print("\nFile saved to 'car_rental_charges_dispute_agent_config.json'") """
                return file_content
            else:
                print(f"Failed to fetch the file. Status code: {response.status_code}")
                print("Response content:", response.text)
                print("Please check your Personal Access Token and repository permissions.")

        except requests.exceptions.RequestException as e:
            print(f"An error occurred: {e}")

async def load_agents_from_config(teamType: str) -> list[AssistantAgent]:
    """Load agent configurations from JSON files."""

    team_config = await get_team_config(teamType)
    if not team_config:
        print(f"❌ No team config found for '{teamType}'")
        return []
    
    agents_config_file = team_config.get("config", "general_agents_config.json")  # Get correct config file
    agents_config_file = f"team_config/{agents_config_file}"
    print(f"📂 Loading agent config file: {agents_config_file}")
    if os.environ.get("TEST_ENV", "cloud") == 'local':
        if not os.path.exists(agents_config_file):
            print(f"❌ File '{agents_config_file}' does not exist!")
            return []
        
        try:
            async with aiofiles.open(agents_config_file, "r") as file:
                file_content = await file.read()
                print("📄 Loaded File Content (Preview):", file_content[:200])  # Print first 200 chars for debugging
                agents_data = json.loads(file_content)
        except json.JSONDecodeError as e:
            print(f"❌ JSON Error in '{agents_config_file}': {e}")
            return []
        except Exception as e:
            print(f"❌ Unexpected error while reading '{agents_config_file}': {e}")
            return []
    
    if os.environ.get("TEST_ENV", "cloud") == 'cloud':
        file_content = await get_teams_config_file(agents_config_file)
        try:
            agents_data = json.loads(file_content)
        except json.JSONDecodeError as e:
            print(f"❌ JSON Error in '{agents_config_file}': {e}")
            return []
        except Exception as e:
            print(f"❌ Unexpected error while reading '{agents_config_file}': {e}")
            return []

    # Local tools
    """ agents = []
        for agent in agents_data.get("agents", []):
            tool_functions = [globals().get(tool) for tool in agent.get("tools", []) if globals().get(tool)]
            
            if len(tool_functions) != len(agent.get("tools", [])):
                missing_tools = [tool for tool in agent.get("tools", []) if tool not in globals()]
                print(f"⚠️ Missing tools: {missing_tools} for agent {agent['name']}")

            agents.append(
                AssistantAgent(
                    name=agent["name"], 
                    model_client=model_client,
                    tools=tool_functions,
                    system_message=agent["system_message"],
                    description=agent["description"],
                    reflect_on_tool_use=True, 
                    model_client_stream=False,
                )
            ) """
    agents = []
    mcp_url = os.getenv('MCP_URL')
    params = SseServerParams(url=mcp_url)
    async with create_mcp_server_session(params) as session:
        await session.initialize()
        available_tools = await mcp_server_tools(server_params=params)
        tools_by_name = {tool.name: tool for tool in available_tools}

        for agent in agents_data.get("agents", []):
            tool_names = agent.get("tools", [])
            for name in tool_names:
                tool = tools_by_name.get(name)
                if tool is None:
                    print(f"[WARN] Tool not found for name: {name}")
                else:
                    print(f"\nTool: {name}")
                    """ print(dir(tool))
                    print(f"tools_by_name[name]._tool::{tools_by_name[name]._tool}") """
            agent_tools = [tools_by_name[name] for name in tool_names if name in tools_by_name]
            print(f"✅ Successfully loaded {agent_tools} tools")

            missing_tools = [name for name in tool_names if name not in tools_by_name]
            if missing_tools:
                print(f"⚠️ Missing tools: {missing_tools} for agent {agent['name']}")
            agent_names.append(agent["name"])
            agents.append(
                AssistantAgent(
                    name=agent["name"],
                    model_client=model_client,
                    tools=agent_tools,
                    system_message=f'{agent["system_message"]}. Please summarize the outcome in clear, plain language that anyone can understand.',
                    description=agent["description"],
                    reflect_on_tool_use=True,
                    model_client_stream=False
                )
            )
    
    print(f"✅ Successfully loaded {len(agents)} agents from '{agents_config_file}'")
    return agents


async def get_team(user_input_func: Callable[[str, Optional[CancellationToken]], Awaitable[str]], request: str) -> MagenticOneGroupChat:
    """Retrieve or create a team based on request type."""
    user_proxy = UserProxyAgent(name="User", description="Kyndryl User", input_func=user_input_func)
    team_config = await get_team_config(request)
    print("::::team_config", team_config)
    team_type = team_config.get("teamName", "GeneralAssistantSupervisor")
    print("::::team_type", team_type)
    agent_list = await load_agents_from_config(team_type)
    agent_list.append(user_proxy)
    """
    Args:
        participants (List[ChatAgent]): The participants in the group chat.
        model_client (ChatCompletionClient): The model client used for generating responses.
        termination_condition (TerminationCondition, optional): The termination condition for the group chat. Defaults to None.
            Without a termination condition, the group chat will run based on the orchestrator logic or until the maximum number of turns is reached.
        max_turns (int, optional): The maximum number of turns in the group chat before stopping. Defaults to 20.
        max_stalls (int, optional): The maximum number of stalls allowed before re-planning. Defaults to 3.
        final_answer_prompt (str, optional): The LLM prompt used to generate the final answer or response from the team's transcript. A default (sensible for GPT-4o class models) is provided.
    """
    team = MagenticOneGroupChat(
        agent_list, 
        model_client=model_client,
        termination_condition=combined_termination,
        max_turns=100,
        max_stalls=30
        )
    
    # Load state from file.
    """if not os.path.exists(state_path):
        return team
    async with aiofiles.open(state_path, "r") as file:
        state = json.loads(await file.read())
    await team.load_state(state)"""
    return team

async def get_history() -> list[dict[str, Any]]:
    """Get chat history from file."""
    if not os.path.exists(history_path):
        return []
    async with aiofiles.open(history_path, "r") as file:
        return json.loads(await file.read())

@app.get("/history")
async def history() -> list[dict[str, Any]]:
    """Fetch chat history from storage."""
    try:
        # return await get_history()
        return []
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e)) from e

# response_prompt = ""
response_prompt="Summarize the whole process. Evaluate the customer sentiment. Respond with 'TERMINATE' if the workflow is complete."

@app.get("/get_client_id")
async def get_client_id(request: Request):
    return {"client_id": shared_state["client_id"]}

@app.get("/get_agent_list")
async def get_agent_list(request: Request):
    return {"agent_list": agent_names}

# Assuming these are defined globally or passed in:
# model_client, combined_termination, input_manager, response_prompt, TEAM_CONFIG

@app.post("/start_chat")
async def start_chat(request: Request):
    """
    Initiates an AI chat workflow via HTTP.
    This endpoint will long-poll if human approval is required.
    """
    try:
        llm_usage = LLMUsageTracker()
        logger.handlers = [llm_usage]
        body = await request.json()
        print(f"🔥 Received request: {body}")
        user_message = body.get("query", "")
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        client_id = body.get("client_id", timestamp)
        shared_state["client_id"] = client_id

        # Define the UserProxyAgent's input function for this specific chat
        async def http_user_input(prompt: str, cancellation_token: CancellationToken | None) -> str:
            print(f"[Agent Flow] UserProxyAgent asked for input for client '{client_id}'. Prompt: '{prompt}'")
            print(f"*** HUMAN INPUT REQUIRED for client_id: {client_id} ***")
            print(f"Prompt from UserProxyAgent: {prompt}")
            return await input_manager.get_user_input(client_id)

        # Create the MAIN UserProxyAgent that will interact with the SelectorGroupChat
        main_user_proxy = UserProxyAgent(
            name="User",
            description="Kyndryl User for top-level interaction",
            input_func=http_user_input # This is the critical link
        )

        # Initialize inner teams as SocietyOfMindAgents
        agents = {}
        TEAM_CONFIG = await load_team_config()
        source_to_team_map = {}  # move this outside the loop if you want a complete map
        for team_config_item in TEAM_CONFIG:
            if isinstance(team_config_item, dict) and "teamName" in team_config_item:
                team_name = team_config_item["teamName"]
                # Get the inner team. The user_input_func for INNER teams could be different
                # or just a dummy one if you don't expect inner teams to ask for direct human input.
                # For simplicity, we'll give them the same http_user_input for now,
                # but in complex flows, you might want separate queues for inner teams.
                inner_team_instance = await get_team(http_user_input, team_name)
                
                agent_description = team_config_item.get("description", "An agent that uses an inner team of agents to generate responses.")
                
                # Map inner agents to the top-level team name
                for agent in inner_team_instance._participants:
                    if hasattr(agent, 'name'):
                        source_to_team_map[agent.name] = team_name

                society_agent = SocietyOfMindAgent(
                    team_name,
                    team=inner_team_instance,
                    model_client=model_client,
                    description=agent_description,
                    response_prompt=response_prompt
                )

                agents[team_name] = society_agent


        selector_prompt = (
            "Select the most suitable agent to perform the next task. {roles}\n\n"
            "Current conversation context: {history}\n\n"
            "- Prioritize expertise.\n"
            "- Planner agent assigns tasks before execution.\n"
            "- Weather related queries: GeneralAssistantSupervisor.\n"
            "- Server issues: ServerRecoverySupervisor.\n"
            "- Incident Mitigation: IncidentMitigationSupervisor.\n"
            "- Pump issues: FuelStationSupervisor.\n"
            "- Car rental no show issues: CarRentalNoShowSupervisor.\n"
            "- Coordinate across teams when necessary.\n"
            "- Select only one agent at a time."
        )
        
        # Create the Orchestrator (SelectorGroupChat)
        # IMPORTANT: Include the main_user_proxy in the list of participants for the orchestrator
        orchestrator_participants = list(agents.values())
        orchestrator_participants.append(main_user_proxy) # Add the user proxy here!

        orchestrator = SelectorGroupChat(
            orchestrator_participants, # Use the list that includes main_user_proxy
            model_client=model_client,
            termination_condition=combined_termination,
            selector_prompt=selector_prompt,
            allow_repeated_speaker=False
        )

        task = TextMessage(content=user_message, sender="User", source="Incident Watcher")
        history = await get_history()

        # File path where you want to stream output
        output_json_file = f"console_streams/console_stream.jsonl"
        output_txt_file = f"console_streams/console_stream_{shared_state['client_id']}.txt"
        os.makedirs("console_streams", exist_ok=True)
        # Create the file if it doesn't exist
        if not os.path.exists(output_txt_file):
            with open(output_txt_file, "w"): pass  # Just create an empty file
        # Clear the file before writing
        with open(output_txt_file, "w") as f:
            pass  # This truncates the file

        activation_sent = False
        async for message in orchestrator.run_stream(task=task):
            if isinstance(message, TaskResult):
                continue
            
            if not activation_sent:
                agent_names = re.findall(r"\n([A-Za-z0-9_]+Agent):", message.content)
                if len(agent_names) != 0:
                    # 2. Count agent name frequency per team
                    team_agent_counts = {}
                    for agent_name in agent_names:
                        team_name = source_to_team_map.get(agent_name)
                        if team_name:
                            team_agent_counts.setdefault(team_name, []).append(agent_name)

                    # 3. Pick the team that has *all* the agent names listed
                    for team_name, agents_in_team in team_agent_counts.items():
                        if len(agents_in_team) == len(agent_names):
                            activation_msg = {
                                "source": "team_name",
                                "content": team_name
                            }
                            print(f"🚀 Activating team: {team_name}")
                            await manager.broadcast(activation_msg)
                            activation_sent = True
                            break

            # Send to console
            # await manager.send_admin_message(message.model_dump())
            await manager.broadcast(message.model_dump())


            # Save to local file in JSONL format
            """ async with aiofiles.open(output_file, "a") as f:
                await f.write(json.dumps(message.model_dump()) + "\n") """

            async with aiofiles.open(output_txt_file, "a") as f:
                if isinstance(message.content, str):
                    content = message.content.strip()
                    if not (content.startswith("[FunctionCall(") or content.startswith("[FunctionExecutionResult(")):
                        formatted_text = f'{"Kyndryl Orchestrator" if message.source == "MagenticOneOrchestrator" else message.source}: {content}'
                        await f.write(formatted_text + "\n\n" + "-------------------------------------------------------------------------------\n\n")

            if not isinstance(message, UserInputRequestedEvent):
                history.append(message.model_dump())
        """
        # Save team state to file.
        async with aiofiles.open(state_path, "w") as file:
            state = await team.save_state()
            await file.write(json.dumps(state))

        # Save chat history to file.
        async with aiofiles.open(history_path, "w") as file:
            await file.write(json.dumps(history))
        """  
        print("******************************Task Completed**********************************")
        print("llm usage....")
        print(f"prompt tokens: {llm_usage.prompt_tokens}")
        print(f"completion tokens: {llm_usage.completion_tokens}")
        metrics_message = {"source":"Flow Metrics", "content":""}
        metrics_message["content"] = f"Prompt tokens: {llm_usage.prompt_tokens}\n\nCompletion tokens: {llm_usage.completion_tokens}"
        await manager.broadcast(metrics_message)
    except Exception as e:
        print(f"❌ Error in start_chat for client_id: {client_id} \n{e}")
        raise HTTPException(status_code=500, detail=str(e))

# --- HTTP Endpoint for Receiving Approval ---
class ApprovalRequest(BaseModel):
    client_id: str
    response: str # Expected: "Approve" or "Reject"
    approval_token: Optional[str] = None # For real-world token validation

@app.post("/receive_approval")
async def receive_approval(approval_data: ApprovalRequest):
    """ Endpoint for a human to send their approval/rejection back to the system.
    This will unblock the waiting /start_chat request.
    It does NOT send any notification; it just handles the incoming response. """
    client_id = approval_data.client_id
    response_message = approval_data.response

    print(f"✅ Received approval callback for client '{client_id}' with response: '{response_message}'")

    # In a production system, you would validate approval_token here
    # to ensure the request is legitimate and corresponds to an active task.

    if response_message.lower() in ["approve", "reject"]:
        await input_manager.put_user_input(client_id, response_message)
        return JSONResponse(status_code=200, content={"message": f"Response '{response_message}' recorded for client '{client_id}'. AI flow will continue."})
    else:
        print(f"⚠️ Invalid response '{response_message}' for client '{client_id}'. Expected 'Approve' or 'Reject'.")
        raise HTTPException(status_code=400, detail="Invalid response. Expected 'Approve' or 'Reject'.")

@app.get("/simulated_approval", response_class=HTMLResponse)
async def simulated_approval(client_id: str, response: str):
    """ Endpoint for a human to send their approval/rejection back to the system.
    This will unblock the waiting /start_chat request.
    It does NOT send any notification; it just handles the incoming response. """

    print(f"✅ Received approval callback for client '{client_id}' with response: '{response}'")

    # In a production system, you would validate approval_token here
    # to ensure the request is legitimate and corresponds to an active task.

    if response.lower() in ["approve", "reject"]:
        if response.lower() == "reject":
            response = "I Reject the proposed resolution, please terminate the flow"
        # await asyncio.sleep(3) # time for demonstrator to switch back to the console
        await input_manager.put_user_input(client_id, response)
        return JSONResponse(status_code=200, content={"message": f"Response '{response}' recorded for client '{client_id}'. AI flow will continue."})
    else:
        print(f"⚠️ Invalid response '{response}' for client '{client_id}'. Expected 'Approve' or 'Reject'.")
        raise HTTPException(status_code=400, detail="Invalid response. Expected 'Approve' or 'Reject'.")

@app.websocket("/ws/chat/{client_id}")
async def chat(websocket: WebSocket, client_id: str):
    """WebSocket handler for real-time chat."""
    global model_client
    await manager.connect(websocket, client_id)
    shared_state["client_id"] = client_id  # Ensure consistent client ID tracking

    # Track LLM usage
    llm_usage = LLMUsageTracker()
    logger.handlers = [llm_usage]

    # User input function used by the team
    async def _user_input(prompt: str, cancellation_token: CancellationToken | None) -> str:
        print(f"[Agent Flow] UserProxyAgent asked for input for client '{client_id}'. Prompt: '{prompt}'")
        print(f"*** HUMAN INPUT REQUIRED for client_id: {client_id} ***")
        print(f"Prompt from UserProxyAgent: {prompt}")
        data = await websocket.receive_json()
        return TextMessage.model_validate(data).content

    try:
        while True:
            try:
                # Receive user input and wrap as task
                data = await websocket.receive_json()
                user_input = TextMessage.model_validate(data)
                task = TextMessage(content=user_input.content, sender="User", source="Incident Watcher")

                # Load team configuration and agents
                TEAM_CONFIG = await load_team_config()
                print("TEAM_CONFIG:", TEAM_CONFIG)

                teams = {}
                for team in TEAM_CONFIG:
                    if isinstance(team, dict) and "teamName" in team:
                        teams[team["teamName"]] = await get_team(_user_input, team["teamName"])

                agents = {}
                for team_name, team_instance in teams.items():
                    if isinstance(team_instance, MagenticOneGroupChat):
                        description = next((cfg["description"] for cfg in TEAM_CONFIG if cfg.get("teamName") == team_name), "An agent that uses an inner team of agents to generate responses.")
                        agents[team_name] = SocietyOfMindAgent(
                            team_name,
                            team=team_instance,
                            model_client=model_client,
                            description=description,
                            response_prompt=response_prompt
                        )

                # Add UserProxyAgent
                main_user_proxy = UserProxyAgent(
                    name="User",
                    description="Kyndryl User for top-level interaction",
                    input_func=_user_input
                )

                # Create orchestrator with all participants
                selector_prompt = (
                    "Select the most suitable agent to perform the next task. {roles}\n\n"
                    "Current conversation context: {history}\n\n"
                    "- Prioritize expertise.\n"
                    "- Planner agent assigns tasks before execution.\n"
                    "- Weather related queries: GeneralAssistantSupervisor.\n"
                    "- Server issues: ServerRecoverySupervisor.\n"
                    "- Incident Mitigation: IncidentMitigationSupervisor.\n"
                    "- Pump issues: FuelStationSupervisor.\n"
                    "- Car rental no show issues: CarRentalNoShowSupervisor.\n"
                    "- Coordinate across teams when necessary.\n"
                    "- Select only one agent at a time."
                )

                orchestrator_participants = list(agents.values())
                orchestrator_participants.append(main_user_proxy)

                orchestrator = SelectorGroupChat(
                    orchestrator_participants,
                    model_client=model_client,
                    termination_condition=combined_termination,
                    selector_prompt=selector_prompt,
                    allow_repeated_speaker=False
                )

                history = await get_history()

                # File setup
                output_file = f"console_streams/console_stream.jsonl"
                os.makedirs("console_streams", exist_ok=True)
                output_txt_file = f"console_streams/console_stream_{client_id}.txt"
                with open(output_txt_file, "w"): pass  # Clear or create

                # Run orchestrator and stream output
                async for message in orchestrator.run_stream(task=task):
                    if isinstance(message, TaskResult):
                        continue

                    await manager.broadcast(message.model_dump())

                    # Save to local file in JSONL format
                    """ async with aiofiles.open(output_file, "a") as f:
                        await f.write(json.dumps(message.model_dump()) + "\n") """

                    async with aiofiles.open(output_txt_file, "a") as f:
                        if isinstance(message.content, str):
                            content = message.content.strip()
                            if not (content.startswith("[FunctionCall(") or content.startswith("[FunctionExecutionResult(")):
                                formatted_text = f'{"Kyndryl Orchestrator" if message.source == "MagenticOneOrchestrator" else message.source}: {content}'
                                await f.write(formatted_text + "\n\n" + "-" * 79 + "\n\n")

                    if not isinstance(message, UserInputRequestedEvent):
                        history.append(message.model_dump())

                """
                # Save team state to file.
                async with aiofiles.open(state_path, "w") as file:
                    state = await team.save_state()
                    await file.write(json.dumps(state))

                # Save chat history to file.
                async with aiofiles.open(history_path, "w") as file:
                    await file.write(json.dumps(history))
                """  
                # Final LLM token usage
                print("******************************Task Completed**********************************")
                print(f"prompt tokens: {llm_usage.prompt_tokens}")
                print(f"completion tokens: {llm_usage.completion_tokens}")
                metrics_message = {
                    "source": "Flow Metrics",
                    "content": f"Prompt tokens: {llm_usage.prompt_tokens}\n\nCompletion tokens: {llm_usage.completion_tokens}"
                }
                await manager.broadcast(metrics_message)

            except Exception as e:
                print(f"❌ Unexpected error: {str(e)}")
                print(f"❌ Unexpected error: {e}")
                await websocket.send_json({"type": "error", "content": f"Unexpected error: {str(e)}", "source": "system"})

    except WebSocketDisconnect:
        print("Client disconnected")

    except Exception as e:
        print(f"❌ Fatal error in WebSocket chat: {str(e)}")
        await websocket.send_json({"type": "error", "content": f"Fatal error: {str(e)}", "source": "system"})


# Example usage
if __name__ == "__main__":
    import uvicorn
    port = int(os.getenv('AGENTICAI_PORT', 8002))
    # uvicorn.run(app, host="0.0.0.0", port=8002, timeout_keep_alive=1200.0, ws_ping_interval=3000.0, ws_ping_timeout=3000.0)
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=port,
        log_level="debug",
        timeout_keep_alive=600,
        ws_ping_interval=None, # No WebSocket pings for HTTP-only
        ws_ping_timeout=None   # No WebSocket pings for HTTP-only
    )

Expected behavior
System should be conversational.

Screenshots
If applicable, add screenshots to help explain your problem.

Additional context
Add any other context about the problem here.

[karun19]

Can you please provide a sample of SOP driven implementation.

[ekzhu]

What is SOP?

Have you seen GraphFlow? https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/graph-flow.html