@unify.traced
def call_agent(
example_id,
subq_system_message,
mark_system_message,
question_num,
question,
sub_questions,
markscheme,
answer,
available_marks,
):
subq_agents = {k: agent.copy() for k in markscheme.keys()}
with_subqs = len(markscheme) > 1
response_formats = {k: MarksAndReasoning for k, v in markscheme.items()}
[
agnt.set_response_format(rf)
for agnt, rf in zip(
subq_agents.values(),
response_formats.values(),
)
]
mark_sys_msgs = list()
parsed_markschemes = list()
for i, k in enumerate(markscheme.keys()):
parsed_markscheme = parse_marks_from_markscheme(
f"_{k}" if k != "_" else "",
markscheme[k],
)
parsed_markschemes.append(parsed_markscheme)
this_markscheme = markscheme[k]
for i, (mark, chunk) in enumerate(parsed_markscheme):
this_markscheme = this_markscheme.replace(
chunk,
chunk.replace(
mark,
f"{mark}({len([m for m, _ in parsed_markscheme[0:i] if m == mark])})",
),
)
subq_agents[k].set_system_message(
subq_system_message.replace(
"{subq}",
k.replace("_", str(question_num)),
)
.replace(
"{question}",
textwrap.indent(question, " " * 4),
)
.replace(
"{subquestion}",
textwrap.indent(sub_questions[k], " " * 4),
)
.replace(
"{markscheme}",
textwrap.indent(this_markscheme, " " * 4),
)
.replace(
"{mark_types_explanation}",
textwrap.indent(
extract_mark_type_explanation(
f"_{k}" if k != "_" else "",
markscheme[k],
),
" " * 4,
),
)
.replace(
"{answer}",
textwrap.indent(answer[k], " " * 4),
)
.replace(
"{available_marks}",
str(available_marks[k.replace("_", "total")]),
)
.replace(
"{output_response_explanation}",
output_response_explanation,
)
.replace(
"{prior_context}",
(
(
prior_context_exp
+ pretty_print_dict(
{
k: {
"sub-question": sub_questions[k],
"markscheme": markscheme[k],
"answer": answer[k],
}
for k in list(sub_questions.keys())[0:i]
},
indent=4,
)
)
if with_subqs and i > 0
else ""
),
),
)
mark_sys_msgs.append(
mark_system_message.replace(
"{subq}",
k.replace("_", str(question_num)),
)
.replace(
"{question}",
textwrap.indent(question, " " * 4),
)
.replace(
"{subquestion}",
textwrap.indent(sub_questions[k], " " * 4),
)
.replace(
"{answer}",
textwrap.indent(answer[k], " " * 4),
)
.replace(
"{prior_context}",
(
(
prior_context_exp
+ pretty_print_dict(
{
k: {
"sub-question": sub_questions[k],
"markscheme": markscheme[k],
"answer": answer[k],
}
for k in list(sub_questions.keys())[0:i]
},
indent=4,
)
)
if with_subqs and i > 0
else ""
),
),
)
rets = unify.map(
lambda *a: call_subq_agent(example_id, *a),
list(sub_questions.keys()),
list(subq_agents.values()),
list(markscheme.values()),
parsed_markschemes,
mark_sys_msgs,
from_args=True,
name=f"Evals[{example_id}]->SubQAgent",
)
return dict(zip(markscheme.keys(), rets))