chatomon¶

from pydantic_ai.models.test import TestModel

from chatomon import Atom, Knowledge, OrbitalContext, Verdict, orbital
from chatomon.connectors.pydantic_ai import PydanticAIConnector


@orbital
def no_refunds(ctx: OrbitalContext) -> Verdict:
    if "refund" in ctx.message.lower():
        return Verdict.block(guidance="Explain that refunds are handled by email.")
    return Verdict.allow()


atom = Atom(
    name="speedypack",
    instructions="You are the customer assistant of SpeedyPack, a parcel courier.",
    knowledge=Knowledge("Opening hours: Monday to Friday, 9-18."),
    connector=PydanticAIConnector(TestModel(custom_output_text="Refunds are handled by email.")),
    input_orbitals=[no_refunds],
)
agent = atom.build()
result = agent.run_sync("I want a refund")
assert result.output == "Refunds are handled by email."

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, HumanMessage

from chatomon import Atom, Knowledge, OrbitalContext, Verdict, orbital
from chatomon.connectors.langchain import LangChainConnector


@orbital
def no_refunds(ctx: OrbitalContext) -> Verdict:
    if "refund" in ctx.message.lower():
        return Verdict.block(guidance="Explain that refunds are handled by email.")
    return Verdict.allow()


model = GenericFakeChatModel(messages=iter([AIMessage(content="Refunds are handled by email.")]))
atom = Atom(
    name="speedypack",
    instructions="You are the customer assistant of SpeedyPack, a parcel courier.",
    knowledge=Knowledge("Opening hours: Monday to Friday, 9-18."),
    connector=LangChainConnector(model),
    input_orbitals=[no_refunds],
)
agent = atom.build()
result = agent.invoke({"messages": [HumanMessage(content="I want a refund")]})
assert result["messages"][-1].content == "Refunds are handled by email."

import asyncio
import os

from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import DefaultAzureCredential

from chatomon import Atom, OrbitalContext, Verdict, orbital
from chatomon.connectors.maf import MAFConnector


@orbital
async def no_refunds(ctx: OrbitalContext) -> Verdict:
    if "refund" in ctx.message.lower():
        return Verdict.block(guidance="Explain that refunds are handled by email.")
    return Verdict.allow()


async def main() -> None:
    async with DefaultAzureCredential() as credential:
        client = FoundryChatClient(
            project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
            model=os.environ["FOUNDRY_MODEL_DEPLOYMENT"],
            credential=credential,
        )
        try:
            agent = Atom(
                name="speedypack",
                instructions="You are the customer assistant of SpeedyPack.",
                connector=MAFConnector(client),
                input_orbitals=[no_refunds],
            ).build()
            reply = await agent.run("I want a refund", session=agent.create_session())
            print(reply.text)
        finally:
            await client.project_client.close()


asyncio.run(main())

import asyncio

from google.adk.runners import InMemoryRunner

from chatomon import Atom, OrbitalContext, Verdict, orbital
from chatomon.connectors.adk import ADKConnector


@orbital
async def no_refunds(ctx: OrbitalContext) -> Verdict:
    if "refund" in ctx.message.lower():
        return Verdict.block(guidance="Explain that refunds are handled by email.")
    return Verdict.allow()


async def main() -> None:
    app = Atom(
        name="speedypack",
        instructions="You are the customer assistant of SpeedyPack.",
        connector=ADKConnector("gemini-3.5-flash"),
        input_orbitals=[no_refunds],
    ).build()
    events = await InMemoryRunner(app=app).run_debug("I want a refund", quiet=True)
    print(events[-1].content.parts[0].text)


asyncio.run(main())