GPT Assistants API Parallel Function Calling

export OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxx
export GNEWS_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxxxx

import openai
import time
import json
from yfinance import Ticker
import requests, os

def get_stock_price(symbol: str) -> float:
stock = Ticker(symbol)
price = stock.history(period="1d")['Close'].iloc[-1]
return price

GNEWS_API_KEY = os.getenv('GNEWS_API_KEY')

def get_latest_company_news(company_name):
url = f"https://gnews.io/api/v4/search?q={company_name}&token={GNEWS_API_KEY}&lang=en&sortBy=publishedAt"
response = requests.get(url)
if response.status_code == 200:
return response.json()['articles']
else:
return []

def usd_to_gbp(usd_amount):
url = "https://api.exchangerate-api.com/v4/latest/USD"
try:
response = requests.get(url)
data = response.json()
gbp_rate = data['rates']['GBP']
return usd_amount * gbp_rate
except Exception as e:
return f"Error: {e}"

tools_list = [
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "Retrieve the latest closing price of a stock using its ticker symbol",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "The ticker symbol of the stock"
}
},
"required": ["symbol"]
}
}
},
{
"type": "function",
"function": {
"name": "get_latest_company_news",
"description": "Fetches the latest news articles related to a specified company",
"parameters": {
"type": "object",
"properties": {
"company_name": {
"type": "string",
"description": "The name of the company"
}
},
"required": ["company_name"]
}
}
},
{
"type": "function",
"function": {
"name": "usd_to_gbp",
"description": "Converts an amount in USD to GBP using the current exchange rate",
"parameters": {
"type": "object",
"properties": {
"usd_amount": {
"type": "number",
"description": "The amount in USD to be converted"
}
},
"required": ["usd_amount"]
}
}
}
]

Initialize the client

client = openai.OpenAI()

Step 1: Create an Assistant

assistant = client.beta.assistants.create(
name="Data Analyst Assistant",
instructions="You are a personal Data Analyst Assistant",
model="gpt-4-1106-preview",
tools=tools_list,
)

Step 2: Create a Thread

thread = client.beta.threads.create()

Step 3: Add a Message to a Thread

message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="""Can you please provide me stock price,
stock price in GBP,
and the latest company news of Apple?
"""
)

Step 4: Run the Assistant

run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
instructions="Please address the user as Mervin Praison."
)

print(run.model_dump_json(indent=4))

Define a dispatch table

function_dispatch_table = {
"get_stock_price": get_stock_price,
"get_latest_company_news": get_latest_company_news,
"usd_to_gbp": usd_to_gbp
}

while True:
# Wait for 5 seconds
time.sleep(5)

# Retrieve the run status
run_status = client.beta.threads.runs.retrieve(
    thread_id=thread.id,
    run_id=run.id
)
print(run_status.model_dump_json(indent=4))

# If run is completed, get messages
if run_status.status == 'completed':
    messages = client.beta.threads.messages.list(
        thread_id=thread.id
    )

    # Loop through messages and print content based on role
    for msg in messages.data:
        role = msg.role
        content = msg.content[0].text.value
        print(f"{role.capitalize()}: {content}")
    break
elif run_status.status == 'requires_action':
    print("Requires action")
    required_actions = run_status.required_action.submit_tool_outputs.model_dump()
    print(required_actions)
    tools_output = []

    for action in required_actions["tool_calls"]:
        func_name = action["function"]["name"]
        arguments = json.loads(action["function"]["arguments"])

        func = function_dispatch_table.get(func_name)
        if func:
            result = func(**arguments)
            # Ensure the output is a JSON string
            output = json.dumps(result) if not isinstance(result, str) else result
            tools_output.append({
                "tool_call_id": action["id"],
                "output": output
            })
        else:
            print(f"Function {func_name} not found")

    # Submit the tool outputs to Assistant API
    client.beta.threads.runs.submit_tool_outputs(
        thread_id=thread.id,
        run_id=run.id,
        tool_outputs=tools_output
    )


else:
    print("Waiting for the Assistant to process...")
    time.sleep(5)

Categories
AI
AutoGen TeachableAgent

Post author

By praison

Post date

November 23, 2023

Install

pip install "pyautogen[teachable]"

Prerequisite

export OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxx

Code

from autogen import UserProxyAgent, config_list_from_json
from autogen.agentchat.contrib.teachable_agent import TeachableAgent

Create TeachableAgent

teachable_agent = TeachableAgent(
name="teachableagent",
teach_config={
"reset_db": False,
"path_to_db_dir": "teachable_agent_db"
}
)

Create UserProxyAgent

user = UserProxyAgent("user", human_input_mode="ALWAYS")

Chat with TeachableAgent

teachable_agent.initiate_chat(user, message="Hi, I'm a teachable user assistant! What's on your mind?")

Update the database

teachable_agent.learn_from_user_feedback()
teachable_agent.close_db()