Osmosis Python SDK 安装

python -m pip install "osmosis-ai>=0.3.10,<0.4"

python -c "import osmosis_ai; print(osmosis_ai.__version__)"
osmosis --version

python -m pip install "osmosis-ai[server,strands]>=0.3.10,<0.4"

import os

import uvicorn
from osmosis_ai.rollout import (
    AgentWorkflow,
    AgentWorkflowContext,
    AgentWorkflowOutput,
    Grader,
    GraderContext,
    LocalBackend,
)
from osmosis_ai.rollout.server import create_rollout_server


class EchoWorkflow(AgentWorkflow):
    async def run(self, ctx: AgentWorkflowContext) -> AgentWorkflowOutput:
        return AgentWorkflowOutput(
            messages=[*ctx.prompt, {"role": "assistant", "content": "ready"}]
        )


class ReadyGrader(Grader):
    async def grade(self, ctx: GraderContext) -> None:
        if ctx.sample is None:
            raise ValueError("workflow produced no sample")
        ctx.set_reward(1.0)


def main() -> None:
    backend = LocalBackend(workflow=EchoWorkflow, grader=ReadyGrader)
    app = create_rollout_server(backend=backend)
    port = os.environ.get("_OSMOSIS_ROLLOUT_PORT")
    uvicorn.run(
        app,
        host="0.0.0.0" if port else "127.0.0.1",
        port=int(port or "8000"),
        timeout_keep_alive=2160,
    )


if __name__ == "__main__":
    main()

python main.py