nemo_voice_agent.evaluation.tools.tau2_telecom_tools

View as Markdown

Tau2-telecom agent-side LLM tool ports.

Mirrors upstream TelecomTools. Tools operate on the agent-side TelecomDB dict stored in shared_state["db"]. Reads return data; writes mutate state["db"] and append an entry to state["actions"] via WriteScenarioTool._record_action.

Conventions match tau2_airline_tools / tau2_retail_tools: _Tau2InvokeMixin handles sync invoke + async _execute, _Tau2TelecomAgentReadTool / _Tau2TelecomAgentWriteTool are the base classes, and every tool declares a snake_case name class attribute that drives the registry key + LLM-visible function name + gold-replay dispatch key (matching how upstream policy.md references tools).

Module Contents

Classes

Functions

NameDescription
_bills_awaiting_payment-
_customer_to_user_viewReturn a JSON-safe shallow copy of a customer dict for tool output.
_db_not_initialized-
_find_bill-
_find_customer-
_find_device-
_find_line-
_find_line_by_phone-
_find_plan-
_get_target_lineResolve a line + verify it belongs to the customer. Raises ValueError on miss.
_get_todayReturn today as a date object.

Data

TAU2_TELECOM_AGENT_ACTION_TYPES

TAU2_TELECOM_AGENT_TOOL_NAME_TO_CLASS

API

class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.DisableRoamingTool()

Bases: _Tau2TelecomAgentWriteTool

DESCRIPTION
= 'Disables international roaming on a line.'
name
= 'disable_roaming'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.DisableRoamingTool._do_work(
p
) -> str
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.EnableRoamingTool()

Bases: _Tau2TelecomAgentWriteTool

DESCRIPTION
= 'Enables international roaming on a line.'
name
= 'enable_roaming'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.EnableRoamingTool._do_work(
p
) -> str
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetBillsForCustomerTool()

Bases: _Tau2TelecomAgentReadTool

DESCRIPTION
name
= 'get_bills_for_customer'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetBillsForCustomerTool._do_work(
p
) -> typing.List[dict]
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetCustomerByIdTool()

Bases: _Tau2TelecomAgentReadTool

DESCRIPTION
name
= 'get_customer_by_id'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetCustomerByIdTool._do_work(
p
) -> dict
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetCustomerByNameTool()

Bases: _Tau2TelecomAgentReadTool

DESCRIPTION
name
= 'get_customer_by_name'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetCustomerByNameTool._do_work(
p
) -> typing.List[dict]
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetCustomerByPhoneTool()

Bases: _Tau2TelecomAgentReadTool

DESCRIPTION
name
= 'get_customer_by_phone'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetCustomerByPhoneTool._do_work(
p
) -> dict
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetDataUsageTool()

Bases: _Tau2TelecomAgentReadTool

DESCRIPTION
name
= 'get_data_usage'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetDataUsageTool._do_work(
p
) -> dict
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetDetailsByIdTool()

Bases: _Tau2TelecomAgentReadTool

DESCRIPTION
name
= 'get_details_by_id'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.GetDetailsByIdTool._do_work(
p
) -> dict
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.RefuelDataTool()

Bases: _Tau2TelecomAgentWriteTool

DESCRIPTION
name
= 'refuel_data'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.RefuelDataTool._apply_one_time_charge(
db: dict,
customer_id: str,
amount: float,
description: str
) -> None
staticmethod

Mirrors upstream TelecomTools._apply_one_time_charge.

Finds an existing DRAFT bill for the customer, or creates one. Appends a charge line item and updates total_due. Does NOT mint a stable bill_id when creating — upstream uses uuid4().hex[:8], which is non-deterministic; we use the same shape so the hash comparison sees an extra-bill structural change but not a deterministic ID match. Scenarios that exercise refuel_data should not depend on the new bill_id text.

nemo_voice_agent.evaluation.tools.tau2_telecom_tools.RefuelDataTool._do_work(
p
) -> dict
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.ResumeLineTool()

Bases: _Tau2TelecomAgentWriteTool

DESCRIPTION
name
= 'resume_line'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.ResumeLineTool._do_work(
p
) -> dict
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.SendPaymentRequestTool()

Bases: _Tau2TelecomAgentWriteTool

DESCRIPTION
name
= 'send_payment_request'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.SendPaymentRequestTool._do_work(
p
) -> str
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.SuspendLineTool()

Bases: _Tau2TelecomAgentWriteTool

