Spaces:
Sleeping
Sleeping
File size: 8,323 Bytes
dec332b |
1 2 3 4 5 6 7 8 9 10 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 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 |
import json
import random
from typing import Dict, List, Optional
import pandas as pd
import streamlit as st
from trulens_eval.app import ComponentView
from trulens_eval.keys import REDACTED_VALUE
from trulens_eval.keys import should_redact_key
from trulens_eval.schema import Metadata
from trulens_eval.schema import Record
from trulens_eval.schema import RecordAppCall
from trulens_eval.schema import Select
from trulens_eval.utils.containers import is_empty
from trulens_eval.utils.json import jsonify
from trulens_eval.utils.pyschema import CLASS_INFO
from trulens_eval.utils.pyschema import is_noserio
from trulens_eval.utils.serial import GetItemOrAttribute
from trulens_eval.utils.serial import JSON_BASES
from trulens_eval.utils.serial import Lens
def write_or_json(st, obj):
"""
Dispatch either st.json or st.write depending on content of `obj`. If it is
a string that can parses into strictly json (dict), use st.json, otherwise
use st.write.
"""
if isinstance(obj, str):
try:
content = json.loads(obj)
if not isinstance(content, str):
st.json(content)
else:
st.write(content)
except BaseException:
st.write(obj)
def copy_to_clipboard(path, *args, **kwargs):
st.session_state.clipboard = str(path)
def draw_selector_button(path) -> None:
st.button(
key=str(random.random()),
label=f"{Select.render_for_dashboard(path)}",
on_click=copy_to_clipboard,
args=(path,)
)
def render_selector_markdown(path) -> str:
return f"[`{Select.render_for_dashboard(path)}`]"
def render_call_frame(frame: RecordAppCall, path=None) -> str: # markdown
path = path or frame.path
return (
f"__{frame.method.name}__ (__{frame.method.obj.cls.module.module_name}.{frame.method.obj.cls.name}__)"
)
def dict_to_md(dictionary: dict) -> str:
if len(dictionary) == 0:
return "No metadata."
mdheader = "|"
mdseparator = "|"
mdbody = "|"
for key, value in dictionary.items():
mdheader = mdheader + str(key) + "|"
mdseparator = mdseparator + "-------|"
mdbody = mdbody + str(value) + "|"
mdtext = mdheader + "\n" + mdseparator + "\n" + mdbody
return mdtext
def draw_metadata(metadata: Metadata) -> str:
if isinstance(metadata, Dict):
return dict_to_md(metadata)
else:
return str(metadata)
def draw_call(call: RecordAppCall) -> None:
top = call.stack[-1]
path = Select.for_record(
top.path._append(
step=GetItemOrAttribute(item_or_attribute=top.method.name)
)
)
with st.expander(label=f"Call " + render_call_frame(top, path=path) + " " +
render_selector_markdown(path)):
args = call.args
rets = call.rets
for frame in call.stack[::-1][1:]:
st.write("Via " + render_call_frame(frame, path=path))
st.subheader(f"Inputs {render_selector_markdown(path.args)}")
if isinstance(args, Dict):
st.json(args)
else:
st.write(args)
st.subheader(f"Outputs {render_selector_markdown(path.rets)}")
if isinstance(rets, Dict):
st.json(rets)
else:
st.write(rets)
def draw_calls(record: Record, index: int) -> None:
"""
Draw the calls recorded in a `record`.
"""
calls = record.calls
app_step = 0
for call in calls:
app_step += 1
if app_step != index:
continue
draw_call(call)
def draw_prompt_info(query: Lens, component: ComponentView) -> None:
prompt_details_json = jsonify(component.json, skip_specials=True)
st.caption(f"Prompt details")
path = Select.for_app(query)
prompt_types = {
k: v for k, v in prompt_details_json.items() if (v is not None) and
not is_empty(v) and not is_noserio(v) and k != CLASS_INFO
}
for key, value in prompt_types.items():
with st.expander(key.capitalize() + " " +
render_selector_markdown(getattr(path, key)),
expanded=True):
if isinstance(value, (Dict, List)):
st.write(value)
else:
if isinstance(value, str) and len(value) > 32:
st.text(value)
else:
st.write(value)
def draw_llm_info(query: Lens, component: ComponentView) -> None:
llm_details_json = component.json
st.subheader(f"*LLM Details*")
# path_str = str(query)
# st.text(path_str[:-4])
llm_kv = {
k: v for k, v in llm_details_json.items() if (v is not None) and
not is_empty(v) and not is_noserio(v) and k != CLASS_INFO
}
# CSS to inject contained in a string
hide_table_row_index = """
<style>
thead tr th:first-child {display:none}
tbody th {display:none}
</style>
"""
df = pd.DataFrame.from_dict(llm_kv, orient='index').transpose()
# Redact any column whose name indicates it might be a secret.
for col in df.columns:
if should_redact_key(col):
df[col] = REDACTED_VALUE
# TODO: What about columns not indicating a secret but some values do
# indicate it as per `should_redact_value` ?
# Iterate over each column of the DataFrame
for column in df.columns:
path = getattr(Select.for_app(query), str(column))
# Check if any cell in the column is a dictionary
if any(isinstance(cell, dict) for cell in df[column]):
# Create new columns for each key in the dictionary
new_columns = df[column].apply(
lambda x: pd.Series(x) if isinstance(x, dict) else pd.Series()
)
new_columns.columns = [
f"{key} {render_selector_markdown(path)}"
for key in new_columns.columns
]
# Remove extra zeros after the decimal point
new_columns = new_columns.applymap(
lambda x: '{0:g}'.format(x) if isinstance(x, float) else x
)
# Add the new columns to the original DataFrame
df = pd.concat([df.drop(column, axis=1), new_columns], axis=1)
else:
# TODO: add selectors to the output here
pass
# Inject CSS with Markdown
st.markdown(hide_table_row_index, unsafe_allow_html=True)
st.table(df)
def draw_agent_info(query: Lens, component: ComponentView) -> None:
# copy of draw_prompt_info
# TODO: dedup
prompt_details_json = jsonify(component.json, skip_specials=True)
st.subheader(f"*Agent Details*")
path = Select.for_app(query)
prompt_types = {
k: v for k, v in prompt_details_json.items() if (v is not None) and
not is_empty(v) and not is_noserio(v) and k != CLASS_INFO
}
for key, value in prompt_types.items():
with st.expander(key.capitalize() + " " +
render_selector_markdown(getattr(path, key)),
expanded=True):
if isinstance(value, (Dict, List)):
st.write(value)
else:
if isinstance(value, str) and len(value) > 32:
st.text(value)
else:
st.write(value)
def draw_tool_info(query: Lens, component: ComponentView) -> None:
# copy of draw_prompt_info
# TODO: dedup
prompt_details_json = jsonify(component.json, skip_specials=True)
st.subheader(f"*Tool Details*")
path = Select.for_app(query)
prompt_types = {
k: v for k, v in prompt_details_json.items() if (v is not None) and
not is_empty(v) and not is_noserio(v) and k != CLASS_INFO
}
for key, value in prompt_types.items():
with st.expander(key.capitalize() + " " +
render_selector_markdown(getattr(path, key)),
expanded=True):
if isinstance(value, (Dict, List)):
st.write(value)
else:
if isinstance(value, str) and len(value) > 32:
st.text(value)
else:
st.write(value)
|