11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196 | class WorkflowHandler(ABC):
def __init__(self):
self.total_cost = 0.0
def _get_filename(self):
# should not be forced. datafeasibility, for example, wouldn't use.
raise NotImplementedError
def _get_mime_type(self):
# should not be forced. datafeasibility, for example, wouldn't use.
raise NotImplementedError
@abstractmethod
def process(self):
raise NotImplementedError
def _update_total_cost(self, response_meta):
self.total_cost += response_meta.total_cost
def _get_db_connection(self, db_server, db_name, db_user, db_password):
"""
The function `get_db_connection` creates a database connection using the provided server, database
name, user, and password.
Args:
db_server: The `db_server` parameter refers to the server where the database is hosted. This could
be an IP address or a domain name pointing to the server where the SQL Server instance is running.
db_name: The `db_name` parameter in the `get_db_connection` function refers to the name of the
database you want to connect to on the specified database server. This parameter is used to
construct the connection string that includes information such as the database name, server details,
user credentials, and driver information for
db_user: The `db_user` parameter in the `get_db_connection` function refers to the username used
to authenticate and access the database. It is typically associated with a specific user account
that has the necessary permissions to interact with the database specified by `db_name` on the
server `db_server`.
db_password: It seems like you were about to provide information about the `db_password` parameter
but the text got cut off. Could you please provide the details or let me know how I can assist you
further with the `db_password` parameter?
Returns:
The function `get_db_connection` returns a connection object to a SQL Server database using the
provided server, database name, user, and password details.
"""
conn_str = (
"DRIVER={ODBC Driver 17 for SQL Server};SERVER="
+ db_server
+ ";DATABASE="
+ db_name
+ ";UID="
+ db_user
+ ";PWD="
+ db_password
)
return pyodbc.connect(conn_str)
def _write_to_db(
self,
app_config,
input_as_json,
submit_time,
response_time,
cost,
name_suffix="",
):
"""
The `write_to_db` function inserts data into a database table `api_interactions` with specified
columns using the provided parameters.
Args:
app_config: The `app_config` parameter is a configuration object that contains information about
the database server, database name, database user, and database password needed to establish a
connection to the database. It likely contains attributes like `DB_SERVER`, `DB_NAME`, `DB_USER`,
`DB_PASSWORD`, and `NAME
input_as_json: The `input_as_json` parameter in the `write_to_db` function is expected to be a
JSON object representing the user input data that you want to store in the database. This JSON
object should be a serializable format that can be stored in a database column, typically a string
representation of the
submit_time: Submit time is the timestamp when the request was submitted to the API. It typically
includes the date and time when the request was made.
response_time: Response time is the time taken for the server to process a request and send a
response back to the client. It is usually measured in milliseconds and indicates the efficiency of
the system in handling requests.
cost: The `cost` parameter in the `write_to_db` function represents the total cost associated with
the API interaction being recorded in the database. This cost could be related to any expenses
incurred during the interaction, such as processing fees, data storage costs, or any other relevant
expenses. It is a numerical
name_suffix: The `name_suffix` parameter in the `write_to_db` function is an optional parameter
that allows you to append a suffix to the `app_name` field in the database. This can be useful if
you need to differentiate between multiple instances of the same application in the database. If a
`name
"""
with self._get_db_connection(
db_server=app_config.DB_SERVER,
db_name=app_config.DB_NAME,
db_user=app_config.DB_USER,
db_password=app_config.DB_PASSWORD,
) as conn:
cursor = conn.cursor()
query = """
INSERT INTO api_interactions (app_name, user_input, submit_time, response_time, total_cost)
VALUES (?, ?, ?, ?, ?)
"""
cursor.execute(
query,
(
app_config.NAME + name_suffix,
input_as_json,
submit_time,
response_time,
cost,
),
)
conn.commit()
def check_content_type(self, returned_content):
# TODO: consider changing to if hasattr content
if isinstance(returned_content, AIMessage):
extracted_content = returned_content.content
if isinstance(returned_content, str):
extracted_content = returned_content
else:
raise TypeError(
"Content not of type AIMessage or str. Check what invoke is returning. Langchain interfaces are inconsistent per API provider."
)
return extracted_content
# TODO is there a way to make this cleaner since self.promtpy_path and self._validate are only called in grandchildren
def load_prompty(self):
# self.prompty_path initialized by child
# This should likely be broken up more with to isolate functionality further
if not self.prompty_path.exists():
raise FileNotFoundError(f"Prompty file not found at: {self.prompty_path}")
with open(self.prompty_path, "r") as f:
prompty_content = f.read()
prompty_data = list(yaml.safe_load_all(prompty_content))
if not prompty_data or len(prompty_data) < 2:
raise ValueError("Invalid prompty file format.")
prompt_section = prompty_data[1]
prompt_template = prompt_section.get("prompt", {}).get("template", None)
if prompt_template is None:
raise ValueError("Prompt template not found in prompty file.")
self._validate_prompt_template(prompt_template)
return ChatPromptTemplate.from_template(prompt_template, template_format="jinja2")
def log_to_database(
self, app_config, content_to_log, start, finish, background_tasks, label=""
):
"""
This Python function logs content to a database using background tasks and handles KeyError
exceptions.
Args:
app_config: The `app_config` parameter likely contains configuration settings for the application,
such as database connection details, API keys, or other settings needed for logging to the database.
content_to_log: The `content_to_log` parameter typically refers to the data or information that
you want to log into the database. This could be any relevant information that you want to store for
later analysis or reference. It could be a string, a dictionary, a list, or any other data structure
that you find
start: The `start` parameter in the `log_to_database` method likely represents the start time of
the logging operation. It is used to indicate when the logging process began.
finish: Finish is a parameter representing the time when the logging process finishes. It is used
in the `log_to_database` method to log information to a database along with other parameters such as
`app_config`, `content_to_log`, `start`, `background_tasks`, and an optional `label`.
background_tasks: The `background_tasks` parameter in the `log_to_database` method seems to be an
instance of some class that allows you to add tasks to be executed in the background. In this
method, a task is added to write data to a database in the background using the `_write_to_db`
method
label: The `label` parameter in the `log_to_database` method is a string that can be used to
provide additional information or context for the log entry being written to the database. It is an
optional parameter with a default value of an empty string. If provided, the `label` will be
included
"""
try:
background_tasks.add_task(
self._write_to_db,
app_config,
content_to_log,
start,
finish,
self.total_cost,
label,
)
except KeyError:
raise KeyError(
"Failed writing to database. Check interface configuration and try again."
)
|