mirror of
https://github.com/vee1e/InstaFuel_Chatbot_public.git
synced 2026-09-01 18:57:24 +00:00
53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""Smoke test script for the InstaFuel chatbot.
|
|
|
|
Runs a short conversation through the orchestrator without requiring
|
|
Google Cloud connectivity. Useful for verifying that the local agent
|
|
stack is wired correctly after installation.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.main import bootstrap_system
|
|
from src.models.schemas import AgentType, Message, MessageType, Platform
|
|
|
|
|
|
async def run_smoke_test() -> None:
|
|
orchestrator, shared_memory = await bootstrap_system(enable_cloud=False)
|
|
|
|
try:
|
|
conversation_id = f"smoke-{uuid.uuid4()}"
|
|
user_message = Message(
|
|
id=str(uuid.uuid4()),
|
|
conversation_id=conversation_id,
|
|
sender_id="smoke-tester",
|
|
sender_type=AgentType.ORCHESTRATOR,
|
|
message_type=MessageType.USER_MESSAGE,
|
|
content="Hey! I'm trying to build muscle. What should I take?",
|
|
platform=Platform.WEBSITE,
|
|
)
|
|
|
|
response = await orchestrator.process_message(user_message, None)
|
|
|
|
print("Agent:", response.agent_type.value)
|
|
print("Response:\n", response.response_text)
|
|
|
|
if response.metadata:
|
|
print("\nMetadata:")
|
|
for key, value in response.metadata.items():
|
|
print(f" {key}: {value}")
|
|
|
|
finally:
|
|
await orchestrator.stop()
|
|
for agent in shared_memory.agents.values():
|
|
await agent.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_smoke_test())
|