वॉइस बॉट आर्किटेक्चर व्यापक, एंड-टू-एंड सिस्टम डिज़ाइन का प्रतिनिधित्व करता है जो रीयल-टाइम स्पीच प्रोसेसिंग (ASR/TTS), संवादात्मक ऑर्केस्ट्रेशन, टूल इंटीग्रेशन और टेलीफोनी नियंत्रण को एक सुसंगत सिस्टम में जोड़ता है जो इंसानों जैसी बातचीत करने में सक्षम है। जैसे-जैसे उपयोगकर्ताओं की अपेक्षाएं विकसित हो रही हैं, दांव पहले कभी इतने ऊंचे नहीं रहे; ग्राहक उप-सेकंड रिस्पॉन्स टाइम की मांग करते हैं, उद्यमों को पूर्ण ऑडिटेबिलिटी और अनुपालन क्षमताओं की आवश्यकता होती है, और डेवलपर्स को बड़े पैमाने पर प्रोडक्शन सिस्टम बनाने के लिए स्पष्ट, रखरखाव योग्य पैटर्न की आवश्यकता होती है।
यह गहन विश्लेषण स्मॉलेस्ट एआई (Smallest AI) के Atoms SDK के माध्यम से आधुनिक वॉइस बॉट आर्किटेक्चर के मूलभूत घटकों की पड़ताल करता है।
हम मुख्य SDK अवधारणाओं जैसे AtomsApp और AgentSession समन्वय, प्रोडक्शन-रेडी पैटर्न्स जिसमें मल्टी-नोड आर्किटेक्चर और टूल चेनिंग शामिल हैं, परफॉर्मेंस ऑप्टिमाइज़ेशन्स जो प्राकृतिक बातचीत प्रवाह को सक्षम करते हैं, और कैसे smallest.ai का रीयल-टाइम ASR/TTS इंफ्रास्ट्रक्चर उन वॉयस अनुभवों के लिए उच्च-प्रदर्शन आधार के रूप में कार्य करता है जो वास्तव में मानवीय महसूस होते हैं, की जांच करेंगे।
वॉइस बॉट आर्किटेक्चर क्या है?
अपने मूल में, एक वॉयस बॉट आर्किटेक्चर एक परिष्कृत प्रवाह को ऑर्केस्ट्रेट करता है: एक माइक्रोफ़ोन से आने वाला ऑडियो रीयल-टाइम ऑटोमैटिक स्पीच रिकग्निशन (ASR) के माध्यम से स्ट्रीम होता है, जो ट्रांसक्राइब किए गए टेक्स्ट को एक एजेंट ऑर्केस्ट्रेशन लेयर को फीड करता है जहाँ लार्ज लैंग्वेज मॉडल्स (LLMs) इरादे को समझते हैं और आवश्यकतानुसार टूल्स को इनवोक करते हैं, इससे पहले कि टेक्स्ट-टू-स्पीच (TTS) सिंथेसिस प्रतिक्रिया को वापस प्राकृतिक ऑडियो में बदलकर स्पीकर तक पहुँचाए।
इस आर्किटेक्चर में चार महत्वपूर्ण लेयर्स शामिल हैं जो मिलकर काम करती हैं:
स्पीच I/O लेयर: न्यूनतम विलंबता के लिए वेबसॉकेट कनेक्शन के साथ द्विदिश (bidirectional) ऑडियो स्ट्रीमिंग को संभालती है
सेशन और नोड मैनेजमेंट लेयर: बातचीत की स्थिति, मल्टी-नोड वर्कफ़्लोज़ और इवेंट-ड्रिवन कम्युनिकेशन को समन्वित करती है
टूल और एक्शन लेयर: उपयोगकर्ता के अनुरोधों को पूरा करने के लिए बैकएंड सिस्टम, डेटाबेस और एपीआई को एकीकृत करती है
ऑब्जर्वेबिलिटी और कंप्लायंस लेयर: ऑडिट लॉगिंग, मॉनिटरिंग और नियामक अनुपालन क्षमताएं प्रदान करती है
पारंपरिक इंटरएक्टिव वॉयस रिस्पॉन्स (IVR) सिस्टम के साथ इसका अंतर बिल्कुल स्पष्ट है। जहाँ विरासत IVR उपयोगकर्ताओं को हार्डकोडेड प्रवाह वाले मेनू-संचालित, स्टेटलेस नेविगेशन के माध्यम से मजबूर करता है, वहीं आधुनिक AI वॉयस बॉट्स इरादे-संचालित समझ का लाभ उठाते हैं, पूरी बातचीत में स्टेटफुल संदर्भ बनाए रखते हैं, और उपयोगकर्ताओं की आवश्यकताओं के अनुसार गतिशील रूप से अनुकूलित होने के लिए परिष्कृत तर्क का उपयोग करते हैं।
smallest.ai प्राकृतिक बातचीत को संभव बनाने वाले उच्च-प्रदर्शन स्पीच इंफ्रास्ट्रक्चर प्रदान करके इस इकोसिस्टम में एक महत्वपूर्ण भूमिका निभाता है। Pulse STT 4.5% इंग्लिश वर्ड एरर रेट के साथ 32 भाषाओं में 64ms टाइम-टू-फर्स्ट-ट्रांसक्रिप्ट लेटेंसी प्राप्त करता है, जबकि Lightning TTS सिर्फ 175ms लेटेंसी के साथ स्टूडियो-ग्रेड 44.1kHz ऑडियो को सिंथेसाइज करता है, जिससे उप-800ms एंड-टू-एंड टर्न टाइम्स सक्षम होता है जो वास्तव में संवादात्मक AI को परिभाषित करता है।
एक आधुनिक वॉइस बॉट के मुख्य घटक
रीयल-टाइम स्पीच इंजेक्शन (ASR)
स्ट्रीमिंग वेबसॉकेट ASR कम-विलंबता वाले वॉयस अनुभवों की नींव बनाता है। संपूर्ण कथनों की प्रतीक्षा करने के बजाय, आधुनिक ASR सिस्टम ऑडियो को वृद्धिशील रूप से संसाधित करते हैं, जैसे-जैसे बोलना जारी रहता है आंशिक ट्रांसक्रिप्ट उत्सर्जित करते हैं और प्राकृतिक भाषण सीमाओं का पता लगाने पर परिणामों को अंतिम रूप देते हैं। यह दृष्टिकोण उस महत्वपूर्ण समय सीमा में कथित विलंबता को नाटकीय रूप से कम करता है जहाँ उपयोगकर्ता यह तय करते हैं कि सिस्टम उत्तरदायी है या खराब है।
Pulse STT 64ms टाइम-टू-फर्स्ट-ट्रांसक्रिप्ट और 32 भाषाओं के समर्थन के साथ इस स्ट्रीमिंग आर्किटेक्चर का उदाहरण देता है, जो प्रोडक्शन वॉयस सिस्टम के लिए आवश्यक सटीकता और गति प्रदान करता है। हमेशा चालू रहने वाले सहायकों के लिए, वेक-वर्ड डिटेक्शन और इंटेलिजेंट एंडपॉइंटिंग जैसी क्षमताएं यह सुनिश्चित करती हैं कि सिस्टम केवल आवश्यकता होने पर ही सक्रिय हो और सटीक रूप से निर्धारित करे कि उपयोगकर्ताओं ने कब बोलना समाप्त कर दिया है।
एजेंट ऑर्केस्ट्रेशन लेयर
Atoms SDK वॉयस बॉट लॉजिक को तीन मूलभूत प्रिमिटिव्स के आसपास संरचित करता है: AtomsApp, AgentSession, और Nodes। यह डिज़ाइन रीयल-टाइम बातचीत के लिए आवश्यक कड़े समन्वय को बनाए रखते हुए चिंताओं के स्पष्ट अलगाव को सक्षम बनाता है।
यहाँ एक न्यूनतम उदाहरण दिया गया है जो AtomsApp लाइफसाइकल और setup_handler पैटर्न को दिखाता है:
import os
from smallestai.atoms.agent.nodes import OutputAgentNode
from smallestai.atoms.agent.clients.openai import OpenAIClient
from smallestai.atoms.agent.server import AtomsApp
from smallestai.atoms.agent.session import AgentSession
from smallestai.atoms.agent.events import SDKSystemUserJoinedEvent
class MyAgent(OutputAgentNode):
def __init__(self):
super().__init__(name="my-agent")
self.llm = OpenAIClient(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY")
)
self.context.add_message({
"role": "system",
"content": "You are a helpful assistant. Be concise and friendly."
})
async def generate_response(self):
response = await self.llm.chat(
messages=self.context.messages,
stream=True
)
full_response = ""
async for chunk in response:
if chunk.content:
full_response += chunk.content
yield chunk.content
if full_response:
self.context.add_message({"role": "assistant", "content": full_response})
async def on_start(session: AgentSession):
agent = MyAgent()
session.add_node(agent)
await session.start()
await session.wait_until_complete()
if __name__ == "__main__":
app = AtomsApp(setup_handler=on_start)
app.run()import os
from smallestai.atoms.agent.nodes import OutputAgentNode
from smallestai.atoms.agent.clients.openai import OpenAIClient
from smallestai.atoms.agent.server import AtomsApp
from smallestai.atoms.agent.session import AgentSession
from smallestai.atoms.agent.events import SDKSystemUserJoinedEvent
class MyAgent(OutputAgentNode):
def __init__(self):
super().__init__(name="my-agent")
self.llm = OpenAIClient(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY")
)
self.context.add_message({
"role": "system",
"content": "You are a helpful assistant. Be concise and friendly."
})
async def generate_response(self):
response = await self.llm.chat(
messages=self.context.messages,
stream=True
)
full_response = ""
async for chunk in response:
if chunk.content:
full_response += chunk.content
yield chunk.content
if full_response:
self.context.add_message({"role": "assistant", "content": full_response})
async def on_start(session: AgentSession):
agent = MyAgent()
session.add_node(agent)
await session.start()
await session.wait_until_complete()
if __name__ == "__main__":
app = AtomsApp(setup_handler=on_start)
app.run()import os
from smallestai.atoms.agent.nodes import OutputAgentNode
from smallestai.atoms.agent.clients.openai import OpenAIClient
from smallestai.atoms.agent.server import AtomsApp
from smallestai.atoms.agent.session import AgentSession
from smallestai.atoms.agent.events import SDKSystemUserJoinedEvent
class MyAgent(OutputAgentNode):
def __init__(self):
super().__init__(name="my-agent")
self.llm = OpenAIClient(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY")
)
self.context.add_message({
"role": "system",
"content": "You are a helpful assistant. Be concise and friendly."
})
async def generate_response(self):
response = await self.llm.chat(
messages=self.context.messages,
stream=True
)
full_response = ""
async for chunk in response:
if chunk.content:
full_response += chunk.content
yield chunk.content
if full_response:
self.context.add_message({"role": "assistant", "content": full_response})
async def on_start(session: AgentSession):
agent = MyAgent()
session.add_node(agent)
await session.start()
await session.wait_until_complete()
if __name__ == "__main__":
app = AtomsApp(setup_handler=on_start)
app.run()आउटपुट:
AgentSession रनटाइम कंटेनर के रूप में कार्य करता है, जो वेबसॉकेट कनेक्शन, इवेंट डिस्पैच और नोड लाइफसाइकल का प्रबंधन करता है, विस्तृत विवरण के लिए देखें त्वरित प्रारंभ मार्गदर्शिका (quickstart guide)।
प्रत्येक सेशन एक सैंडबॉक्स बनाता है जो पूर्ण अलगाव सुनिश्चित करता है, एक बातचीत से वेरिएबल्स और स्टेट कभी दूसरी बातचीत में लीक नहीं होते हैं।
नोड्स कार्यात्मक बिल्डिंग ब्लॉक्स का प्रतिनिधित्व करते हैं। OutputAgentNode संवादात्मक बातचीत को संभालता है, संदर्भ और स्थिति का प्रबंधन करते हुए उपयोगकर्ताओं को LLM प्रतिक्रियाओं को स्ट्रीम करता है। BackgroundAgentNode पृष्ठभूमि में समानांतर रूप से चुपचाप इवेंट्स को संसाधित करता है, जो बातचीत की विलंबता को प्रभावित किए बिना ऑडिट लॉगिंग, सेंटिमेंट एनालिसिस या रीयल-टाइम मॉनिटरिंग के लिए एकदम सही है।
संवादात्मक प्रवाह और टूल निष्पादन
वास्तविक दुनिया के वॉयस बॉट्स को बातचीत और कार्रवाई के बीच की दूरी को पाटना होगा। Atoms SDK का टूल सिस्टम स्वचालित खोज के साथ डेकोरेटर पैटर्न का उपयोग करता है, जिससे LLMs के लिए कॉल करने योग्य टूल के रूप में पायथन फ़ंक्शंस को प्रदर्शित करना सीधा हो जाता है:
import os
from smallestai.atoms.agent.nodes import OutputAgentNode
from smallestai.atoms.agent.clients import OpenAIClient
from smallestai.atoms.agent.server import AtomsApp
from smallestai.atoms.agent.session import AgentSession
from smallestai.atoms.agent.tools import ToolRegistry, function_tool
from smallestai.atoms.agent.clients.types import ToolCall, ToolResult
class AssistantAgent(OutputAgentNode):
def __init__(self):
super().__init__(name="assistant-agent")
self.llm = OpenAIClient(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
self.tool_registry = ToolRegistry()
self.tool_registry.discover(self)
self.tool_schemas = self.tool_registry.get_schemas()
self.context.add_message({ # added
"role": "system",
"content": "You are a helpful weather assistant. Be concise."
})
@function_tool()
def get_weather(self, city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to check weather for.
"""
return f"The weather in {city} is sunny, 72°F"
async def generate_response(self):
response = await self.llm.chat(
messages=self.context.messages,
stream=True,
tools=self.tool_schemas
)
tool_calls = []
full_response = ""
async for chunk in response:
if chunk.content:
full_response += chunk.content
yield chunk.content
if chunk.tool_calls:
tool_calls.extend(chunk.tool_calls)
if not tool_calls: # added
self.context.add_message({"role": "assistant", "content": full_response})
return
yield "One moment while I check that for you. "
results = await self.tool_registry.execute(tool_calls=tool_calls, parallel=True)
self.context.add_messages([ # added
{
"role": "assistant",
"content": "",
"tool_calls": [
{"id": tc.id, "type": "function",
"function": {"name": tc.name, "arguments": str(tc.arguments)}}
for tc in tool_calls
],
},
*[
{"role": "tool", "tool_call_id": tc.id, "content": str(result.content or "")}
for tc, result in zip(tool_calls, results)
],
])
final_response = await self.llm.chat(messages=self.context.messages, stream=True) # added
final_text = ""
async for chunk in final_response:
if chunk.content:
final_text += chunk.content
yield chunk.content
self.context.add_message({"role": "assistant", "content": final_text})
async def on_start(session: AgentSession):
agent = AssistantAgent() # fixed: was MyAgent()
session.add_node(agent)
await session.start()
await session.wait_until_complete()
if __name__ == "__main__":
app = AtomsApp(setup_handler=on_start)
app.run()import os
from smallestai.atoms.agent.nodes import OutputAgentNode
from smallestai.atoms.agent.clients import OpenAIClient
from smallestai.atoms.agent.server import AtomsApp
from smallestai.atoms.agent.session import AgentSession
from smallestai.atoms.agent.tools import ToolRegistry, function_tool
from smallestai.atoms.agent.clients.types import ToolCall, ToolResult
class AssistantAgent(OutputAgentNode):
def __init__(self):
super().__init__(name="assistant-agent")
self.llm = OpenAIClient(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
self.tool_registry = ToolRegistry()
self.tool_registry.discover(self)
self.tool_schemas = self.tool_registry.get_schemas()
self.context.add_message({ # added
"role": "system",
"content": "You are a helpful weather assistant. Be concise."
})
@function_tool()
def get_weather(self, city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to check weather for.
"""
return f"The weather in {city} is sunny, 72°F"
async def generate_response(self):
response = await self.llm.chat(
messages=self.context.messages,
stream=True,
tools=self.tool_schemas
)
tool_calls = []
full_response = ""
async for chunk in response:
if chunk.content:
full_response += chunk.content
yield chunk.content
if chunk.tool_calls:
tool_calls.extend(chunk.tool_calls)
if not tool_calls: # added
self.context.add_message({"role": "assistant", "content": full_response})
return
yield "One moment while I check that for you. "
results = await self.tool_registry.execute(tool_calls=tool_calls, parallel=True)
self.context.add_messages([ # added
{
"role": "assistant",
"content": "",
"tool_calls": [
{"id": tc.id, "type": "function",
"function": {"name": tc.name, "arguments": str(tc.arguments)}}
for tc in tool_calls
],
},
*[
{"role": "tool", "tool_call_id": tc.id, "content": str(result.content or "")}
for tc, result in zip(tool_calls, results)
],
])
final_response = await self.llm.chat(messages=self.context.messages, stream=True) # added
final_text = ""
async for chunk in final_response:
if chunk.content:
final_text += chunk.content
yield chunk.content
self.context.add_message({"role": "assistant", "content": final_text})
async def on_start(session: AgentSession):
agent = AssistantAgent() # fixed: was MyAgent()
session.add_node(agent)
await session.start()
await session.wait_until_complete()
if __name__ == "__main__":
app = AtomsApp(setup_handler=on_start)
app.run()import os
from smallestai.atoms.agent.nodes import OutputAgentNode
from smallestai.atoms.agent.clients import OpenAIClient
from smallestai.atoms.agent.server import AtomsApp
from smallestai.atoms.agent.session import AgentSession
from smallestai.atoms.agent.tools import ToolRegistry, function_tool
from smallestai.atoms.agent.clients.types import ToolCall, ToolResult
class AssistantAgent(OutputAgentNode):
def __init__(self):
super().__init__(name="assistant-agent")
self.llm = OpenAIClient(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
self.tool_registry = ToolRegistry()
self.tool_registry.discover(self)
self.tool_schemas = self.tool_registry.get_schemas()
self.context.add_message({ # added
"role": "system",
"content": "You are a helpful weather assistant. Be concise."
})
@function_tool()
def get_weather(self, city: str) -> str:
"""Get the current weather for a city.
Args:
city: The city name to check weather for.
"""
return f"The weather in {city} is sunny, 72°F"
async def generate_response(self):
response = await self.llm.chat(
messages=self.context.messages,
stream=True,
tools=self.tool_schemas
)
tool_calls = []
full_response = ""
async for chunk in response:
if chunk.content:
full_response += chunk.content
yield chunk.content
if chunk.tool_calls:
tool_calls.extend(chunk.tool_calls)
if not tool_calls: # added
self.context.add_message({"role": "assistant", "content": full_response})
return
yield "One moment while I check that for you. "
results = await self.tool_registry.execute(tool_calls=tool_calls, parallel=True)
self.context.add_messages([ # added
{
"role": "assistant",
"content": "",
"tool_calls": [
{"id": tc.id, "type": "function",
"function": {"name": tc.name, "arguments": str(tc.arguments)}}
for tc in tool_calls
],
},
*[
{"role": "tool", "tool_call_id": tc.id, "content": str(result.content or "")}
for tc, result in zip(tool_calls, results)
],
])
final_response = await self.llm.chat(messages=self.context.messages, stream=True) # added
final_text = ""
async for chunk in final_response:
if chunk.content:
final_text += chunk.content
yield chunk.content
self.context.add_message({"role": "assistant", "content": final_text})
async def on_start(session: AgentSession):
agent = AssistantAgent() # fixed: was MyAgent()
session.add_node(agent)
await session.start()
await session.wait_until_complete()
if __name__ == "__main__":
app = AtomsApp(setup_handler=on_start)
app.run()
आउटपुट:
Weather Agent का उदाहरण कोर टूल एक्ज़ीक्यूशन लूप को प्रदर्शित करता है जहाँ एजेंट एक पूर्ण अनुरोध चक्र को संभालता है: एक प्राकृतिक भाषा क्वेरी प्राप्त करना, सही टूल का चयन करना, उसे निष्पादित करना और परिणाम को एक संवादात्मक प्रतिक्रिया में सिंथेसाइज़ करना — यह सब एक ही बार में। उदाहरण के लिए, एक उपयोगकर्ता द्वारा यह पूछने पर कि "नई दिल्ली में मौसम कैसा है?" एक लुकअप → रिस्पॉन्स चेन ट्रिगर होती है जो वर्तमान परिस्थितियों को लाती है और उन्हें स्वाभाविक रूप से प्रस्तुत करती है, बिना उपयोगकर्ता को यह जाने कि किसी टूल को कॉल किया गया था।
स्पीच आउटपुट (TTS)
स्ट्रीमिंग TTS सिंथेसिस पूर्ण प्रतिक्रिया उत्पन्न होने से पहले ऑडियो प्लेबैक शुरू करके कथित विलंबता को कम करता है। Lightning TTS कई वॉयस विकल्पों में 175ms विलंबता और स्टूडियो-ग्रेड 44kHz आउटपुट गुणवत्ता के साथ यह क्षमता प्रदान करता है।
प्रमुख अनुकूलन तकनीक में डायनेमिक टेक्स्ट स्प्लिटिंग शामिल है—LLM आउटपुट को वाक्य-स्तरीय टुकड़ों में तोड़ना और पूर्ण प्रतिक्रिया की प्रतीक्षा करने के बजाय प्रत्येक को तुरंत TTS में स्ट्रीम करना। यह पृष्ठभूमि में जनरेशन जारी रहने के बावजूद रीयल-टाइम सिंथेसिस का अहसास पैदा करता है।
उत्पादन पैटर्न और सर्वोत्तम अभ्यास
अनुपालन और अवलोकन के लिए मल्टी-नोड आर्किटेक्चर
प्रोडक्शन वॉयस सिस्टम संवादात्मक प्रदर्शन का त्याग किए बिना व्यापक ऑडिट ट्रेल्स की मांग करते हैं। डुअल-नोड पैटर्न संवादात्मक एजेंट के समानांतर एक मूक (silent) BackgroundAgentNode चलाकर इसे प्राप्त करता है, जो अनुपालन डेटाबेस में प्रत्येक इवेंट, टूल कॉल और स्थिति परिवर्तन को लॉग करता है:
from smallestai.atoms.agent.nodes import BackgroundAgentNode
from smallestai.atoms.agent.events import SDKEvent, SDKAgentTranscriptUpdateEvent, SDKSystemUserJoinedEvent
class AuditLogger(BackgroundAgentNode):
def __init__(self, db):
super().__init__(name="audit-logger")
self.db = db
self._call_start = None
self._transcript = []
async def process_event(self, event: SDKEvent):
if isinstance(event, SDKSystemUserJoinedEvent):
self._call_start = datetime.utcnow().isoformat()
self.db.log_audit("CALL_START", json.dumps({"timestamp": self._call_start}))
elif isinstance(event, SDKAgentTranscriptUpdateEvent):
entry = {"role": event.role, "content": event.content}
self._transcript.append(entry)
self.db.log_audit("TRANSCRIPT", json.dumps(entry))
def log_tool_call(self, tool_name: str, args: dict, result: str):
self.db.log_audit("TOOL_CALL", json.dumps({
"tool": tool_name,
"arguments": args,
"result_preview": result[:500] if result else ""
}))
async def setup_session(session: AgentSession):
db = BankingDB()
# Background audit logger -- silent compliance node
audit = AuditLogger(db=db)
session.add_node(audit)
# Main conversational agent
csr = CSRAgent(db=db, audit=audit)
session.add_node(csr)
await session.start()from smallestai.atoms.agent.nodes import BackgroundAgentNode
from smallestai.atoms.agent.events import SDKEvent, SDKAgentTranscriptUpdateEvent, SDKSystemUserJoinedEvent
class AuditLogger(BackgroundAgentNode):
def __init__(self, db):
super().__init__(name="audit-logger")
self.db = db
self._call_start = None
self._transcript = []
async def process_event(self, event: SDKEvent):
if isinstance(event, SDKSystemUserJoinedEvent):
self._call_start = datetime.utcnow().isoformat()
self.db.log_audit("CALL_START", json.dumps({"timestamp": self._call_start}))
elif isinstance(event, SDKAgentTranscriptUpdateEvent):
entry = {"role": event.role, "content": event.content}
self._transcript.append(entry)
self.db.log_audit("TRANSCRIPT", json.dumps(entry))
def log_tool_call(self, tool_name: str, args: dict, result: str):
self.db.log_audit("TOOL_CALL", json.dumps({
"tool": tool_name,
"arguments": args,
"result_preview": result[:500] if result else ""
}))
async def setup_session(session: AgentSession):
db = BankingDB()
# Background audit logger -- silent compliance node
audit = AuditLogger(db=db)
session.add_node(audit)
# Main conversational agent
csr = CSRAgent(db=db, audit=audit)
session.add_node(csr)
await session.start()from smallestai.atoms.agent.nodes import BackgroundAgentNode
from smallestai.atoms.agent.events import SDKEvent, SDKAgentTranscriptUpdateEvent, SDKSystemUserJoinedEvent
class AuditLogger(BackgroundAgentNode):
def __init__(self, db):
super().__init__(name="audit-logger")
self.db = db
self._call_start = None
self._transcript = []
async def process_event(self, event: SDKEvent):
if isinstance(event, SDKSystemUserJoinedEvent):
self._call_start = datetime.utcnow().isoformat()
self.db.log_audit("CALL_START", json.dumps({"timestamp": self._call_start}))
elif isinstance(event, SDKAgentTranscriptUpdateEvent):
entry = {"role": event.role, "content": event.content}
self._transcript.append(entry)
self.db.log_audit("TRANSCRIPT", json.dumps(entry))
def log_tool_call(self, tool_name: str, args: dict, result: str):
self.db.log_audit("TOOL_CALL", json.dumps({
"tool": tool_name,
"arguments": args,
"result_preview": result[:500] if result else ""
}))
async def setup_session(session: AgentSession):
db = BankingDB()
# Background audit logger -- silent compliance node
audit = AuditLogger(db=db)
session.add_node(audit)
# Main conversational agent
csr = CSRAgent(db=db, audit=audit)
session.add_node(csr)
await session.start()दोनों नोड्स को समान इवेंट स्ट्रीम प्राप्त होती है लेकिन वे अलग-अलग उद्देश्यों की पूर्ति करते हैं: CSRAgent बातचीत को संभालता है जबकि AuditLogger अनुपालन, विश्लेषण और प्रशिक्षण डेटा जनरेशन के लिए चुपचाप सब कुछ रिकॉर्ड करता है। चूँकि बैकग्राउंड नोड एसिंक्रोनस रूप से संचालित होता है, यह उपयोगकर्ता-सामना करने वाले इंटरैक्शन में शून्य विलंबता पेश करता है।
पहचान सत्यापन और सुरक्षा उपाय (Guardrails)
बैंकिंग वॉयस एजेंटों को संवेदनशील जानकारी प्रकट करने से पहले उपयोगकर्ताओं को प्रमाणित करना होगा। नॉलेज-बेस्ड ऑथेंटिकेशन (KBA) पैटर्न सत्र-आधारित सत्यापन स्थिति के साथ श्रेणीबद्ध एक्सेस कंट्रोल को लागू करता है:
@function_tool()
def verify_customer(
self,
name: str = "",
dob: str = "",
account_last_four: str = "",
city: str = "",
debit_card_last_four: str = ""
) -> str:
"""Verify customer identity using Knowledge-Based Authentication.
Level 1 (info queries): 2 matching factors
Level 2 (banking actions): 3 matching factors
Args:
name: Customer's full name
dob: Date of birth (YYYY-MM-DD)
account_last_four: Last 4 digits of savings account
city: City from address
debit_card_last_four: Last 4 digits of debit card
"""
if self.is_verified:
return f"Customer already verified at Level {self.verification_level}"
# Fetch ground truth from database
row = self.db.execute_read_query(
"SELECT c.name, c.dob, c.city, a.account_number, ca.last_four AS debit_last_four "
"FROM customers c JOIN accounts a ON a.customer_id = c.id "
"JOIN cards ca ON ca.customer_id = c.id WHERE ca.type = 'debit'"
)
truth = row[0]
factors_matched = []
if name and name.strip().lower() == truth["name"].strip().lower():
factors_matched.append("name")
if dob and dob.strip() == truth["dob"]:
factors_matched.append("dob")
if account_last_four and account_last_four.strip() == truth["account_number"][-4:]:
factors_matched.append("account_last_four")
# ... additional factor checking
n = len(factors_matched)
if n >= 3:
self.is_verified = True
self.verification_level = 2
return "Level 2 verification successful -- high-risk actions allowed"
elif n >= 2:
self.is_verified = True
self.verification_level = 1
return "Level 1 verification successful -- info queries allowed"
else:
return "Verification failed -- insufficient matching factors"@function_tool()
def verify_customer(
self,
name: str = "",
dob: str = "",
account_last_four: str = "",
city: str = "",
debit_card_last_four: str = ""
) -> str:
"""Verify customer identity using Knowledge-Based Authentication.
Level 1 (info queries): 2 matching factors
Level 2 (banking actions): 3 matching factors
Args:
name: Customer's full name
dob: Date of birth (YYYY-MM-DD)
account_last_four: Last 4 digits of savings account
city: City from address
debit_card_last_four: Last 4 digits of debit card
"""
if self.is_verified:
return f"Customer already verified at Level {self.verification_level}"
# Fetch ground truth from database
row = self.db.execute_read_query(
"SELECT c.name, c.dob, c.city, a.account_number, ca.last_four AS debit_last_four "
"FROM customers c JOIN accounts a ON a.customer_id = c.id "
"JOIN cards ca ON ca.customer_id = c.id WHERE ca.type = 'debit'"
)
truth = row[0]
factors_matched = []
if name and name.strip().lower() == truth["name"].strip().lower():
factors_matched.append("name")
if dob and dob.strip() == truth["dob"]:
factors_matched.append("dob")
if account_last_four and account_last_four.strip() == truth["account_number"][-4:]:
factors_matched.append("account_last_four")
# ... additional factor checking
n = len(factors_matched)
if n >= 3:
self.is_verified = True
self.verification_level = 2
return "Level 2 verification successful -- high-risk actions allowed"
elif n >= 2:
self.is_verified = True
self.verification_level = 1
return "Level 1 verification successful -- info queries allowed"
else:
return "Verification failed -- insufficient matching factors"@function_tool()
def verify_customer(
self,
name: str = "",
dob: str = "",
account_last_four: str = "",
city: str = "",
debit_card_last_four: str = ""
) -> str:
"""Verify customer identity using Knowledge-Based Authentication.
Level 1 (info queries): 2 matching factors
Level 2 (banking actions): 3 matching factors
Args:
name: Customer's full name
dob: Date of birth (YYYY-MM-DD)
account_last_four: Last 4 digits of savings account
city: City from address
debit_card_last_four: Last 4 digits of debit card
"""
if self.is_verified:
return f"Customer already verified at Level {self.verification_level}"
# Fetch ground truth from database
row = self.db.execute_read_query(
"SELECT c.name, c.dob, c.city, a.account_number, ca.last_four AS debit_last_four "
"FROM customers c JOIN accounts a ON a.customer_id = c.id "
"JOIN cards ca ON ca.customer_id = c.id WHERE ca.type = 'debit'"
)
truth = row[0]
factors_matched = []
if name and name.strip().lower() == truth["name"].strip().lower():
factors_matched.append("name")
if dob and dob.strip() == truth["dob"]:
factors_matched.append("dob")
if account_last_four and account_last_four.strip() == truth["account_number"][-4:]:
factors_matched.append("account_last_four")
# ... additional factor checking
n = len(factors_matched)
if n >= 3:
self.is_verified = True
self.verification_level = 2
return "Level 2 verification successful -- high-risk actions allowed"
elif n >= 2:
self.is_verified = True
self.verification_level = 1
return "Level 1 verification successful -- info queries allowed"
else:
return "Verification failed -- insufficient matching factors"
यह दृष्टिकोण प्रति सत्र एक बार सत्यापित करता है, जिससे बातचीत के दौरान स्थिति बनी रहती है ताकि उपयोगकर्ताओं को बार-बार प्रमाणीकरण चुनौतियों का सामना न करना पड़े। लेवल 1 एक्सेस बैलेंस पूछताछ और खर्च विश्लेषण को सक्षम बनाता है; लेवल 2 फिक्स्ड डिपॉजिट तोड़ने जैसे लेनदेन की अनुमति देता है।
कॉल कंट्रोल और एस्केलेशन (हस्तांतरण)
वॉइस बॉट्स को कॉल एस्केलेशन को शालीनता से संभालना चाहिए। Atoms SDK कॉल समाप्त करने और पूर्ण संदर्भ संरक्षण के साथ मानव एजेंटों को स्थानांतरित करने के लिए संरचित इवेंट्स प्रदान करता है:
from smallestai.atoms.agent.events import (
SDKAgentEndCallEvent,
SDKAgentTransferConversationEvent,
TransferOption,
TransferOptionType,
WarmTransferPrivateHandoffOption,
WarmTransferHandoffOptionType
)
@function_tool()
async def transfer_to_human_agent(self) -> None:
"""Cold transfer: immediate handoff to human agent."""
await self.send_event(
SDKAgentTransferConversationEvent(
transfer_call_number=os.getenv("TRANSFER_NUMBER"),
transfer_options=TransferOption(
type=TransferOptionType.COLD_TRANSFER
),
on_hold_music="relaxing_sound"
)
)
return None
@function_tool()
async def warm_transfer_to_supervisor(self, reason: str) -> None:
"""Warm transfer: brief supervisor first, then connect customer.
Args:
reason: Summary of customer issue for supervisor briefing
"""
await self.send_event(
SDKAgentTransferConversationEvent(
transfer_call_number=os.getenv("TRANSFER_NUMBER"),
transfer_options=TransferOption(
type=TransferOptionType.WARM_TRANSFER,
private_handoff_option=WarmTransferPrivateHandoffOption(
type=WarmTransferHandoffOptionType.PROMPT,
prompt=f"Customer escalation: {reason}"
)
),
on_hold_music="uplifting_beats"
)
)
return Nonefrom smallestai.atoms.agent.events import (
SDKAgentEndCallEvent,
SDKAgentTransferConversationEvent,
TransferOption,
TransferOptionType,
WarmTransferPrivateHandoffOption,
WarmTransferHandoffOptionType
)
@function_tool()
async def transfer_to_human_agent(self) -> None:
"""Cold transfer: immediate handoff to human agent."""
await self.send_event(
SDKAgentTransferConversationEvent(
transfer_call_number=os.getenv("TRANSFER_NUMBER"),
transfer_options=TransferOption(
type=TransferOptionType.COLD_TRANSFER
),
on_hold_music="relaxing_sound"
)
)
return None
@function_tool()
async def warm_transfer_to_supervisor(self, reason: str) -> None:
"""Warm transfer: brief supervisor first, then connect customer.
Args:
reason: Summary of customer issue for supervisor briefing
"""
await self.send_event(
SDKAgentTransferConversationEvent(
transfer_call_number=os.getenv("TRANSFER_NUMBER"),
transfer_options=TransferOption(
type=TransferOptionType.WARM_TRANSFER,
private_handoff_option=WarmTransferPrivateHandoffOption(
type=WarmTransferHandoffOptionType.PROMPT,
prompt=f"Customer escalation: {reason}"
)
),
on_hold_music="uplifting_beats"
)
)
return Nonefrom smallestai.atoms.agent.events import (
SDKAgentEndCallEvent,
SDKAgentTransferConversationEvent,
TransferOption,
TransferOptionType,
WarmTransferPrivateHandoffOption,
WarmTransferHandoffOptionType
)
@function_tool()
async def transfer_to_human_agent(self) -> None:
"""Cold transfer: immediate handoff to human agent."""
await self.send_event(
SDKAgentTransferConversationEvent(
transfer_call_number=os.getenv("TRANSFER_NUMBER"),
transfer_options=TransferOption(
type=TransferOptionType.COLD_TRANSFER
),
on_hold_music="relaxing_sound"
)
)
return None
@function_tool()
async def warm_transfer_to_supervisor(self, reason: str) -> None:
"""Warm transfer: brief supervisor first, then connect customer.
Args:
reason: Summary of customer issue for supervisor briefing
"""
await self.send_event(
SDKAgentTransferConversationEvent(
transfer_call_number=os.getenv("TRANSFER_NUMBER"),
transfer_options=TransferOption(
type=TransferOptionType.WARM_TRANSFER,
private_handoff_option=WarmTransferPrivateHandoffOption(
type=WarmTransferHandoffOptionType.PROMPT,
prompt=f"Customer escalation: {reason}"
)
),
on_hold_music="uplifting_beats"
)
)
return None
कोल्ड ट्रांसफर तुरंत उपयोगकर्ताओं को एजेंटों से जोड़ते हैं—जो सीधे हस्तांतरण के लिए आदर्श हैं। वार्म ट्रांसफर ग्राहक को जोड़ने से पहले प्राप्तकर्ता एजेंट को संदर्भ के साथ संक्षिप्त जानकारी देते हैं, जिससे जटिल मुद्दों के लिए निर्बाध निरंतरता सक्षम होती है।
प्रदर्शन और विलंबता अनुकूलन
उप-सेकंड संवादात्मक विलंबता प्राप्त करने के लिए संपूर्ण पाइपलाइन में स्ट्रीमिंग की आवश्यकता होती है। प्रत्येक घटक को बैच मोड के बजाय वृद्धिशील रूप से डेटा संसाधित करना चाहिए: ASR आंशिक ट्रांसक्रिप्ट को स्ट्रीम करता है, LLM टोकन-दर-टोकन प्रतिक्रियाओं को स्ट्रीम करता है, और TTS तुरंत वाक्य-स्तरीय टुकड़ों को सिंथेसाइज करता है।
मध्यवर्ती प्रतिक्रिया पैटर्न टूल निष्पादन के दौरान जुड़ाव बनाए रखता है। बाहरी एपीआई या डेटाबेस को कॉल करते समय, एजेंट टूल को इनवोक करने से पहले पावती वाक्यांश जैसे "कृपया एक क्षण प्रतीक्षा करें, मैं आपके लिए इसकी जांच कर रहा हूँ" देता है। यह अजीब चुप्पी को रोकता है और बैकएंड संचालन पूरा होने के दौरान भी सिस्टम की जवाबदेही का संकेत देता है।
smallest.ai का इंफ्रास्ट्रक्चर इन अनुकूलनों को सक्षम बनाता है जिसमें Pulse STT 64ms फर्स्ट-ट्रांसक्रिप्ट लेटेंसी प्रदान करता है और Lightning TTS 175ms सिंथेसिस समय प्राप्त करता है। कुशल ऑर्केस्ट्रेशन के साथ मिलकर, 800ms से कम का कुल टर्न टाइम प्राप्त करना संभव हो जाता है—वह सीमा जहाँ बातचीत वास्तव में प्राकृतिक महसूस होती है।
वास्तविक दुनिया के उपयोग के मामले (Use Cases)
बैंकिंग वॉयस एजेंट (Bank CSR)
Bank CSR का उदाहरण जटिल वित्तीय वर्कफ़्लोज़ को संभालने वाले एंटरप्राइज़-ग्रेड वॉयस AI को प्रदर्शित करता है। जब कोई ग्राहक पूछता है "मैंने जनवरी 2024 से अमेज़न पर कितना खर्च किया?", तो यह एक बहु-चरणीय प्रक्रिया को ऑर्केस्ट्रेट करता है:
सबसे पहले, एजेंट KBA का उपयोग करके ग्राहक की पहचान सत्यापित करता है, जिसके लिए खाता जानकारी तक पहुँचने के लिए दो मिलान कारकों की आवश्यकता होती है। एक बार सत्यापित होने के बाद, यह लेनदेन डेटाबेस के विरुद्ध एक SQL क्वेरी निष्पादित करता है:
@function_tool()
def execute_query(self, sql: str) -> str:
"""Execute a SELECT query against the banking database.
Args:
sql: Valid SELECT statement
"""
# Validate query is read-only
if not re.match(r"(?i)^\s*SELECT\b", sql.strip()):
raise ValueError("Only SELECT queries allowed")
rows = self.db.execute_read_query(sql)
return json.dumps(rows)@function_tool()
def execute_query(self, sql: str) -> str:
"""Execute a SELECT query against the banking database.
Args:
sql: Valid SELECT statement
"""
# Validate query is read-only
if not re.match(r"(?i)^\s*SELECT\b", sql.strip()):
raise ValueError("Only SELECT queries allowed")
rows = self.db.execute_read_query(sql)
return json.dumps(rows)@function_tool()
def execute_query(self, sql: str) -> str:
"""Execute a SELECT query against the banking database.
Args:
sql: Valid SELECT statement
"""
# Validate query is read-only
if not re.match(r"(?i)^\s*SELECT\b", sql.strip()):
raise ValueError("Only SELECT queries allowed")
rows = self.db.execute_read_query(sql)
return json.dumps(rows)
कच्चे क्वेरी परिणाम फिर एक नियतात्मक (deterministic) विश्लेषण फ़ंक्शन में जाते हैं जो योग की गणना करता है, प्रवृत्तियों की पहचान करता है, और आउटपुट को स्वरूपित करता है—यह सुनिश्चित करते हुए कि गणितीय संचालन संभावित रूप से मतिभ्रम करने वाले LLM अंकगणित पर भरोसा करने के बजाय शुद्ध पायथन में हो:
@function_tool()
def analyze_data(self, data_json: str, analysis_type: str) -> dict:
"""Perform deterministic analysis on query results.
Args:
data_json: JSON string of query results
analysis_type: One of 'total', 'trend_yearly', 'comparison'
"""
rows = json.loads(data_json)
if analysis_type == "total":
total = sum(self._get_amount(r) for r in rows)
return {"total": total, "count": len(rows), "currency": "INR"}
elif analysis_type == "trend_yearly":
# Group by year, compute YoY changes
yearly = defaultdict(int)
for row in rows:
year = row["date"][:4]
yearly[year] += self._get_amount(row)
return {"yearly_totals": dict(yearly)}@function_tool()
def analyze_data(self, data_json: str, analysis_type: str) -> dict:
"""Perform deterministic analysis on query results.
Args:
data_json: JSON string of query results
analysis_type: One of 'total', 'trend_yearly', 'comparison'
"""
rows = json.loads(data_json)
if analysis_type == "total":
total = sum(self._get_amount(r) for r in rows)
return {"total": total, "count": len(rows), "currency": "INR"}
elif analysis_type == "trend_yearly":
# Group by year, compute YoY changes
yearly = defaultdict(int)
for row in rows:
year = row["date"][:4]
yearly[year] += self._get_amount(row)
return {"yearly_totals": dict(yearly)}@function_tool()
def analyze_data(self, data_json: str, analysis_type: str) -> dict:
"""Perform deterministic analysis on query results.
Args:
data_json: JSON string of query results
analysis_type: One of 'total', 'trend_yearly', 'comparison'
"""
rows = json.loads(data_json)
if analysis_type == "total":
total = sum(self._get_amount(r) for r in rows)
return {"total": total, "count": len(rows), "currency": "INR"}
elif analysis_type == "trend_yearly":
# Group by year, compute YoY changes
yearly = defaultdict(int)
for row in rows:
year = row["date"][:4]
yearly[year] += self._get_amount(row)
return {"yearly_totals": dict(yearly)}
इस पूरे वर्कफ़्लो के दौरान, मूक AuditLogger अनुपालन ऑडिट ट्रेल्स के लिए प्रत्येक क्वेरी, टूल कॉल और सत्यापन प्रयास को रिकॉर्ड करता है। एजेंट एक प्राकृतिक भाषा प्रतिक्रिया को सिंथेसाइज़ करके निष्कर्ष निकालता है: "जनवरी 2024 से आपका कुल अमेज़न खर्च 13 लेनदेन में तीन लाख छिहत्तर हजार रुपये था।"
ग्राहक सहायता और IVR रिप्लेसमेंट
आधुनिक वॉयस बॉट्स निराशाजनक मेनू-संचालित IVR सिस्टम को इरादे-संचालित बातचीत से बदल देते हैं। background_agent उदाहरण समानांतर रूप से चलने वाले रीयल-टाइम सेंटीमेंट विश्लेषण को प्रदर्शित करता है—निराशा के स्तरों की निगरानी करना और पैटर्न असंतोष का संकेत देने पर स्वचालित रूप से मानव एजेंटों तक एस्केलेट करना:
class SentimentAnalyzer(BackgroundAgentNode):
def __init__(self):
super().__init__(name="sentiment-analyzer")
self.frustration_count = 0
async def process_event(self, event: SDKEvent):
if isinstance(event, SDKAgentTranscriptUpdateEvent):
if event.role == "user":
sentiment = await self._analyze_sentiment(event.content)
if sentiment in ["negative", "frustrated"]:
self.frustration_count += 1
if self.frustration_count >= 3:
# Automatically trigger escalation
await self.notify_main_agent_to_escalate()class SentimentAnalyzer(BackgroundAgentNode):
def __init__(self):
super().__init__(name="sentiment-analyzer")
self.frustration_count = 0
async def process_event(self, event: SDKEvent):
if isinstance(event, SDKAgentTranscriptUpdateEvent):
if event.role == "user":
sentiment = await self._analyze_sentiment(event.content)
if sentiment in ["negative", "frustrated"]:
self.frustration_count += 1
if self.frustration_count >= 3:
# Automatically trigger escalation
await self.notify_main_agent_to_escalate()class SentimentAnalyzer(BackgroundAgentNode):
def __init__(self):
super().__init__(name="sentiment-analyzer")
self.frustration_count = 0
async def process_event(self, event: SDKEvent):
if isinstance(event, SDKAgentTranscriptUpdateEvent):
if event.role == "user":
sentiment = await self._analyze_sentiment(event.content)
if sentiment in ["negative", "frustrated"]:
self.frustration_count += 1
if self.frustration_count >= 3:
# Automatically trigger escalation
await self.notify_main_agent_to_escalate()इरादा पहचान (intent recognition), अनुकूली एस्केलेशन और व्यापक विश्लेषण का यह संयोजन वॉयस बॉट्स को ग्राहक इंटरैक्शन को उस परिष्कार के साथ संभालने में सक्षम बनाता है जिसका मुकाबला विरासत IVR सिस्टम बस नहीं कर सकते।
मुख्य निष्कर्ष
वॉयस बॉट आर्किटेक्चर कई विषयों को सिंथेसाइज़ करता है: कम-विलंबता ASR और TTS के माध्यम से रीयल-टाइम स्पीच प्रोसेसिंग, बातचीत की स्थिति और टूल निष्पादन को प्रबंधित करने वाला परिष्कृत एजेंट ऑर्केस्ट्रेशन, ऑडिट लॉगिंग और पहचान सत्यापन के साथ मजबूत अनुपालन इंफ्रास्ट्रक्चर, और उप-सेकंड राउंड-ट्रिप समय प्राप्त करने वाला प्रदर्शन अनुकूलन। सफलता के लिए संपूर्ण पाइपलाइन में डेटा स्ट्रीमिंग की आवश्यकता होती है, चिंताओं को अलग करने के लिए मल्टी-नोड पैटर्न का लाभ उठाना और टूल निष्पादन के दौरान मध्यवर्ती प्रतिक्रिया प्रदान करना शामिल है। smallest.ai मूलभूत स्पीच और एजेंट इंफ्रास्ट्रक्चर—Pulse STT, Lightning TTS, और Atoms SDK—प्रदान करता है, जिससे डेवलपर्स को निम्न-स्तरीय घटकों को दोबारा बनाए बिना प्रोडक्शन वॉयस सिस्टम बनाने की सुविधा मिलती है।
निष्कर्ष
एक अच्छी तरह से आर्किटेक्ट किया गया वॉयस बॉट एक सक्षम कर्मचारी की तरह महसूस होता है: तेज, सटीक, संदर्भ के प्रति जागरूक, और उपयोगकर्ताओं की ओर से कार्रवाई करने में सक्षम। यहाँ खोजे गए पैटर्न—बुनियादी AtomsApp सेटअप से लेकर परिष्कृत मल्टी-नोड अनुपालन आर्किटेक्चर तक—उन प्रोडक्शन सिस्टम्स के निर्माण के लिए एक रोडमैप प्रदान करते हैं जो उपभोक्ता-ग्रेड अनुभव प्रदान करते हुए उद्यम की आवश्यकताओं को पूरा करते हैं। Atoms SDK के त्वरित प्रारंभ पैटर्न्स के साथ शुरू करें, smallest.ai की रीयल-टाइम ASR और TTS क्षमताओं को एकीकृत करें, और व्यावसायिक-विशिष्ट टूल्स और वर्कफ़्लोज़ को उत्तरोत्तर शामिल करें। आज ऐसा इंफ्रास्ट्रक्चर मौजूद है जो ऐसे वॉयस अनुभव का निर्माण कर सकता है जो वास्तव में इस बात को बदल देता है कि संगठन अपने ग्राहकों के साथ कैसे बातचीत करते हैं। प्रोडक्शन वॉयस AI में अपनी यात्रा शुरू करने के लिए smallest.ai पर Pulse STT, Lightning TTS और Atoms एजेंट फ्रेमवर्क का पता लगाएं।
