Google ADK

pip install -U google-adk litellm

import os
import asyncio
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner

# Initialize an agent with a Neosantara model
researcher = Agent(
    name="researcher_agent",
    model=LiteLlm(model="neosantara/gemini-3.8-flash"),
    instruction="You are a technology research assistant that presents data concisely and objectively."
)

runner = InMemoryRunner(agent=researcher)

async def main():
    events = await runner.run_debug(
        "Explain the benefits of a regional AI gateway for developers in Indonesia.",
        quiet=True
    )
    for event in events:
        if event.is_final_response() and event.content:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    print(part.text)

if __name__ == "__main__":
    asyncio.run(main())

import os
import asyncio
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner

def calculate_token_estimate(text: str) -> str:
    """Calculate token and compute estimation based on character length."""
    estimated_tokens = len(text) // 4
    return f"Text length: {len(text)} characters. Estimated tokens: {estimated_tokens}."

analyst = Agent(
    name="token_analyst",
    model=LiteLlm(
        model="neosantara/gemini-3.8-flash",
        allowed_openai_params=["tools"]
    ),
    instruction="Use calculate_token_estimate when the user asks about text payload sizing.",
    tools=[calculate_token_estimate]
)

runner = InMemoryRunner(agent=analyst)

async def main():
    events = await runner.run_debug(
        "What is the token estimate for a 1000-character payload?",
        quiet=True
    )
    for event in events:
        if event.is_final_response() and event.content:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    print(part.text)

if __name__ == "__main__":
    asyncio.run(main())

import os
import asyncio
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner

# Specialized analytical reasoning sub-agent
analyst_agent = Agent(
    name="analyst",
    model=LiteLlm(model="neosantara/deepseek-v4.1-flash"),
    instruction="Perform rigorous quantitative reasoning and throughput latency analysis."
)

# Root orchestrator delegating domain queries
lead_orchestrator = Agent(
    name="coordinator",
    model=LiteLlm(model="neosantara/gemini-3.8-flash"),
    instruction="Coordinate technical questions. Delegate in-depth calculations to the analyst sub-agent.",
    sub_agents=[analyst_agent]
)

runner = InMemoryRunner(agent=lead_orchestrator)

async def main():
    events = await runner.run_debug(
        "Calculate concurrency efficiency for 1000 RPM on the Basic tier.",
        quiet=True
    )
    for event in events:
        if event.is_final_response() and event.content:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    print(part.text)

if __name__ == "__main__":
    asyncio.run(main())