SelectorGroupChat is not hitting run running LLM

[sunilsvsnlr]

Based on user input, it should call the QueryProcessingAgent and prepare the prompt and then send to LLM. Once LLM generated python code it should call CodeExecutionAgent to run python code locally by reading local file data to show result. If any of the error while executing code it should call QueryProcessingAgent to correct the code. Its not hitting run or on_messages in the agent.

Can any one help please?

Below is my code

import asyncio
import json
import os
import pandas as pd
import logging
from datetime import datetime, timedelta
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import TaskResult
from autogen_agentchat.messages import ChatMessage
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_ext.tools.code_execution import PythonCodeExecutionTool
from typing import Any, AsyncGenerator, List, Mapping, Sequence

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

openai_client = OpenAIChatCompletionClient(
    model="lmstudio-community/DeepSeek-R1-Distill-Qwen-7B-GGUF/DeepSeek-R1-Distill-Qwen-7B-Q4_K_M.gguf",
    seed=42,
    api_key="lm-studio",
    base_url="http://127.0.0.1:1234/v1",
    max_retries=3,
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": False,
        "family": "unknown",
    },
)

json_folder_path = "C:/projects/genai"
metadata_file = "C:/projects/genai/metadata/metadata.json"
merged_data_file = "C:/projects/genai/merged/merged_data.json"  # Merged data file for analysis


def generate_metadata(folder_path):
    os.makedirs(os.path.dirname(metadata_file), exist_ok=True)  # Ensure directory exists

    for file_name in os.listdir(folder_path):
        if file_name.endswith(".json"):
            file_path = os.path.join(folder_path, file_name)
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
                if "items" in data and isinstance(data["items"], list) and len(data["items"]) > 0:
                    sample_record = data["items"][0]  # Take the first record as a sample
                    metadata = {"columns": list(sample_record.keys())}
                    with open(metadata_file, "w", encoding="utf-8") as meta_file:
                        json.dump(metadata, meta_file, indent=2)
                    print(f"Metadata generated from {file_name} and saved to {metadata_file}")
                    return metadata
    print("No valid JSON files found for metadata generation.")
    return None


def load_all_json_data(folder_path):
    os.makedirs(os.path.dirname(merged_data_file), exist_ok=True)  # Ensure directory exists

    cutoff_date = pd.Timestamp.utcnow() - pd.Timedelta(days=5 * 365)  # Convert to Pandas Timestamp in UTC
    all_data = []
    for file_name in os.listdir(folder_path):
        if file_name.endswith(".json"):
            file_path = os.path.join(folder_path, file_name)
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
                if "items" in data:  # Ensure JSON contains 'items' key
                    all_data.extend(data["items"])

    df = pd.DataFrame(all_data)

    if df.empty:
        logging.warning("No data found in JSON files.")
        return df

    date_columns = [col for col in df.columns if "date" in col.lower()]
    print(f"Identified date columns: {date_columns}")
    logging.info(f"Identified date columns: {date_columns}")

    for col in date_columns:
        df[col] = pd.to_datetime(df[col], errors="coerce", utc=True)  # Convert invalid dates to NaT

    df.to_json(merged_data_file, orient="records", indent=2)
    logging.info(f"Loaded {len(df)} records.")
    return df

class QueryProcessingAgent(AssistantAgent):
    def __init__(self):
        super().__init__(
            name="QueryProcessingAgent",
            system_message="You generate Python code to analyze project data based on user questions. "
                           "If execution fails, refine the code and retry.",
            model_client=openai_client,
    )

    async def on_messages(self, messages, cancellation_token):
        logging.info(f"QueryProcessingAgent received task: {messages[-1].content}")
        task = messages[-1].content  # Extract the latest user question
        metadata = json.load(open(metadata_file))
        prompt = f"""
        Generate a Python script to analyze project data based on the following metadata:
        {json.dumps(metadata, indent=2)}

        The script should:
        - Load data from the JSON file: "{merged_data_file}".
        - Answer the following question: "{task}"
        - Ensure proper date handling (convert date columns to datetime before calculations).
        - Strictly use columns specified in meta data only for processing.
        - Don't use any depreciated libraries.
        - Use pandas for data analysis.
        - Return a structured result.
        
        Ensure the script is properly formatted and outputs relevant insights.
        If execution fails, refine the code and try again.
        """
        response = await super().on_messages([ChatMessage(content=prompt, source="user")], cancellation_token)
        # response = await super().run(task=prompt)
        if not response:
            logging.error("QueryProcessingAgent failed to generate code.")
            return TaskResult(messages=["QueryProcessingAgent: Failed to generate code. Trying again..."])
        logging.info("QueryProcessingAgent successfully generated code.")
        return response


