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
|
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
"""
FILE: sample_analyze_healthcare_action_async.py
DESCRIPTION:
This sample demonstrates how to submit a collection of text documents for analysis, which uses the
AnalyzeHealthcareEntitiesAction (plus FHIR feature) and RecognizePiiEntitiesAction to recognize healthcare entities,
along with any PII entities.
The response will contain results from each of the individual actions specified in the request.
USAGE:
python sample_analyze_healthcare_action_async.py
Set the environment variables with your own values before running the sample:
1) AZURE_LANGUAGE_ENDPOINT - the endpoint to your Language resource.
2) AZURE_LANGUAGE_KEY - your Language subscription key
"""
import asyncio
async def sample_analyze_healthcare_action() -> None:
import os
from azure.core.credentials import AzureKeyCredential
from azure.ai.textanalytics.aio import TextAnalyticsClient
from azure.ai.textanalytics import (
AnalyzeHealthcareEntitiesAction,
RecognizePiiEntitiesAction,
)
endpoint = os.environ["AZURE_LANGUAGE_ENDPOINT"]
key = os.environ["AZURE_LANGUAGE_KEY"]
text_analytics_client = TextAnalyticsClient(
endpoint=endpoint,
credential=AzureKeyCredential(key),
)
documents = [
"""
Patient needs to take 100 mg of ibuprofen, and 3 mg of potassium. Also needs to take
10 mg of Zocor.
""",
"""
Patient needs to take 50 mg of ibuprofen, and 2 mg of Coumadin.
"""
]
async with text_analytics_client:
poller = await text_analytics_client.begin_analyze_actions(
documents,
display_name="Sample Text Analysis",
actions=[
AnalyzeHealthcareEntitiesAction(),
RecognizePiiEntitiesAction(domain_filter="phi"),
],
)
pages = await poller.result()
# To enumerate / zip for async, unless you install a third party library,
# you have to read in all of the elements into memory first.
# If you're not looking to enumerate / zip, we recommend you just asynchronously
# loop over it immediately, without going through this step of reading them into memory
document_results = []
async for page in pages:
document_results.append(page)
for doc, action_results in zip(documents, document_results):
print(f"\nDocument text: {doc}")
for result in action_results:
if result.kind == "Healthcare":
print("...Results of Analyze Healthcare Entities Action:")
for entity in result.entities:
print(f"Entity: {entity.text}")
print(f"...Normalized Text: {entity.normalized_text}")
print(f"...Category: {entity.category}")
print(f"...Subcategory: {entity.subcategory}")
print(f"...Offset: {entity.offset}")
print(f"...Confidence score: {entity.confidence_score}")
if entity.data_sources is not None:
print("...Data Sources:")
for data_source in entity.data_sources:
print(f"......Entity ID: {data_source.entity_id}")
print(f"......Name: {data_source.name}")
if entity.assertion is not None:
print("...Assertion:")
print(f"......Conditionality: {entity.assertion.conditionality}")
print(f"......Certainty: {entity.assertion.certainty}")
print(f"......Association: {entity.assertion.association}")
for relation in result.entity_relations:
print(f"Relation of type: {relation.relation_type} has the following roles")
for role in relation.roles:
print(f"...Role '{role.name}' with entity '{role.entity.text}'")
elif result.kind == "PiiEntityRecognition":
print("Results of Recognize PII Entities action:")
for pii_entity in result.entities:
print(f"......Entity: {pii_entity.text}")
print(f".........Category: {pii_entity.category}")
print(f".........Confidence Score: {pii_entity.confidence_score}")
elif result.is_error is True:
print(f"...Is an error with code '{result.error.code}' and message '{result.error.message}'")
print("------------------------------------------")
async def main():
await sample_analyze_healthcare_action()
if __name__ == '__main__':
asyncio.run(main())
|
