Quickstart

from infy import HumanMessage
from infy.providers.openai import OpenAIChat

model = OpenAIChat("gpt-4o")

reply = model.generate([
    HumanMessage(content="Summarize bulk-synchronous parallelism in one sentence."),
])
print(reply.text)

async for chunk in model.astream([
    HumanMessage(content="Stream me a haiku about schedulers."),
]):
    print(chunk.content, end="", flush=True)

from infy import create_agent, tool

@tool
def get_weather(city: str) -> str:
    """Return the current weather for a city."""
    return f"{city}: 17C, clear"

agent = create_agent(model, tools=[get_weather])
result = agent("What's the weather in Boston?")

print(result.response.text)
print(result.iterations, result.tool_calls_made)

from infy import create_agent, tool
from infy.governance import Governance, Policy, CallbackApprover, ApprovalRequest

@tool
def search(query: str) -> str:
    """Search internal docs."""
    return "..."

@tool
def deploy(service: str) -> str:
    """Deploy a service to production."""
    return f"deployed {service}"

@tool
def delete_database(name: str) -> str:
    """Permanently delete a database."""
    return f"deleted {name}"

def ask_a_human(request: ApprovalRequest) -> bool:
    # Wire this to Slack, a console prompt, or a queue. Returns True to allow.
    answer = input(f"Approve {request.tool}({request.args})? [y/N] ")
    return answer.strip().lower() == "y"

gov = Governance(
    policy=Policy(
        deny=["delete_database"],       # never runs, regardless of approval
        require_approval=["deploy"],     # runs only on a human yes
    ),
    approver=CallbackApprover(ask_a_human),
    principal="agent://acme/assistant",
)

agent = create_agent(
    model,
    tools=[search, deploy, delete_database],
    governance=gov,
)
agent("ship the release")

assert gov.audit.verify()   # tamper-evident receipt of everything that happened