class CodeExecutionAgent(AssistantAgent):
    def __init__(self):
        super().__init__(
            name="CodeExecutionAgent",
            system_message="You execute Python code locally. If execution fails, return a clear failure message "
                           "so that the QueryProcessingAgent can refine the code.",
            model_client=openai_client,
        )
        self.executor = PythonCodeExecutionTool(
            executor=LocalCommandLineCodeExecutor(work_dir="coding")
        )

    async def on_messages(self, messages, cancellation_token):
        logging.info("CodeExecutionAgent received task to execute.")
        task = messages[-1].content
        if not task:
            logging.error("No valid Python code received.")
            return TaskResult(messages=["QueryProcessingAgent"])  #Redirect to QueryProcessingAgent for new code

        generated_code = None
        for message in messages:
            if "```python" in message.content:
                generated_code = message.content.split("```python")[1].split("```")[0].strip()
                break

        if not generated_code:
            logging.error("No valid Python code found.")
            return TaskResult(messages=["QueryProcessingAgent"])

        cancellation_token = CancellationToken()
        try:
            result = await self.executor.run_json({"code": generated_code}, cancellation_token)
            logging.info("CodeExecutionAgent executed successfully.")
            return TaskResult(messages=[
                ChatMessage(content=f"ExecutionAgent: Successfully executed. Result: {result}")])  # Fixed!
            # return result
        except Exception as e:
            logging.error(f"Execution failed: {e}")
            return TaskResult(messages=[ChatMessage(content="QueryProcessingAgent")])
            # return TaskResult(messages=["QueryProcessingAgent"])



async def main():
    metadata = generate_metadata(json_folder_path)
    if not metadata:
        logging.error("Metadata generation failed. Exiting...")
        print("Metadata generation failed. Exiting...")
        return

    data = load_all_json_data(json_folder_path)  # Load merged JSON data

    if data.empty:
        logging.error("No data found in the given folder.")
        print("No data found in the given folder.")
        return

    logging.info(f"Loaded {len(data)} records from JSON files. Data saved to {merged_data_file}")
    print(f"Loaded {len(data)} records from JSON files. Data saved to {merged_data_file}")

    while True:
        user_question = input("\nEnter your question about the project data (or type 'exit' to quit): ").strip()
        if user_question.lower() == "exit":
            logging.info("Exiting...")
            print("Exiting...")
            break

        query_agent = QueryProcessingAgent()
        execution_agent = CodeExecutionAgent()

        team = SelectorGroupChat([query_agent, execution_agent],
                                 model_client=openai_client,
                                 termination_condition=MaxMessageTermination(5))

        logging.info("\nAgents are working together...\n")
        print("\nAgents are working together...\n")

        stream = team.run_stream(task=f"{user_question}")

        async for message in stream:
            logging.info(f"{message}")
            print(message)

        await team.reset()

asyncio.run(main())

[ekzhu]

1. AssistantAgent is not meant to be subclassed. Please create custom agents and implement BaseChatAgent base class instead. The group chat uses on_messages_stream not on_messages.

2. Rather than separate code executor agent can you try just using PythonCodeExecutionTool for an agent as a tool? Otherwise, use builtin CodeExecutoAgent directly.

3. Agent should return autogen_agentchat.base.Response, not TaskResult.

[sunilsvsnlr]

Hi @ekzhu
when i use BaseChatAgent its not hitting model to generate python code its directly passing to code executor. Below is the modified code. Could you please check this

import asyncio
import json
import os
import pandas as pd
import logging
from datetime import datetime, timedelta

from autogen_agentchat.agents import AssistantAgent, CodeExecutorAgent, UserProxyAgent
from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import TextMessage, ChatMessage, AgentEvent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_ext.tools.code_execution import PythonCodeExecutionTool
from typing import AsyncGenerator, Sequence

async def main() -> None:
# Folder containing JSON files (Modify as needed)
json_folder_path = "C:/projects/genai"
metadata_file = "C:/projects/genai/metadata/metadata.json"
merged_data_file = "C:/projects/genai/merged/merged_data.json" # Merged data file for analysis

