Files
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
chore: import upstream snapshot with attribution
2026-07-13 13:39:52 +08:00

70 lines
1.8 KiB
Python

"""
Sample script to test the Basic Chat MAF workflow.
Run:
python test_chat_flow.py
"""
import asyncio
from chat_flow import ChatInput, create_workflow
async def main():
# Test 1: Single-turn — no chat history
print("=" * 60)
print("Test 1: Single-turn (no history)")
print("=" * 60)
workflow = create_workflow()
result = await workflow.run(ChatInput(question="What is ChatGPT?"))
print("Q: What is ChatGPT?")
print(f"A: {result.get_outputs()[0]}\n")
# Test 2: Multi-turn — with one prior exchange
print("=" * 60)
print("Test 2: Multi-turn (with history)")
print("=" * 60)
history = [
{
"inputs": {"question": "What is ChatGPT?"},
"outputs": {"answer": "ChatGPT is a large language model chatbot developed by OpenAI."},
}
]
workflow = create_workflow()
result = await workflow.run(
ChatInput(
question="How is it different from GPT-4?",
chat_history=history,
)
)
print("Q: How is it different from GPT-4?")
print(f"A: {result.get_outputs()[0]}\n")
# Test 3: Multi-turn — longer conversation
print("=" * 60)
print("Test 3: Multi-turn (longer conversation)")
print("=" * 60)
history = [
{
"inputs": {"question": "What is 2+2?"},
"outputs": {"answer": "4"},
},
{
"inputs": {"question": "Multiply that by 3"},
"outputs": {"answer": "12"},
},
]
workflow = create_workflow()
result = await workflow.run(
ChatInput(
question="Now divide by 6",
chat_history=history,
)
)
print("Q: Now divide by 6")
print(f"A: {result.get_outputs()[0]}\n")
if __name__ == "__main__":
asyncio.run(main())