[ad_1]
Think about streamlining your whole enterprise administration by means of a single, user-friendly interface in your cellphone. Whereas juggling a number of apps is frequent observe, the longer term lies in consolidating all of your interactions into one chat-based platform, powered by the capabilities of Massive Language Fashions (LLMs).
For small companies, this method gives important benefits. By centralizing information administration duties inside a unified chat interface, house owners can save time, scale back complexity, and decrease reliance on disparate software program instruments. The result’s a extra environment friendly allocation of assets, permitting a higher deal with core enterprise progress actions.
Nevertheless, the potential extends past simply small companies. The ideas and methods detailed on this tutorial are adaptable to non-public use instances as nicely. From managing to-do lists and monitoring bills to organizing collections, a chat-based interface supplies an intuitive and environment friendly approach to work together together with your information.
This text is the second installment in a collection that guides you thru the method of growing such a software program challenge, from preliminary idea to sensible implementation. Constructing upon the elements launched within the earlier article, we’ll set up the foundational parts of our software, together with:
- Organising the database schema
- Defining core software performance
- Structuring the challenge repository
- Creating Instruments able to interacting with a number of SQL database tables utilizing pure language instructions
By the tip of this tutorial, you should have a transparent understanding of tips on how to architect a chat-based interface that leverages LLMs to simplify information administration duties. Whether or not you’re a small enterprise proprietor trying to streamline operations or a person looking for to optimize private group, the ideas lined right here will present a stable place to begin to your personal tasks.
Let’s start by briefly recapping the important thing takeaways from the earlier article to set the context for our present targets.
Within the first a part of this collection, we constructed a prototype agent workflow able to interacting with software objects. Our purpose was to cut back hallucination in software arguments generated by the underlying language mannequin, in our case gpt-3.5-turbo
.
To attain this, we carried out two key modifications:
- Eliminated required parameters within the software schema
- Added a parameter validation step earlier than executing the specified operate
By setting all software parameters to non-obligatory and manually checking for lacking parameters, we eradicated the urge for the Agent/LLM to hallucinate lacking values.
The important thing objects launched within the earlier article had been:
OpenAiAgent
: The primary agent workflow classDevice
: A category representing a software the agent can useToolResult
andStepResult
: Courses for encapsulating software execution outcomes
These elements fashioned the muse of our agent system, permitting it to course of person requests, choose applicable instruments, and generate responses.
When you’d like a extra detailed rationalization or wish to know the reasoning behind particular design selections, be happy to take a look at the earlier article: Leverage OpenAI Device Calling: Constructing a Dependable AI Agent from Scratch
With this recap in thoughts, let’s dive into the subsequent section of our challenge — integrating database performance to retailer and handle enterprise information.
Small companies typically face distinctive challenges on the subject of information upkeep. Like bigger firms, they should usually replace and keep numerous forms of information, corresponding to accounting information, time monitoring, invoices, and extra. Nevertheless, the complexity and prices related to trendy ERP (Enterprise Useful resource Planning) techniques could be prohibitive for small companies. Because of this, many resort to utilizing a collection of Excel spreadsheets to seize and keep important information.
The issue with this method is that small enterprise house owners, who’re hardly ever devoted solely to administrative duties, can’t afford to speculate important effort and time into complicated administration and management processes. The bottom line is to outline lean processes and replace information because it arises, minimizing the overhead of knowledge administration.
By leveraging the facility of Massive Language Fashions and making a chat interface, we intention to simplify and streamline information administration for small companies. The chatbot will act as a unified interface, permitting customers to enter information, retrieve info, and carry out numerous duties utilizing pure language instructions. This eliminates the necessity for navigating a number of spreadsheets or growing complicated net functions with a number of kinds and dashboards.
All through this collection, we’ll regularly improve the chatbot’s capabilities, including options corresponding to role-based entry management, superior querying and analysis, multimodal assist, and integration with widespread communication platforms like WhatsApp. By the tip of the collection, you should have a robust and versatile software that may adapt to your particular wants, whether or not you’re operating a small enterprise or just trying to manage your private life extra effectively.
Let’s get began!
To make sure a well-organized and maintainable challenge, we’ve structured our repository to encapsulate totally different functionalities and elements systematically. Right here’s an summary of the repository construction:
project-root/
│
├── database/
│ ├── db.py # Database connection and setup
│ ├── fashions.py # Database fashions/schemas
| └── utils.py # Database utilities
│
├── instruments/
│ ├── base.py # Base class for instruments
│ ├── add.py # Device for including information to the database
│ ├── question.py # Device for querying information from the database
| └── utils.py # Device utilities
│
├── brokers/
│ ├── base.py # Primary AI agent logic
│ ├── routing.py # Specialised agent for routing duties
│ ├── activity.py # Device wrapper for OpenAI subagents
| └── utils.py # agent utilities
│
└── utils.py # Utility features and lessons
This construction permits for a transparent separation of considerations, making it simpler to develop, keep, and scale our software.
Selecting the best database and ORM (Object-Relational Mapping) library is essential for our software. For this challenge, we’ve chosen the next frameworks:
- SQLAlchemy: A robust SQL toolkit and Object-Relational Mapping (ORM) library for Python. It supplies a set of instruments for interacting with databases utilizing Python objects and lessons.
- SQLModel: A library that builds on prime of SQLAlchemy and Pydantic, providing a easy and intuitive approach to outline database fashions and carry out database operations.
By leveraging SQLModel, we are able to seamlessly combine with Pydantic and SQLAlchemy, enabling environment friendly information validation and database operations whereas eliminating the danger of SQL injection assaults. Furthermore, SQLModel permits us to simply construct upon our beforehand designed Device
class, which makes use of Pydantic fashions for making a software schema.
To make sure the safety and robustness of our software, we implement the next measures:
- Function-based entry management: Executable operations are sure to person roles, guaranteeing that customers can solely carry out actions they’re approved to do. This provides an additional layer of safety and prevents unauthorized entry to delicate information.
- Prevention of SQL injection assaults: By using ChatGPT’s pure language understanding capabilities, we are able to validate and sanitize person inputs, mitigating the danger of SQL injection vulnerabilities. SQLModel’s integration with Pydantic helps us implement strict information validation guidelines.
With our tech stack determined, let’s dive into establishing the database and defining our fashions.
To start constructing our prototype software, we’ll outline the important database tables and their corresponding SQLModel definitions. For this tutorial, we’ll deal with three core tables:
These tables will function the muse for our software, permitting us to display the important thing functionalities and interactions.
Create a brand new file named fashions.py
within the database
listing and outline the tables utilizing SQLModel:
# databasemodels.py
from typing import Non-obligatory from pydantic import BeforeValidator, model_validator
from sqlmodel import SQLModel, Discipline
from datetime import time, datetime
from typing_extensions import Annotated
def validate_date(v):
if isinstance(v, datetime):
return v
for f in ["%Y-%m-%d", "%Y-%m-%d %H:%M:%S"]:
attempt:
return datetime.strptime(v, f)
besides ValueError:
move
increase ValueError("Invalid date format")
def numeric_validator(v):
if isinstance(v, int):
return float(v)
elif isinstance(v, float):
return v
increase ValueError("Worth have to be a quantity")
DateFormat = Annotated[datetime, BeforeValidator(validate_date)]
Numeric = Annotated[float, BeforeValidator(numeric_validator)]
class Buyer(SQLModel, desk=True):
id: Non-obligatory[int] = Discipline(primary_key=True, default=None)
firm: str
first_name: str
last_name: str
cellphone: str
handle: str
metropolis: str
zip: str
nation: str
class Income(SQLModel, desk=True):
id: Non-obligatory[int] = Discipline(primary_key=True, default=None)
description: str
net_amount: Numeric
gross_amount: Numeric
tax_rate: Numeric
date: DateFormat
class Expense(SQLModel, desk=True):
id: Non-obligatory[int] = Discipline(primary_key=True, default=None)
description: str
net_amount: Numeric = Discipline(description="The online quantity of the expense")
gross_amount: Numeric
tax_rate: Numeric
date: DateFormat
Along with the usual SQLModel fields, we’ve outlined three customized sort annotations: DateFormat
, TimeFormat
, and Numeric
. These annotations leverage Pydantic’s BeforeValidator
to make sure that the enter information is appropriately formatted earlier than being saved within the database. The validate_date
operate handles the conversion of string enter to the suitable datetime
. This method permits us to simply accept quite a lot of date codecs from the Massive Language Mannequin, decreasing the necessity for strict format enforcement within the prompts.
With our fashions outlined, we want a script to arrange the database engine and create the corresponding tables. Let’s create a db.py
file within the database
listing to deal with this:
# database/db.py
from database.fashions import *
from sqlmodel import SQLModel, create_engine
import os # native saved database
DATABASE_URL = "sqlite:///app.db"
engine = create_engine(DATABASE_URL, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
create_db_and_tables()
On this script, we import our fashions and the required SQLModel elements. We outline the DATABASE_URL
to level to an area SQLite database file named app.db
. We create an engine
utilizing create_engine
from SQLModel, passing within the DATABASE_URL
. The echo=True
parameter permits verbose output for debugging functions.
The create_db_and_tables
operate makes use of SQLModel.metadata.create_all
to generate the corresponding tables within the database primarily based on our outlined fashions. Lastly, we name this operate to make sure the database and tables are created when the script is run.
With our database setup full, we are able to now deal with updating our Device
class to work seamlessly with SQLModel and improve our software schema conversion course of.
On this part, we’ll talk about the updates made to the Device
class to deal with SQLModel situations and enhance the validation course of. For a extra detailed rationalization of the Device
class, go to my earlier article.
First, we’ve added Kind[SQLModel]
as a doable sort for the mannequin
discipline utilizing the Union
sort trace. This enables the Device
class to simply accept each Pydantic’s BaseModel
and SQLModel’s SQLModel
as legitimate mannequin varieties.
Subsequent, we’ve launched a brand new attribute known as exclude_keys
of sort listing[str]
with a default worth of ["id"]
. The aim of this attribute is to specify which keys needs to be excluded from the validation course of and the OpenAI software schema era. On this case the default excluded key’s id
since for information entry creation with SqlModel
the id is mechanically generated throughout ingestion.
On prime of that we launched parse_model
boolean attribute to our Device class. The place we are able to principally determined if the software operate is named with our pydantic/SQLModel or with key phrase arguments.
Within the validate_input()
methodology, we have added a verify to make sure that the keys laid out in exclude_keys
will not be thought of as lacking keys in the course of the validation course of. That is notably helpful for fields like id
, that are mechanically generated by SQLModel and shouldn’t be required within the enter.
Equally, within the openai_tool_schema
property, we have added a loop to take away the excluded keys from the generated schema. This ensures that the excluded keys will not be included within the schema despatched to the OpenAI API. For recap we use the openai_tool_schema
property to take away required
arguments from our software schema. That is performed to elimenate hallucination by our language mannequin.
Furthermore, we modified the import from from pydantic.v1 import BaseModel
to from pydantic import BaseModel
. Since SQLModel
relies on Pydantic v2, we wish to be constant and use Pydantic v2 at this level.
Right here’s the up to date code for the Device
class:
# instruments/base.py
from typing import Kind, Callable, Unionfrom instruments.convert import convert_to_openai_tool
from pydantic import BaseModel, ConfigDict
from sqlmodel import SQLModel
class ToolResult(BaseModel):
content material: str
success: bool
class Device(BaseModel):
title: str
mannequin: Union[Type[BaseModel], Kind[SQLModel], None]
operate: Callable
validate_missing: bool = True
parse_model: bool = False
exclude_keys: listing[str] = ["id"]
model_config = ConfigDict(arbitrary_types_allowed=True)
def run(self, **kwargs) -> ToolResult:
if self.validate_missing and mannequin isn't None:
missing_values = self.validate_input(**kwargs)
if missing_values:
content material = f"Lacking values: {', '.be part of(missing_values)}"
return ToolResult(content material=content material, success=False)
if self.parse_model:
if hasattr(self.mannequin, "model_validate"):
input_ = self.mannequin.model_validate(kwargs)
else:
input_ = self.mannequin(**kwargs)
outcome = self.operate(input_)
else:
outcome = self.operate(**kwargs)
return ToolResult(content material=str(outcome), success=True)
def validate_input(self, **kwargs):
if not self.validate_missing or not self.mannequin:
return []
model_keys = set(self.mannequin.__annotations__.keys()) - set(self.exclude_keys)
input_keys = set(kwargs.keys())
missing_values = model_keys - input_keys
return listing(missing_values)
@property
def openai_tool_schema(self):
schema = convert_to_openai_tool(self.mannequin)
# set operate title
schema["function"]["name"] = self.title
# take away required discipline
if schema["function"]["parameters"].get("required"):
del schema["function"]["parameters"]["required"]
# take away exclude keys
if self.exclude_keys:
for key in self.exclude_keys:
if key in schema["function"]["parameters"]["properties"]:
del schema["function"]["parameters"]["properties"][key]
return schema
These updates to the Device
class present extra flexibility and management over the validation course of and schema era when working with SQLModel situations.
In our Device
class, we create a schema from a Pydantic mannequin utilizing the convert_to_openai_tool
operate from Langchain. Nevertheless, this operate relies on Pydantic v1, whereas SQLModel makes use of Pydantic v2. To make the conversion operate appropriate, we have to adapt it. Let’s create a brand new script known as convert.py
:
# instruments/convert.py
from langchain_core.utils.function_calling import _rm_titles
from typing import Kind, Non-obligatory
from langchain_core.utils.json_schema import dereference_refs
from pydantic import BaseModeldef convert_to_openai_tool(
mannequin: Kind[BaseModel],
*,
title: Non-obligatory[str] = None,
description: Non-obligatory[str] = None,
) -> dict:
"""Converts a Pydantic mannequin to a operate description for the OpenAI API."""
operate = convert_pydantic_to_openai_function(
mannequin, title=title, description=description
)
return {"sort": "operate", "operate": operate}
def convert_pydantic_to_openai_function(
mannequin: Kind[BaseModel],
*,
title: Non-obligatory[str] = None,
description: Non-obligatory[str] = None,
rm_titles: bool = True,
) -> dict:
"""Converts a Pydantic mannequin to a operate description for the OpenAI API."""
model_schema = mannequin.model_json_schema() if hasattr(mannequin, "model_json_schema") else mannequin.schema()
schema = dereference_refs(model_schema)
schema.pop("definitions", None)
title = schema.pop("title", "")
default_description = schema.pop("description", "")
return {
"title": title or title,
"description": description or default_description,
"parameters": _rm_titles(schema) if rm_titles else schema,
}
This tailored conversion operate handles the variations between Pydantic v1 and v2, guaranteeing that our Device
class can generate appropriate schemas for the OpenAI API.
Subsequent, replace the import assertion in instruments/base.py
to make use of the brand new convert_to_openai_tool
operate:
# instruments/base.py
from typing import Kind, Callable, Unionfrom instruments.convert import convert_to_openai_tool
from pydantic import BaseModel
from sqlmodel import SQLModel
#...remainder of the code ...
With these modifications in place, our Device
class can now deal with SQLModel situations and generate schemas which can be appropriate with the OpenAI API.
Word: When you encounter dependency points, you might think about eradicating the Langchain dependency solely and together with the
_rm_titles
anddereference_refs
features instantly within theconvert.py
file.
By adapting the software schema conversion course of, we’ve ensured that our software can seamlessly work with SQLModel and Pydantic v2, enabling us to leverage the advantages of those libraries whereas sustaining compatibility with the OpenAI API.
On this part, we’ll create features and instruments to work together with our database tables utilizing SQL.
4.1 Add Knowledge Device
First, let’s outline a generic operate add_row_to_table
that takes a SQLModel occasion and provides it to the corresponding desk:
# instruments/add.py
from sqlmodel import SQLModel, Session, choosedef add_row_to_table(model_instance: SQLModel):
with Session(engine) as session:
session.add(model_instance)
session.commit()
session.refresh(model_instance)
return f"Efficiently added {model_instance} to the desk"
Subsequent, we’ll create a model-specific operate add_expense_to_table
that takes enter arguments for an Expense entry and provides it to the desk:
# instruments/add.py
# ...
def add_expense_to_table(**kwargs):
model_instance = Expense.model_validate(kwargs)
return add_row_to_table(model_instance)
In add_expense_to_table
, we use the model_validate()
methodology to set off the execution of the beforehand outlined BeforeValidator and guarantee information validation.
To keep away from writing separate features for every desk or SQLModel, we are able to dynamically generate the features:
# instance utilizationdef add_entry_to_table(sql_model: Kind[SQLModel]):
# return a Callable that takes a SQLModel occasion and provides it to the desk
return lambda **information: add_row_to_table(model_instance=sql_model.model_validate(information))
add_expense_to_table = add_entry_to_table(Expense)
This method produces the identical outcome and can be utilized to dynamically generate features for all different fashions.
With these features in place, we are able to create instruments utilizing our Device
class so as to add entries to our database tables by way of the OpenAIAgent:
add_expense_tool = Device(
title="add_expense_tool",
description="helpful for including bills to database",
operate=add_entry_to_table(Expense),
mannequin=Expense,
validate_missing=True
)add_revenue_tool = Device(
title="add_revenue_tool",
description="helpful for including income to database",
operate=add_entry_to_table(Income),
mannequin=Income,
validate_missing=True
)
Whereas we have to create an add_xxx_tool for every desk as a consequence of various enter schemas, we solely want one question software for querying all tables. To get rid of the danger of SQL injection, we’ll use the SQL sanitization offered by SQLAlchemy and SQLModel. This implies we’ll question the database by means of customary Python lessons and objects as a substitute of parsing SQL statements instantly.
For the queries we wish to carry out on our tables, we’ll want the next logic:
- choose assertion ->
SELECT * FROM table_name
Arguments:columns
,table_name
- the place assertion ->
WHERE column_name = worth
Arguments:column
,operator
,worth
In SQLModel, this corresponds to the next sanitized code once we wish to discover all bills for espresso within the Expense
desk:
outcome = database.execute(
choose(Expense).the place(Expense.description == "Espresso")
)
To summary this right into a pydantic mannequin:
# instruments/question.py
from typing import Union, Literal
from pydantic import BaseModelclass WhereStatement(BaseModel):
column: str
operator: Literal["eq", "gt", "lt", "gte", "lte", "ne", "ct"]
worth: str
class QueryConfig(BaseModel):
table_name: str
columns: listing[str]
the place: listing[Union[WhereStatement, None]]
The QueryConfig
mannequin permits us to set a table_name
, columns
, and the place
statements. The the place
property accepts a listing of WhereStatement
fashions or an empty listing (once we wish to return all values with no additional filtering). A WhereStatement
is a submodel defining a column, operator, and worth. The Literal
sort is used to limit the allowed operators to a predefined set.
Subsequent, we outline a operate that executes a question primarily based on the QueryConfig
:
# instruments/question.py
# ...
from database.fashions import Expense, Income, BuyerTABLES = {
"expense": Expense,
"income": Income,
"buyer": Buyer
}
def query_data_function(**kwargs) -> ToolResult:
"""Question the database by way of pure language."""
query_config = QueryConfig.model_validate(kwargs)
if query_config.table_name not in TABLES:
return ToolResult(content material=f"Desk title {query_config.table_name} not present in database fashions", success=False)
sql_model = TABLES[query_config.table_name]
# query_config = validate_query_config(query_config, sql_model)
information = sql_query_from_config(query_config, sql_model)
return ToolResult(content material=f"Question outcomes: {information}", success=True)
def sql_query_from_config(
query_config: QueryConfig,
sql_model: Kind[SQLModel]):
with Session(engine) as session:
choice = []
for column in query_config.select_columns:
if column not in sql_model.__annotations__:
return f"Column {column} not present in mannequin {sql_model.__name__}"
choice.append(getattr(sql_model, column))
assertion = choose(*choice)
wheres = query_config.the place
if wheres:
for the place in wheres:
if the place.column not in sql_model.__annotations__: # noqa
return (f"Column {the place['column']} not discovered "
"in mannequin {sql_model.__name__}")
elif the place.operator == "eq":
assertion = assertion.the place(
getattr(sql_model, the place.column) == the place.worth)
elif the place.operator == "gt":
assertion = assertion.the place(
getattr(sql_model, the place.column) > the place.worth)
elif the place.operator == "lt":
assertion = assertion.the place(
getattr(sql_model, the place.column) < the place.worth)
elif the place.operator == "gte":
assertion = assertion.the place(
getattr(sql_model, the place.column) >= the place.worth)
elif the place.operator == "lte":
assertion = assertion.the place(
getattr(sql_model, the place.column) <= the place.worth)
elif the place.operator == "ne":
assertion = assertion.the place(
getattr(sql_model, the place.column) != the place.worth)
elif the place.operator == "ct":
assertion = assertion.the place(
getattr(sql_model, the place.column).accommodates(the place.worth))
outcome = session.exec(assertion)
information = outcome.all()
attempt:
information = [repr(d) for d in data]
besides:
move
return information
The query_data_function
serves as a high-level abstraction for choosing our desk mannequin from the TABLES
dictionary, whereas sql_query_from_config
is the underlying operate for executing the QueryConfig
on a desk (SQLModel).
In `QueryConfig` you possibly can select to additionally outline table_names as Literal sort, the place you laborious code the accessible desk names into it. You’ll be able to even dynamically outline the Literal utilizing our TABLES dictionary. By doing so you possibly can scale back false arguments for table_name. For now I’ve choosen to not use an enum object, as a result of I’ll present the agent immediate with context concerning the at present accessible tables and there underling ORM schema. I plan so as to add a software for our future agent to create new tables on it’s personal.Whereas I can dynamically change the agent’s immediate, it gained’t be simple to alter the enum object inside `QueryConfig` on our operating server.
Lastly, we are able to outline our question software:
query_data_tool = Device(
title="query_data_tool",
description = "helpful to carry out queries on a database desk",
mannequin=QueryConfig,
operate=query_data_function,
)
With these instruments in place, our OpenAIAgent is now able to including and querying information in our database tables utilizing pure language instructions.
To allow profitable software utilization for our beforehand outlined instruments, the Agent from the earlier article will want extra context info, particularly for utilizing the question software. The Agent immediate might want to embrace details about accessible tables and their schemas. Since we solely use two tables at this level, we are able to embrace the ORM schema and desk names within the system immediate or person immediate. Each choices may work nicely, however I want to incorporate variable info like this within the person immediate. By doing so, we are able to create few-shot examples that display context-aware software utilization.
To make our Agent able to dealing with variable context within the system immediate and person immediate, we are able to replace our Agent class as follows:
import colorama
from colorama import Fore
from openai import OpenAI
from pydantic import BaseModel
from instruments.base import Device, ToolResult
from brokers.utils import parse_function_args, run_tool_from_response class StepResult(BaseModel):
occasion: str
content material: str
success: bool
SYSTEM_MESSAGE = """You might be tasked with finishing particular targets and should report the outcomes. At your disposal, you might have quite a lot of instruments, every specialised in performing a definite sort of activity.
For profitable activity completion:
Thought: Think about the duty at hand and decide which software is greatest suited primarily based on its capabilities and the character of the work. When you can full the duty or reply a query, soley by the data offered you should use the report_tool instantly.
Use the report_tool with an instruction detailing the outcomes of your work or to reply a person query.
When you encounter a difficulty and can't full the duty:
Use the report_tool to speak the problem or motive for the duty's incompletion.
You'll obtain suggestions primarily based on the outcomes of every software's activity execution or explanations for any duties that could not be accomplished. This suggestions loop is essential for addressing and resolving any points by strategically deploying the accessible instruments.
Return just one software name at a time.
{context}
"""
class OpenAIAgent:
def __init__(
self,
instruments: listing[Tool],
consumer: OpenAI = OpenAI(),
system_message: str = SYSTEM_MESSAGE,
model_name: str = "gpt-3.5-turbo-0125",
max_steps: int = 5,
verbose: bool = True,
examples: listing[dict] = None,
context: str = None,
user_context: str = None
):
self.instruments = instruments
self.consumer = consumer
self.model_name = model_name
self.system_message = system_message
self.step_history = []
self.max_steps = max_steps
self.verbose = verbose
self.examples = examples or []
self.context = context or ""
self.user_context = user_context
def to_console(self, tag: str, message: str, colour: str = "inexperienced"):
if self.verbose:
color_prefix = Fore.__dict__[color.upper()]
print(color_prefix + f"{tag}: {message}{colorama.Fashion.RESET_ALL}")
def run(self, user_input: str, context: str = None):
openai_tools = [tool.openai_tool_schema for tool in self.tools]
system_message = self.system_message.format(context=context)
if self.user_context:
context = f"{self.user_context}n{context}" if context else self.user_context
if context:
user_input = f"{context}n---nnUser Message: {user_input}"
self.to_console("START", f"Beginning Agent with Enter:n'''{user_input}'''")
self.step_history = [
{"role": "system", "content": system_message},
*self.examples,
{"role": "user", "content": user_input}
]
step_result = None
i = 0
whereas i < self.max_steps:
step_result = self.run_step(self.step_history, openai_tools)
if step_result.occasion == "end":
break
elif step_result.occasion == "error":
self.to_console(step_result.occasion, step_result.content material, "crimson")
else:
self.to_console(step_result.occasion, step_result.content material, "yellow")
i += 1
self.to_console("Remaining End result", step_result.content material, "inexperienced")
return step_result.content material
def run_step(self, messages: listing[dict], instruments):
# plan the subsequent step
response = self.consumer.chat.completions.create(
mannequin=self.model_name,
messages=messages,
instruments=instruments
)
# verify for a number of software calls
if response.selections[0].message.tool_calls and len(response.selections[0].message.tool_calls) > 1:
messages = [
*self.step_history,
{"role": "user", "content": "Error: Please return only one tool call at a time."}
]
return self.run_step(messages, instruments)
# add message to historical past
self.step_history.append(response.selections[0].message)
# verify if software name is current
if not response.selections[0].message.tool_calls:
msg = response.selections[0].message.content material
step_result = StepResult(occasion="Error", content material=f"No software calls had been returned.nMessage: {msg}", success=False)
return step_result
tool_name = response.selections[0].message.tool_calls[0].operate.title
tool_kwargs = parse_function_args(response)
# execute the software name
self.to_console("Device Name", f"Title: {tool_name}nArgs: {tool_kwargs}", "magenta")
tool_result = run_tool_from_response(response, instruments=self.instruments)
tool_result_msg = self.tool_call_message(response, tool_result)
self.step_history.append(tool_result_msg)
if tool_name == "report_tool":
attempt:
step_result = StepResult(
occasion="end",
content material=tool_result.content material,
success=True
)
besides:
print(tool_result)
increase ValueError("Report Device did not run.")
return step_result
elif tool_result.success:
step_result = StepResult(
occasion="tool_result",
content material=tool_result.content material,
success=True)
else:
step_result = StepResult(
occasion="error",
content material=tool_result.content material,
success=False
)
return step_result
def tool_call_message(self, response, tool_result: ToolResult):
tool_call = response.selections[0].message.tool_calls[0]
return {
"tool_call_id": tool_call.id,
"function": "software",
"title": tool_call.operate.title,
"content material": tool_result.content material,
}
The primary modifications in comparison with our earlier model:
- We positioned a “{context}” placeholder within the default system immediate.
- We added
context
anduser_context
as enter arguments to__init__()
. - We added
context
to therun()
methodology. - In
run()
, we addcontext
to the person message if outlined. - We additionally added an
examples
attribute to__init__()
that, if set, shall be handed between the system and person messages inrun()
.
Now we are able to outline a system context and a person context whereas initializing our agent. Moreover, we are able to move a person context when calling the run methodology. If context
is handed to the run methodology, it’s going to overwrite the user_context
from initialization for that run.
Earlier than we are able to run our Agent, let’s outline a operate that generates context info. We wish to mechanically generate user_context
, which we are able to then move to the Agent’s run operate as carried out above. To maintain it easy, we would like a single line for every desk as context info that ought to embrace:
- Desk title
- Column_name:
<sort>
After a couple of makes an attempt with trial and error, the next operate will do the job:
# utils.py
from typing import Kind
import varieties
import typing import sqlalchemy
from pydantic import BaseModel
def orm_model_to_string(input_model_cls: Kind[BaseModel]):
"""Get the ORM mannequin string from the enter mannequin"""
def process_field(key, worth):
if key.startswith("__"):
return None
if isinstance(worth, typing._GenericAlias):
if worth.__origin__ == sqlalchemy.orm.base.Mapped:
return None
if isinstance(worth, typing._AnnotatedAlias): # noqa
return key, worth.__origin__
elif isinstance(worth, typing._UnionGenericAlias) or isinstance(worth, varieties.UnionType):
return key, worth.__args__[0]
return key, worth
fields = dict(filter(None, (process_field(ok, v) for ok, v in input_model_cls.__annotations__.objects())))
return ", ".be part of([f"{k} = <{v.__name__}>" for k, v in fields.items()])
def generate_context(*table_models) -> str:
context_str = "You'll be able to entry the next tables in database:n"
for desk in table_models:
context_str += f" - {desk.__name__}: {orm_model_to_string(desk)}n"
return context_str
If we move Expense
and Income
to generate_context()
, we should always get the next context string:
We would like the Agent to know the present date and day of the week, so we are able to reference the proper date. So let’s add some date parsing features to our utils class:
# utils.py
from datetime import datetime#... remainder of utils.py ...
def weekday_by_date(date: datetime):
days = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
return days[date.weekday()]
def date_to_string(date: datetime):
return f"{weekday_by_date(date)} {parse_date(date)}"
def parse_date(date: datetime):
return date.strftime("%Y-%m-%d")
Now let’s create the context for a question agent
# utils.py# ...
def generate_query_context(*table_models) -> str:
as we speak = f"At the moment is {date_to_string(datetime.now())}"
context_str = "You'll be able to entry the next tables in database:n"
for desk in table_models:
context_str += f" - {desk.__name__}: {orm_model_to_string(desk)}n"
return f"{as we speak}n{context_str}"
from database.fashions import Expense, Income
print(generate_query_context(Expense, Income))
At the moment is Sunday 2024-04-21
You'll be able to entry the next tables in database:
- Expense: id = <int>, description = <str>, net_amount = <float>, gross_amount = <float>, tax_rate = <float>, date = <datetime>
- Income: id = <int>, description = <str>, net_amount = <float>, gross_amount = <float>, tax_rate = <float>, date = <datetime>
As we add extra instruments, the complexity of our setup might begin to restrict the usability of cheaper fashions like “gpt-3.5-turbo”. Within the subsequent article, we would think about switching to Anthropic Claude, since their newly launched tool-use API function appears promising, even for the extra inexpensive HAIKU mannequin, in dealing with a number of instruments concurrently. Nevertheless, for now, we’ll proceed utilizing OpenAI’s GPT fashions.
When growing for private use and earlier than creating production-ready functions, I discover it helpful to optimize the workflow for smaller fashions, corresponding to gpt-3.5-turbo
on this case. This method forces us to create a streamlined processing logic and prompting system. Whereas we might not obtain 100% reliability with out utilizing probably the most highly effective mannequin, we will catch flaws and determine unclear directions. In case your software works in 9 out of 10 instances with a smaller mannequin, you should have a production-ready logic that may carry out even higher with a stronger mannequin.
To make multi-tool dealing with dependable with gpt-3.5-turbo
we’ll implement a routing agent whose sole function is to route the person question to the suitable activity agent. This enables us to separate execution logic and scale back complexity. Every agent can have a restricted scope, enabling us to separate entry roles and operations sooner or later. I’ve noticed that even with gpt-4, there are situations the place the agent doesn’t know when its activity is completed.
By introducing a routing agent, we are able to break down the issue into smaller, extra manageable components. The routing agent shall be chargeable for understanding the person’s intent and directing the question to the related activity agent. This method not solely simplifies the person brokers’ tasks but additionally makes the system extra modular and simpler to keep up.
Moreover, separating the execution logic and complexity will pave the way in which for implementing role-based entry management sooner or later. Every activity agent could be assigned particular permissions and entry ranges, guaranteeing that delicate operations are solely carried out by approved brokers.
Whereas the routing agent provides an additional step within the course of, it finally results in a extra sturdy and scalable system. By optimizing for smaller fashions and specializing in clear, concise prompts, we are able to create a stable basis that may carry out even higher once we swap to extra highly effective fashions like Claude Opus or GPT-4.
Let’s take a look on the implementation of the routing agent
# brokers/routing.py
from openai import OpenAI
import colorama
from brokers.task_agent import TaskAgent
from brokers.utils import parse_function_argsSYSTEM_MESSAGE = """You're a useful assistant.
Function: You might be an AI Assistant designed to function the first level of contact for customers interacting by means of a chat interface.
Your major function is to grasp customers' requests associated to database operations and route these requests to the suitable software.
Capabilities:
You could have entry to quite a lot of instruments designed for Create, Learn operations on a set of predefined tables in a database.
Tables:
{table_names}
"""
NOTES = """Vital Notes:
All the time verify the completion of the requested operation with the person.
Preserve person privateness and information safety all through the interplay.
If a request is ambiguous or lacks particular particulars, ask follow-up inquiries to make clear the person's wants."""
class RoutingAgent:
def __init__(
self,
instruments: listing[TaskAgent] = None,
consumer: OpenAI = OpenAI(),
system_message: str = SYSTEM_MESSAGE,
model_name: str = "gpt-3.5-turbo-0125",
max_steps: int = 5,
verbose: bool = True,
prompt_extra: dict = None,
examples: listing[dict] = None,
context: str = None
):
self.instruments = instruments or ROUTING_AGENTS
self.consumer = consumer
self.model_name = model_name
self.system_message = system_message
self.reminiscence = []
self.step_history = []
self.max_steps = max_steps
self.verbose = verbose
self.prompt_extra = prompt_extra or PROMPT_EXTRA
self.examples = self.load_examples(examples)
self.context = context or ""
def load_examples(self, examples: listing[dict] = None):
examples = examples or []
for agent in self.instruments:
examples.prolong(agent.routing_example)
return examples
def run(self, user_input: str, employee_id: int = None, **kwargs):
context = create_routing_agent_context(employee_id)
if context:
user_input_with_context = f"{context}n---nnUser Message: {user_input}"
else:
user_input_with_context = user_input
self.to_console("START", f"Beginning Process Agent with Enter:n'''{user_input_with_context}'''")
partial_variables = {**self.prompt_extra, "context": context}
system_message = self.system_message.format(**partial_variables)
messages = [
{"role": "system", "content": system_message},
*self.examples,
{"role": "user", "content": user_input}
]
instruments = [tool.openai_tool_schema for tool in self.tools]
response = self.consumer.chat.completions.create(
mannequin=self.model_name,
messages=messages,
instruments=instruments
)
self.step_history.append(response.selections[0].message)
self.to_console("RESPONSE", response.selections[0].message.content material, colour="blue")
tool_kwargs = parse_function_args(response)
tool_name = response.selections[0].message.tool_calls[0].operate.title
self.to_console("Device Title", tool_name)
self.to_console("Device Args", tool_kwargs)
agent = self.prepare_agent(tool_name, tool_kwargs)
return agent.run(user_input)
def prepare_agent(self, tool_name, tool_kwargs):
for agent in self.instruments:
if agent.title == tool_name:
input_kwargs = agent.arg_model.model_validate(tool_kwargs)
return agent.load_agent(**input_kwargs.dict())
increase ValueError(f"Agent {tool_name} not discovered")
def to_console(self, tag: str, message: str, colour: str = "inexperienced"):
if self.verbose:
color_prefix = colorama.Fore.__dict__[color.upper()]
print(color_prefix + f"{tag}: {message}{colorama.Fashion.RESET_ALL}")
The largest variations to our OpenAIAgent
are:
- No open loop: we would like the routing agent to route person’s queries to the suitable agent. So as a substitute of making an open loop we choose the specified agent by way of software calling and move the person question to it. The routing Agent mustn’t do every other activity or follow-up query.
- Brokers as Instruments: As an alternative of calling a software the routing agent setup a subagent. So our beforehand outlined
OpenAIAgent
is now a software inside our routing agent.
To make use of our OpenAIAgent
as a software, we have to introduce some form of software class devoted for Brokers. We wish to outline a reputation and outline for every agent and automate the initialization course of. Subsequently, we outline our final class for this tutorial theTaskAgent
.
The TaskAgent
class serves related performance because the Device
class. We outline a reputation an outline and an enter mannequin which we name arg_model
.
from typing import Kind, Callable, Non-obligatoryfrom brokers.base import OpenAIAgent
from instruments.base import Device
from instruments.report_tool import report_tool
from pydantic import BaseModel, ConfigDict, Discipline
from instruments.utils import convert_to_openai_tool
SYSTEM_MESSAGE = """You might be tasked with finishing particular targets and should report the outcomes. At your disposal, you might have quite a lot of instruments, every specialised in performing a definite sort of activity.
For profitable activity completion:
Thought: Think about the duty at hand and decide which software is greatest suited primarily based on its capabilities and the character of the work.
When you can full the duty or reply a query, soley by the data offered you should use the report_tool instantly.
Use the report_tool with an instruction detailing the outcomes of your work or to reply a person query.
When you encounter a difficulty and can't full the duty:
Use the report_tool to speak the problem or motive for the duty's incompletion.
You'll obtain suggestions primarily based on the outcomes of every software's activity execution or explanations for any duties that could not be accomplished. This suggestions loop is essential for addressing and resolving any points by strategically deploying the accessible instruments.
On error: If info are lacking think about for those who can deduce or calculate the lacking info and repeat the software name with extra arguments.
Use the data offered by the person to deduct the proper software arguments.
Earlier than utilizing a software take into consideration the arguments and clarify every enter argument used within the software.
Return just one software name at a time! Clarify your ideas!
{context}
"""
class EmptyArgModel(BaseModel):
move
class TaskAgent(BaseModel):
title: str
description: str
arg_model: Kind[BaseModel] = EmptyArgModel
create_context: Callable = None
create_user_context: Callable = None
tool_loader: Callable = None
system_message: str = SYSTEM_MESSAGE
instruments: listing[Tool]
examples: listing[dict] = None
routing_example: listing[dict] = Discipline(default_factory=listing)
model_config = ConfigDict(arbitrary_types_allowed=True)
def load_agent(self, **kwargs) -> OpenAIAgent:
input_kwargs = self.arg_model(**kwargs)
kwargs = input_kwargs.dict()
context = self.create_context(**kwargs) if self.create_context else None
user_context = self.create_user_context(**kwargs) if self.create_user_context else None
if self.tool_loader:
self.instruments.prolong(self.tool_loader(**kwargs))
if report_tool not in self.instruments:
self.instruments.append(report_tool)
return OpenAIAgent(
instruments=self.instruments,
context=context,
user_context=user_context,
system_message=self.system_message,
examples=self.examples,
)
@property
def openai_tool_schema(self):
return convert_to_openai_tool(self.arg_model, title=self.title, description=self.description)
Moreover, we added all related attributes to our TaskAgent
class, which we want for an underlying specialised OpenAIAgent
:
create_context
/create_user_context
: Right here we are able to move a operate to create the context or person context like in part 5.1tool_loader
is one other callable operate which we might have for establishing the underlying agent. As in our dynamic software constructing beforehand defined, we might have instruments which can be dynamically constructed primarily based on the person enter/routing agent enter.system_message
is the agent’s system immediate. In our instance, it will likely be the default system immediate for each agent, however it may be an optimized model for every specialised agent.instruments
: Predefined instruments the agent ought to use.examples
: Examples to incorporate in subagent’s message historical pastrouting_example
: Examples to incorporate in routing agent’s message historical past
Furthermore, we’ve an emty BaseModel known as EmptyArgModel
which is default arg_model
in our TaskAgent
Let’s see if all of it performs collectively!
Now, it’s time to check if our routing and subagents work nicely collectively. As we launched examples as a paremeter we are able to use a number of take a look at runs to examine main flaws within the execution and outline instance utilization for every sub agent.
Let’s outline our subagents first:
from database.fashions import Expense, Income, Buyer
from brokers.activity import TaskAgent
from utils import generate_query_contextfrom instruments.base import Device
from instruments.question import query_data_tool
from instruments.add import add_entry_to_table
query_task_agent = TaskAgent(
title="query_agent",
description="An agent that may carry out queries on a number of information sources",
create_user_context=lambda: generate_query_context(Expense, Income, Buyer),
instruments=[query_data_tool]
)
add_expense_agent = TaskAgent(
title="add_expense_agent",
description="An agent that may add an expense to the database",
create_user_context=lambda: generate_query_context(Expense) + "nRemarks: The tax charge is 0.19. The person present the online quantity that you must calculate the gross quantity.",
instruments=[
Tool(
name="add_expense",
description="Add an expense to the database",
function=add_entry_to_table(Expense),
model=Expense
)
]
)
add_revenue_agent = TaskAgent(
title="add_revenue_agent",
description="An agent that may add a income entry to the database",
create_user_context=lambda: generate_query_context(Income) + "nRemarks: The tax charge is 0.19. The person present the gross_amount you must use the tax charge to calculate the net_amount.",
instruments=[
Tool(
name="add_revenue",
description="Add a revenue entry to the database",
function=add_entry_to_table(Revenue),
model=Revenue
)
]
)
add_customer_agent = TaskAgent(
title="add_customer_agent",
description="An agent that may add a buyer to the database",
create_user_context=lambda: generate_query_context(Buyer),
instruments=[
Tool(
name="add_customer",
description="Add a customer to the database",
function=add_entry_to_table(Customer),
model=Customer
)
]
)
As you possibly can see we added some remarks as string to create_user_context
for income and expense brokers. We would like the sub agent to deal with tax charges and calculate the online or gross quantity mechanically to check the reasoning capabilites of our sub agent.
from brokers.routing import RoutingAgentrouting_agent = RoutingAgent(
instruments=[
query_task_agent,
add_expense_agent,
add_revenue_agent,
add_customer_agent
]
)
routing_agent.run("I've spent 5 € on a workplace stuff. Final Thursday")
START: Beginning Routing Agent with Enter:
I've spent 5 € on a workplace stuff. Final ThursdayDevice Title: add_expense_agent
Device Args: {}
START: Beginning Process Agent with Enter:
"""At the moment is Sunday 2024-04-21
You'll be able to entry the next tables in database:
- expense: id = <int>, description = <str>, net_amount = <float>, gross_amount = <float>, tax_rate = <float>, date = <datetime>
Remarks: The tax charge is 0.19. The person present the online quantity that you must calculate the gross quantity.
---
Consumer Message: I've spent 5 € on a workplace stuff. Final Thursday"""
Device Name: Title: add_expense
Args: {'description': 'workplace stuff', 'net_amount': 5, 'tax_rate': 0.19, 'date': '2024-04-18'}
Message: None
error: Lacking values: gross_amount
Device Name: Title: add_expense
Args: {'description': 'workplace stuff', 'net_amount': 5, 'tax_rate': 0.19, 'date': '2024-04-18', 'gross_amount': 5.95}
Message: None
tool_result: Efficiently added net_amount=5.0 id=2 gross_amount=5.95 description='workplace stuff' date=datetime.datetime(2024, 4, 18, 0, 0) tax_rate=0.19 to the desk
Error: No software calls had been returned.
Message: I've efficiently added the expense for workplace stuff with a internet quantity of 5€, calculated the gross quantity, and recorded it within the database.
Device Name: Title: report_tool
Args: {'report': 'Expense for workplace stuff with a internet quantity of 5€ has been efficiently added. Gross quantity calculated as 5.95€.'}
Message: None
Remaining End result: Expense for workplace stuff with a internet quantity of 5€ has been efficiently added. Gross quantity calculated as 5.95€.
Now let’s add a income:
routing_agent.run("Two weeks in the past on Saturday we had a income of 1000 € within the store")
START: Beginning Routing Agent with Enter:
Two weeks in the past on Saturday we had a income of 1000 € within the storeDevice Title: add_revenue_agent
Device Args: {}
START: Beginning Process Agent with Enter:
"""At the moment is Sunday 2024-04-21
You'll be able to entry the next tables in database:
- income: id = <int>, description = <str>, net_amount = <float>, gross_amount = <float>, tax_rate = <float>, date = <datetime>
Remarks: The tax charge is 0.19. The person present the gross_amount you must use the tax charge to calculate the net_amount.
---
Consumer Message: Two weeks in the past on Saturday we had a income of 1000 € within the store"""
Device Name: Title: add_revenue
Args: {'description': 'Income from the store', 'gross_amount': 1000, 'tax_rate': 0.19, 'date': '2024-04-06'}
Message: None
error: Lacking values: net_amount
Device Name: Title: add_revenue
Args: {'description': 'Income from the store', 'gross_amount': 1000, 'tax_rate': 0.19, 'date': '2024-04-06', 'net_amount': 840.34}
Message: None
tool_result: Efficiently added net_amount=840.34 gross_amount=1000.0 tax_rate=0.19 description='Income from the store' id=1 date=datetime.datetime(2024, 4, 6, 0, 0) to the desk
Error: No software calls had been returned.
Message: The income entry for the store on April 6, 2024, with a gross quantity of 1000€ has been efficiently added to the database. The calculated internet quantity after making use of the tax charge is 840.34€.
Device Name: Title: report_tool
Args: {'report': 'accomplished'}
Message: None
Remaining End result: accomplished
And for the final take a look at let’s attempt to question the income that created from database:
routing_agent.run("How a lot income did we made this month?")
START: Beginning Routing Agent with Enter:
How a lot income did we made this month?Device Title: query_agent
Device Args: {}
START: Beginning Agent with Enter:
"""At the moment is Sunday 2024-04-21
You'll be able to entry the next tables in database:
- expense: id = <int>, description = <str>, net_amount = <float>, gross_amount = <float>, tax_rate = <float>, date = <datetime>
- income: id = <int>, description = <str>, net_amount = <float>, gross_amount = <float>, tax_rate = <float>, date = <datetime>
- buyer: id = <int>, company_name = <str>, first_name = <str>, last_name = <str>, cellphone = <str>, handle = <str>, metropolis = <str>, zip = <str>, nation = <str>
---
Consumer Message: How a lot income did we made this month?"""
Device Name: Title: query_data_tool
Args: {'table_name': 'income', 'select_columns': ['gross_amount'], 'the place': [{'column': 'date', 'operator': 'gte', 'value': '2024-04-01'}, {'column': 'date', 'operator': 'lte', 'value': '2024-04-30'}]}
Message: None
tool_result: content material="Question outcomes: ['1000.0']" success=True
Error: No software calls had been returned.
Message: The income made this month is $1000.00.
Device Name: Title: report_tool
Args: {'report': 'The income made this month is $1000.00.'}
Message: None
Remaining End result: The income made this month is $1000.00.
All instruments labored as anticipated. The Routing Agent labored completely. For theTask Agent I needed to replace the immediate a number of occasions.
I might suggest so as to add some instance software calls to every activity agent when not working with state-of-the-art fashions like gpt-4. Usually I might suggest to sort out flaws with examples and extra intuitive designs as a substitute of immediate engineering. Reapting flaws are indicators for not simple designs. For instance when the agent struggles with calculating the gross or internet quantity simply add a ‘calculate_gross_amount_tool’ or ‘calculate_net_amount_tool’. GPT-4 however would deal with use instances like that with out hestitating.
On this article, we’ve taken a major step ahead in our journey to create a complete chat-based interface for managing small companies utilizing Massive Language Fashions.
By establishing our database schema, defining core functionalities, and structuring our challenge repository, we’ve laid a stable basis for the event of our software.
We began by designing our database fashions utilizing SQLModel, which allowed us to seamlessly combine with Pydantic and SQLAlchemy. This method ensures environment friendly information validation and database operations whereas minimizing the danger of SQL injection assaults.
We then proceeded to replace our Device
class to deal with SQLModel situations and enhance the validation course of. Subsequent, we carried out SQL instruments for including information to our database tables and querying information utilizing pure language instructions. By leveraging the facility of SQLModel and Pydantic, we had been capable of create a sturdy and versatile system that may deal with a variety of person inputs and generate correct SQL queries.
We configured our OpenAIAgent to offer context-aware software utilization by updating the agent class to deal with variable context within the system immediate and person immediate. This enables our agent to grasp the accessible tables and their schemas, enabling extra correct and environment friendly software utilization. Whereas we’ve made important progress, there’s nonetheless rather more to discover and implement.
To additional improve our chatbot, we launched the TaskAgent class, which serves an analogous performance because the Device class. The TaskAgent permits us to outline a reputation, description, and enter mannequin for every agent, automating the initialization course of.
Lastly, we examined our routing and subagents by defining subagents for querying information, including bills, including income. We demonstrated how the brokers deal with tax charges and calculate internet or gross quantities mechanically, showcasing the reasoning capabilities of our subagents.
Subsequent steps
Within the subsequent a part of this collection, we’ll deal with enhancing our agent’s capabilities by including assist for extra instruments and probably testing Claude as a brand new default language mannequin. We’ll additionally discover integrating our software with widespread communication platforms (WhatsApp) to make it much more accessible and user-friendly.
As we proceed to refine and develop our software, the probabilities are countless. By leveraging the facility of Massive Language Fashions and creating intuitive chat-based interfaces, we are able to revolutionize the way in which small companies handle their information and streamline their operations. Keep tuned for the subsequent installment on this thrilling collection!
Moreover, your complete supply code for the tasks lined is accessible on GitHub. You’ll be able to entry it at https://github.com/elokus/ArticleParte2.
[ad_2]