DESCRIPTION
name
= 'suspend_line'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.SuspendLineTool._do_work(
p
) -> dict
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools.TransferToHumanAgentsTool()

Bases: _Tau2TelecomAgentWriteTool

DESCRIPTION
name
= 'transfer_to_human_agents'
properties
Dict[str, Any]
required_properties
List[str]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.TransferToHumanAgentsTool._after_result(
params: pipecat.services.llm_service.FunctionCallParams
) -> None
async

Emit <exit> only after the transfer result has been delivered.

nemo_voice_agent.evaluation.tools.tau2_telecom_tools.TransferToHumanAgentsTool._do_work(
p
) -> str
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.TransferToHumanAgentsTool._execute(
kwargs = {}
) -> typing.Any
async
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools._Tau2InvokeMixin()

Provides sync invoke(**kwargs) + async _execute(**kwargs) routing.

Same shape as _Tau2InvokeMixin in airline / retail / user-side telecom files. Subclasses implement _do_work(p).

PARAMS_MODEL
Optional[Type[BaseModel]] = None
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._Tau2InvokeMixin._do_work(
p
) -> typing.Any
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._Tau2InvokeMixin._execute(
kwargs = {}
) -> typing.Any
async

Run the tool and return its result (pure — never delivers).

StandardSchemaTool.__call__ owns delivery and applies the empty-result normalization centrally, so a falsy result (e.g. an empty match list) can’t be masked as the literal “COMPLETED”.

nemo_voice_agent.evaluation.tools.tau2_telecom_tools._Tau2InvokeMixin.invoke(
kwargs = {}
) -> typing.Any
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools._Tau2TelecomAgentReadTool(
shared_state: typing.Optional[dict] = None,
description: typing.Optional[str] = None
)

Bases: _Tau2InvokeMixin, StandardSchemaTool

Base for telecom agent-side read-only tools.

DESCRIPTION
str = ''
state
= shared_state if shared_state is not None else {}
class nemo_voice_agent.evaluation.tools.tau2_telecom_tools._Tau2TelecomAgentWriteTool(
shared_state: typing.Optional[dict] = None,
description: typing.Optional[str] = None
)

Bases: _Tau2InvokeMixin, WriteScenarioTool

Base for telecom agent-side write tools.

DESCRIPTION
str = ''
state
= shared_state if shared_state is not None else {}
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._bills_awaiting_payment(
db: dict,
customer: dict
) -> typing.List[dict]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._customer_to_user_view(
customer: dict
) -> dict

Return a JSON-safe shallow copy of a customer dict for tool output.

Same shape as upstream’s Customer schema (already dict, just defensive copy so the caller can’t accidentally mutate state["db"] through the result).

nemo_voice_agent.evaluation.tools.tau2_telecom_tools._db_not_initialized() -> dict
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._find_bill(
db: dict,
bill_id: str
) -> typing.Optional[dict]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._find_customer(
db: dict,
customer_id: str
) -> typing.Optional[dict]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._find_device(
db: dict,
device_id: str
) -> typing.Optional[dict]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._find_line(
db: dict,
line_id: str
) -> typing.Optional[dict]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._find_line_by_phone(
db: dict,
phone_number: str
) -> typing.Optional[dict]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._find_plan(
db: dict,
plan_id: str
) -> typing.Optional[dict]
nemo_voice_agent.evaluation.tools.tau2_telecom_tools._get_target_line(
db: dict,
customer_id: str,
line_id: str
) -> dict

Resolve a line + verify it belongs to the customer. Raises ValueError on miss.

nemo_voice_agent.evaluation.tools.tau2_telecom_tools._get_today(
scenario_today: typing.Optional[str] = None
) -> datetime.date

Return today as a date object.

scenario_today (when supplied) is the scenario’s authored anchor date — used for tasks whose timing assumptions depend on a fixed “now” (e.g., bill due dates relative to a current month). Falls back to today when not supplied.

nemo_voice_agent.evaluation.tools.tau2_telecom_tools.TAU2_TELECOM_AGENT_ACTION_TYPES: List[str] = ['suspend_line', 'resume_line', 'enable_roaming', 'disable_roaming', 'send_payme...
nemo_voice_agent.evaluation.tools.tau2_telecom_tools.TAU2_TELECOM_AGENT_TOOL_NAME_TO_CLASS: Dict[str, Type[StandardSchemaTool]] = {(cls.name): cls for cls in (GetCustomerByPhoneTool, GetCustomerByIdTool, GetCus...