wehub-resource-sync
e768098d0e
tools_continuous_delivery / Private PyPI non-main branch release (push) Has been skipped
tools_continuous_delivery / Private PyPI main branch release (push) Failing after 2m42s
Publish Promptflow Doc / Build (push) Has been cancelled
Publish Promptflow Doc / Deploy (push) Has been cancelled
Flake8 Lint / flake8 (push) Has been cancelled
Spell check CI / Spell_Check (push) Has been cancelled
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
import asyncio
|
||
from workflow import create_workflow
|
||
|
||
|
||
async def main():
|
||
"""Run the workflow with multiple queries to demonstrate the switch logic."""
|
||
|
||
# Test cases targeting each switch condition
|
||
test_cases = [
|
||
("When will my order be shipped?", "order_search"),
|
||
("Where is my order?", "order_search"),
|
||
("Can you track my package?", "order_search"),
|
||
("Tell me about this product", "product_info"),
|
||
("What is the specification?", "product_info"),
|
||
("Who manufactures this?", "product_info"),
|
||
("Recommend a good product", "product_recommendation"),
|
||
("What products do you suggest?", "product_recommendation"),
|
||
("Give me product recommendations", "product_recommendation"),
|
||
("Something completely random query", "default/unknown"),
|
||
]
|
||
|
||
print("=" * 80)
|
||
print("Testing Conditional Switch Workflow")
|
||
print("=" * 80)
|
||
print()
|
||
|
||
results_by_intention = {}
|
||
|
||
for i, (query, expected_intention) in enumerate(test_cases, 1):
|
||
print(f"Test {i:2d}: {query}")
|
||
print(f" Expected intention: {expected_intention}")
|
||
|
||
workflow = create_workflow()
|
||
result = await workflow.run(query)
|
||
response = result.get_outputs()[0]
|
||
print(f" Response: {response}")
|
||
|
||
# Track results
|
||
if expected_intention not in results_by_intention:
|
||
results_by_intention[expected_intention] = []
|
||
results_by_intention[expected_intention].append(response)
|
||
|
||
print()
|
||
|
||
print("=" * 80)
|
||
print("Summary of Switch Paths:")
|
||
print("=" * 80)
|
||
|
||
intention_labels = {
|
||
"order_search": "📦 Order Search",
|
||
"product_info": "ℹ️ Product Info",
|
||
"product_recommendation": "⭐ Product Recommendation",
|
||
"default/unknown": "❓ Default/Unknown",
|
||
}
|
||
|
||
for intention, label in intention_labels.items():
|
||
if intention in results_by_intention:
|
||
count = len(results_by_intention[intention])
|
||
print(f"{label}: {count} matched")
|
||
for response in results_by_intention[intention]:
|
||
print(f" → {response}")
|
||
else:
|
||
print(f"{label}: 0 matched")
|
||
|
||
print("=" * 80)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|