On-boarding¶

import sys
!{sys.executable} -m pip install -q ag2==0.3.2 ag2[anthropic]==0.3.2

# pylint: disable=line-too-long,unknown-option-value,unused-argument,unused-import,invalid-name,import-error,inconsistent-quotes,missing-function-docstring,missing-param-doc,missing-return-doc
import csv
import os
import sqlite3

from typing import Any, Callable, Dict, List, Optional, Tuple, Union  # noqa
from typing_extensions import Annotated

from autogen import Agent, AssistantAgent, ChatResult, ConversableAgent, UserProxyAgent, initiate_chats, runtime_logging

from waldiez_api_keys import get_model_api_key

gpt_3_5_turbo_llm_config = {
    "config_list": [
        {
            "model": "gpt-3.5-turbo",
            "temperature": 0.5,
            "api_type": "openai",
            "api_key": get_model_api_key("gpt_3_5_turbo")
        }
    ]
}
claude_3_5_sonnet_20240620_llm_config = {
    "config_list": [
        {
            "model": "claude-3-5-sonnet-20240620",
            "temperature": 0.5,
            "api_type": "anthropic",
            "api_key": get_model_api_key("claude_3_5_sonnet_20240620")
        }
    ]
}

customer_proxy = UserProxyAgent(
    name="customer_proxy",
    description="A user proxy agent",
    llm_config=False,
    human_input_mode="ALWAYS",
    max_consecutive_auto_reply=None,
    default_auto_reply=None,
    code_execution_config=False,
    is_termination_msg=None,
)
personal_information_agent = AssistantAgent(
    name="personal_information_agent",
    description="On-boarding Personal information agent",
    llm_config=claude_3_5_sonnet_20240620_llm_config,
    system_message="You are a helpful customer on-boarding agent, you are here to help new customers get started with our product. Your job is to gather customer's name and location. Do not ask for other information. Return 'TERMINATE' when you have gathered all the information.\n",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=None,
    default_auto_reply=None,
    code_execution_config=False,
    is_termination_msg=lambda x: any(x.get("content", "") and x.get("content", "").endswith(keyword) for keyword in ["TERMINATE"]),
)
topic_preference_agent = AssistantAgent(
    name="topic_preference_agent",
    description="On-boarding Topic Preference Agent",
    llm_config=claude_3_5_sonnet_20240620_llm_config,
    system_message="You are a helpful customer topic preference agent, you are here to help new customers get started with our product. Your job is to gather customer's topic of interest. Do not ask for other information. Return 'TERMINATE' when you have gathered all the information.\n",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=None,
    default_auto_reply=None,
    code_execution_config=False,
    is_termination_msg=lambda x: any(x.get("content", "") and x.get("content", "").endswith(keyword) for keyword in ["TERMINATE"]),
)
customer_engagement_agent = AssistantAgent(
    name="customer_engagement_agent",
    description="On-boarding Customer Engagement Agent",
    llm_config=gpt_3_5_turbo_llm_config,
    system_message="You are a helpful customer service agent here to provide fun for the customer based on the user's personal information and topic preferences. This could include fun facts, jokes, or interesting stories. Make sure to make it engaging and fun! Return 'TERMINATE' when you are done.",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=None,
    default_auto_reply=None,
    code_execution_config=False,
    is_termination_msg=lambda x: any(x.get("content", "") and x.get("content", "").endswith(keyword) for keyword in ["TERMINATE"]),
)

def sqlite_to_csv(dbname: str, table: str, csv_file: str) -> None:
    """Convert a sqlite table to a csv file.

    Parameters
    ----------
    dbname : str
        The sqlite database name.
    table : str
        The table name.
    csv_file : str
        The csv file name.
    """
    conn = sqlite3.connect(dbname)
    query = f"SELECT * FROM {table}"  # nosec
    try:
        cursor = conn.execute(query)
    except sqlite3.OperationalError:
        conn.close()
        return
    rows = cursor.fetchall()
    column_names = [description[0] for description in cursor.description]
    data = [dict(zip(column_names, row)) for row in rows]
    conn.close()
    with open(csv_file, "w", newline="", encoding="utf-8") as file:
        _csv_writer = csv.DictWriter(file, fieldnames=column_names)
        _csv_writer.writeheader()
        _csv_writer.writerows(data)

runtime_logging.start(
    logger_type="sqlite",
    config={"dbname": "flow.db"},
)

initiate_chats([
    {
        "sender": personal_information_agent,
        "recipient": customer_proxy,
        "summary_method": "reflection_with_llm",
        "summary_args": {
            "summary_prompt": "Return the customer information into as JSON object only: {'name': '', 'location': ''}",
            "summary_role": "user"
        },
        "max_turns": 1,
        "clear_history": True,
        "message": "Hello, I'm here to help you get started with our product. Could you tell me your name and location?",
    },
    {
        "sender": topic_preference_agent,
        "recipient": customer_proxy,
        "summary_method": "reflection_with_llm",
        "summary_args": {
            "summary_prompt": "Return the customer information into as JSON object only: {'topic_of_interest': ''}",
            "summary_role": "user"
        },
        "max_turns": 1,
        "clear_history": True,
        "message": "Great! Could you tell me what topics you are interested in reading about?",
    },
    {
        "sender": customer_proxy,
        "recipient": customer_engagement_agent,
        "summary_method": "reflection_with_llm",
        "summary_args": {
            "summary_role": "user"
        },
        "max_turns": 1,
        "clear_history": True,
        "message": "Let's find something fun to read.",
    },
])
runtime_logging.stop()

if not os.path.exists("logs"):
    os.makedirs("logs")
for table in [
    "chat_completions",
    "agents",
    "oai_wrappers",
    "oai_clients",
    "version",
    "events",
    "function_calls",
]:
    dest = os.path.join("logs", f"{table}.csv")
    sqlite_to_csv("flow.db", table, dest)