phoson-engine-minimal

from phoson_llm.schemas import Message, ModelConfig

chat = OpenAIChat()
messages = [Message(role="user", content="Hello!")]
config = ModelConfig(model="openai/gpt-4o-mini", max_tokens=256)

async for event in chat.stream(messages, config):
    match event:
        case TokenEvent(content=c):
            print(c, end="", flush=True)
        case UsageEvent(cost_usd=cost):
            print(f"\nCost: ${cost:.6f}")
        case LLMDoneEvent():
            print("\n[done]")

from phoson_llm.schemas import (
    Message,          # role: "system" | "user" | "assistant", content: str | list[ContentBlock]
    ModelConfig,      # model, temperature, max_tokens, system, thinking_budget, reasoning_effort
    ToolDefinition,   # name, description, parameters (JSON Schema)
    # Content blocks for multimodal messages:
    TextBlock,        # text: str
    ImageBlock,       # source: url | base64 | file://, detail, media_type
    AudioBlock,       # source, format, duration_ms
    VideoBlock,       # source, sampling_interval_ms
    DocumentBlock,    # source (PDF), pages
    ToolUseBlock,     # tool_call_id, tool_name, args
    ToolResultBlock,  # tool_call_id, result, error
)

# Example: multimodal message with image
msg = Message(
    role="user",
    content=[
        TextBlock(text="What's in this image?"),
        ImageBlock(source="https://example.com/photo.jpg", detail="high"),
    ]
)

from phoson_llm.pricing import calculate_cost

cost_usd, cost_known = calculate_cost(
    model="openai/gpt-4o",
    input_tokens=1000,
    output_tokens=500,
    cache_read_tokens=200,
)
# cost_usd ≈ 0.0035, cost_known=True

# Silence warnings for unknown models (e.g. Ollama)
import warnings
from phoson_llm.pricing import UnknownModelWarning
warnings.filterwarnings("ignore", category=UnknownModelWarning)

from typing import Annotated, Literal
from phoson_agent import tool

# Basic tool — schema inferred from type hints
@tool
def search_web(query: str, max_results: int = 5) -> list[dict]:
    """Search the web and return results."""
    return [{"url": "...", "snippet": "..."}]

# Annotated descriptions per parameter
@tool
def calculate(
    expression: Annotated[str, "A mathematical expression to evaluate"],
    precision: Annotated[int, "Decimal places in result"] = 2,
) -> str:
    """Evaluate a mathematical expression safely."""
    return str(round(eval(expression), precision))

# Literal enum constraint
@tool
def set_language(lang: Literal["en", "es", "fr", "de"]) -> str:
    """Set the response language."""
    return f"Language set to {lang}"

# Async tool
@tool
async def fetch_url(url: str) -> str:
    """Fetch the content of a URL."""
    import httpx
    async with httpx.AsyncClient() as client:
        r = await client.get(url)
        return r.text

# Context injection — inject values from AgentContext
@tool(inject=["user_id"])
def get_user_profile(*, user_id: str) -> dict:
    """Get the profile of the current user."""
    return {"id": user_id, "name": "Alice"}

from phoson_agent.middleware import AgentMiddleware
from phoson_llm.schemas import Message, ModelConfig, ToolCallEvent

class LoggingMiddleware(AgentMiddleware):
    """Logs every LLM call and tool execution."""

    async def on_before_llm(self, messages: list[Message], config: ModelConfig) -> list[Message]:
        print(f"[LLM] → {config.model} ({len(messages)} messages)")
        return messages  # must return messages (can modify them)

    async def on_before_tool(self, call: ToolCallEvent) -> ToolCallEvent | None:
        print(f"[Tool] → {call.tool_name}({call.args})")
        return call  # return None to cancel the tool call

    async def on_after_tool(self, call: ToolCallEvent, result: str, error: bool) -> str:
        print(f"[Tool] ← {call.tool_name}: {'ERROR' if error else 'OK'}")
        return result  # can transform the result

engine = AgentEngine(chat=chat, tools=tools, middlewares=[LoggingMiddleware()])

from phoson_agent.middleware import RetryMiddleware

retry = RetryMiddleware(
    max_retries=2,
    base_delay_seconds=0.5,
    backoff_multiplier=2.0,
)
engine = AgentEngine(chat=chat, tools=tools, middlewares=[retry])

from phoson_agent.plugins.summarizer import SummarizationMiddleware

summarizer = SummarizationMiddleware(
    threshold=0.80,            # compact when > 80% of context window used
    min_keep_messages=4,       # always preserve last N messages
    provider="openrouter",
    model="anthropic/claude-sonnet-4-6",
)
engine = AgentEngine(chat=chat, tools=tools, middlewares=[summarizer])

from phoson_agent.plugin import Plugin
from phoson_agent.middleware import AgentMiddleware
from phoson_agent.models import AgentTool

class MyPlugin(Plugin):
    @property
    def name(self) -> str:
        return "my-plugin"

    def get_tools(self) -> list[AgentTool]:
        return [my_tool]

    def get_middlewares(self) -> list[AgentMiddleware]:
        return [LoggingMiddleware()]

    def initialize(self) -> None:
        print("Plugin initialized")

    def cleanup(self) -> None:
        print("Plugin cleaned up")

# Load by instance, package string, or dict with config
engine = AgentEngine(
    chat=chat,
    plugins=[
        MyPlugin(),                                    # instance
        "phoson_plugin_mcp",                           # package string
        {"name": "my-plugin", "config": {"key": "v"}}, # dict with config
    ],
)

from phoson_agent.sessions import ConversationTree, JsonlStorage
from phoson_llm.schemas import Message

# Create a new session tree
tree = ConversationTree.new(session_id="my-session")

# Append messages
node_a = tree.append(parent_id=None, message=Message(role="user", content="Hello"))
node_b = tree.append(parent_id=node_a.id, message=Message(role="assistant", content="Hi!"))

# Branch from node_a to explore a different path
node_c = tree.append(parent_id=node_a.id, message=Message(role="assistant", content="Hey there!"))

# Get the full message path to any node
path = tree.get_path(node_b.id)  # [user: "Hello", assistant: "Hi!"]

# Label a node
tree.label(node_b.id, "main branch")

# Persist with JSONL storage
storage = JsonlStorage(base_dir="~/.phoson/sessions")
await storage.save(tree)
loaded = await storage.load("my-session")

# List all sessions
sessions = await storage.list_sessions()  # list[SessionMeta]

# Delete a session
await storage.delete("my-session")