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
|
# coding=utf-8
# ------------------------------------
# Copyright (c) Microsoft.
# Licensed under the MIT License.
# ------------------------------------
"""
FILE: sample_text_custom_single_label_classification_async.py
DESCRIPTION:
This sample demonstrates how to run a **custom single-label classification** action over text (async LRO).
USAGE:
python sample_text_custom_single_label_classification_async.py
REQUIRED ENV VARS (for AAD / DefaultAzureCredential):
AZURE_TEXT_ENDPOINT
AZURE_CLIENT_ID
AZURE_TENANT_ID
AZURE_CLIENT_SECRET
NOTE:
If you want to use AzureKeyCredential instead, set:
- AZURE_TEXT_ENDPOINT
- AZURE_TEXT_KEY
OPTIONAL ENV VARS:
PROJECT_NAME # defaults to "<project-name>"
DEPLOYMENT_NAME # defaults to "<deployment-name>"
"""
# [START text_custom_single_label_classification_async]
import os
import asyncio
from azure.identity.aio import DefaultAzureCredential
from azure.ai.textanalytics.aio import TextAnalysisClient
from azure.ai.textanalytics.models import (
MultiLanguageTextInput,
MultiLanguageInput,
CustomSingleLabelClassificationActionContent,
CustomSingleLabelClassificationOperationAction,
CustomSingleLabelClassificationOperationResult,
)
async def sample_text_custom_single_label_classification_async():
# get settings
endpoint = os.environ["AZURE_TEXT_ENDPOINT"]
project_name = os.environ.get("PROJECT_NAME", "<project-name>")
deployment_name = os.environ.get("DEPLOYMENT_NAME", "<deployment-name>")
credential = DefaultAzureCredential()
async with TextAnalysisClient(endpoint, credential=credential) as client:
# Build input
text_a = (
"I need a reservation for an indoor restaurant in China. Please don't stop the music. "
"Play music and add it to my playlist."
)
text_input = MultiLanguageTextInput(
multi_language_inputs=[MultiLanguageInput(id="A", text=text_a, language="en")]
)
action = CustomSingleLabelClassificationOperationAction(
name="Custom Single-Label Classification",
action_content=CustomSingleLabelClassificationActionContent(
project_name=project_name,
deployment_name=deployment_name,
),
)
# Start long-running operation (async)
poller = await client.begin_analyze_text_job(
text_input=text_input,
actions=[action],
)
# Operation metadata (pre-final)
print(f"Operation ID: {poller.details.get('operation_id')}")
# Wait for completion and get AsyncItemPaged of TextActions
paged_actions = await poller.result()
# Final-state metadata
d = poller.details
print(f"Job ID: {d.get('job_id')}")
print(f"Status: {d.get('status')}")
print(f"Created: {d.get('created_date_time')}")
print(f"Last Updated: {d.get('last_updated_date_time')}")
if d.get("expiration_date_time"):
print(f"Expires: {d.get('expiration_date_time')}")
if d.get("display_name"):
print(f"Display Name: {d.get('display_name')}")
if d.get("errors"):
print("\nErrors:")
for err in d["errors"]:
print(f" Code: {err.code} - {err.message}")
# Iterate results (async pageable)
async for actions_page in paged_actions:
# Page-level counts (if available)
print(
f"Completed: {actions_page.completed}, "
f"In Progress: {actions_page.in_progress}, "
f"Failed: {actions_page.failed}, "
f"Total: {actions_page.total}"
)
# Items are the individual operation results
for op_result in actions_page.items_property or []:
if isinstance(op_result, CustomSingleLabelClassificationOperationResult):
print(f"\nAction Name: {op_result.task_name}")
print(f"Action Status: {op_result.status}")
print(f"Kind: {op_result.kind}")
result = op_result.results
for doc in result.documents or []:
print(f"\nDocument ID: {doc.id}")
# Single-label: typically one class, but iterate to be general
for cls_item in doc.class_property or []:
print(f" Predicted category: {cls_item.category}")
print(f" Confidence score: {cls_item.confidence_score}")
else:
try:
print(
f"\n[Other action] name={op_result.task_name}, "
f"status={op_result.status}, kind={op_result.kind}"
)
except Exception:
print("\n[Other action present]")
# [END text_custom_single_label_classification_async]
async def main():
await sample_text_custom_single_label_classification_async()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
|