# Function to generate metadata from one sample file
def generate_metadata(folder_path):
    os.makedirs(os.path.dirname(metadata_file), exist_ok=True)  # Ensure directory exists

    for file_name in os.listdir(folder_path):
        if file_name.endswith(".json"):
            file_path = os.path.join(folder_path, file_name)
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
                if "items" in data and isinstance(data["items"], list) and len(data["items"]) > 0:
                    sample_record = data["items"][0]  # Take the first record as a sample
                    metadata = {"columns": list(sample_record.keys())}
                    with open(metadata_file, "w", encoding="utf-8") as meta_file:
                        json.dump(metadata, meta_file, indent=2)
                    print(f"Metadata generated from {file_name} and saved to {metadata_file}")
                    return metadata
    print("No valid JSON files found for metadata generation.")
    return None

# Function to load and merge all JSON files in the folder
def load_all_json_data(folder_path):
    os.makedirs(os.path.dirname(merged_data_file), exist_ok=True)  # Ensure directory exists

    cutoff_date = pd.Timestamp.utcnow() - pd.Timedelta(days=5 * 365)  # Convert to Pandas Timestamp in UTC
    all_data = []
    for file_name in os.listdir(folder_path):
        if file_name.endswith(".json"):
            file_path = os.path.join(folder_path, file_name)
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
                if "items" in data:  # Ensure JSON contains 'items' key
                    all_data.extend(data["items"])

    # Convert to DataFrame and save as JSON
    df = pd.DataFrame(all_data)

    if df.empty:
        return df

    # Check which columns contain dates
    date_columns = [col for col in df.columns if "date" in col.lower()]
    print(f"📅 Identified date columns: {date_columns}")

    # Convert date columns to datetime format
    for col in date_columns:
        df[col] = pd.to_datetime(df[col], errors="coerce", utc=True)  # Convert invalid dates to NaT

    # Save filtered data
    # df = df[df["createDate"] >= cutoff_date]
    df.to_json(merged_data_file, orient="records", indent=2)
    return df


# Initialize OpenAI client
model_client = OpenAIChatCompletionClient(
    model="lmstudio-community/DeepSeek-R1-Distill-Qwen-7B-GGUF/DeepSeek-R1-Distill-Qwen-7B-Q4_K_M.gguf",
    seed=42,
    api_key="lm-studio",
    base_url="http://127.0.0.1:1234/v1",
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": False,
        "family": "unknown",
    },
)

class HelpfullAgent(BaseChatAgent):
    def __init__(self):
        super().__init__(name="QueryProcessingAgent",
                         description="You generate Python code to analyze project data based on user questions. "
                       "If execution fails, refine the code and retry.")

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        task = messages[-1].content  # Extract the latest user question
        metadata = json.load(open(metadata_file))
        messages[-1].content = f"""
                Generate a Python script to analyze project data based on the following metadata:
                {json.dumps(metadata, indent=2)}

                The script should:
                - Load data from the JSON file: "{merged_data_file}".
                - Answer the following question: "{task}"
                - Ensure proper date handling (convert date columns to datetime before calculations).
                - Strictly use columns specified in meta data only for processing.
                - Don't use any depreciated libraries.
                - Use pandas for data analysis.
                - Return a structured result.

                Ensure the script is properly formatted and outputs relevant insights.
                If execution fails, refine the code and try again.
                """
        return Response(chat_message=messages[-1])

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass

    @property
    def produced_message_types(self) -> Sequence[type[ChatMessage]]:
        return [ChatMessage]

helpfulAgent = HelpfullAgent()

assistant = AssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant. Write all code in python. Reply only 'TERMINATE' if the task is done.",
    model_client=model_client,
)

code_executor = CodeExecutorAgent(
    name="code_executor",
    code_executor=LocalCommandLineCodeExecutor(work_dir="coding"),
)

user_proxy = UserProxyAgent("user_proxy", input_func=input)

# The termination condition is a combination of text termination and max message termination, either of which will cause the chat to terminate.
termination = TextMentionTermination("TERMINATE")

# The group chat will alternate between the assistant and the code executor.
group_chat = RoundRobinGroupChat([user_proxy, helpfulAgent, assistant, code_executor], termination_condition=termination)

# `run_stream` returns an async generator to stream the intermediate messages.
stream = group_chat.run_stream(task="Write a python script")
# `Console` is a simple UI to display the stream.
await Console(stream)

asyncio.run(main())

[ekzhu]

Can you post the output from console? It seems odd to me because you are using RoundRobinGroupChat, it shouldn't skip any agent.