API

class towhee.serve.api_service.APIService(*, desc: str = '', routers: List[RouterConfig] | None = None)

Bases: BaseModel

APIService is used to define a service. After the definition is completed, you can use HTTPServer or the GRPCServer to started it.

Example:

import typing as T
from pydantic import BaseModel
from towhee import AutoPipes, api_service
from towhee.serve.io import JSON

service = api_service.APIService(desc="Welcome")
stn = AutoPipes.pipeline('sentence_embedding')

@service.api(path='/embedding')
def chat(params: T.List[T.Any]):
    return stn(*params).to_list()

class Item(BaseModel):
    url: str
    ids: T.List[int]

@service.api(path='/echo', input_model = JSON(Item), output_model=JSON(Item))
def echo(item: Item):
    return item

if __name__ == '__main__':
    import sys
    if sys.argv[1] == 'http':
        from towhee.serve.http.server import HTTPServer
        HTTPServer(service).run('0.0.0.0', 8000)
    else:
        import logging
        from towhee.utils.log import engine_log
        engine_log.setLevel(logging.INFO)
        from towhee.serve.grpc.server import GRPCServer
        GRPCServer(service).run('0.0.0.0', 8000)

Client Example:
    http:
        import requests
        requests.post('http://127.0.0.1:8000/echo', json={'url': 1, 'ids': [1, 2]}).json()
        requests.post('http://127.0.0.1:8000/embedding', json=['hello world']).json()

    grpc:
        from towhee.serve.grpc.client import Client
        c = Client('0.0.0.0', 8000)
        c('/echo', {'url': 1, 'ids': [1, 2]})
        c('/embedding', ['hello'])

towhee.serve.api_service.build_service(pipelines: Tuple[RuntimPipeline, str] | List[Tuple[RuntimPipeline, str]], desc: str = 'Welcome to use towhee pipeline service')

Convert multiple pipelines into an APIService

Examples:

from towhee import AutoPipes
from towhee import api_service

stn = AutoPipes.pipeline('sentence_embedding')
img = AutoPipes.pipeline('image-embedding')
service = api_service.build_service([(stn, '/embedding/text'), (img, '/embedding/image')])

if __name__ == '__main__':
    from towhee.serve.http.server import HTTPServer
    HTTPServer(service).run('0.0.0.0', 8000)

class towhee.serve.io.JSON(model: BaseModel | None = None)

Bases: IOBase

JSON IO

Example:

import typing as T
from pydantic import BaseModel
from towhee import AutoPipes, api_service
from towhee.serve.io import JSON

service = api_service.APIService(desc="Welcome")
stn = AutoPipes.pipeline('sentence_embedding')

@service.api(path='/embedding', input_model = JSON(), output_model=JSON())
def chat(params: T.List[T.Any]):
    return stn(*params).to_list()

class Item(BaseModel):
    url: str
    ids: T.List[int]

@service.api(path='/echo', input_model = JSON(Item), output_model=JSON(Item))
def echo(item: Item):
    return item

Client Example:
    http:

    .. code-block:: python

        import requests
        requests.post('http://127.0.0.1:8000/echo', json={'url': 1, 'ids': [1, 2]}).json()
        requests.post('http://127.0.0.1:8000/embedding', json=['hello world']).json()

    grpc:

    .. code-block:: python

        from towhee.serve.grpc.client import Client
        c = Client('0.0.0.0', 8000)
        c('/echo', {'url': 1, 'ids': [1, 2]})
        c('/embedding', ['hello'])

class towhee.serve.io.TEXT

Bases: IOBase

TEXT IO Example:

from towhee import AutoPipes, api_service
from towhee.serve.io import TEXT, NDARRAY, JSON, BYTES

service = api_service.APIService(desc="Welcome")
stn = AutoPipes.pipeline('sentence_embedding')

@service.api(path='/embedding', input_model=TEXT(), output_model=NDARRAY())
def chat(text: str):
    return stn(text).to_list()[0][0]

Client:

HTTP:

import requests
requests.post('http://127.0.0.1:8000/embedding', json='Hello, Towhee').json()

GRPC:

from towhee.serve.grpc.client import Client
c = Client('0.0.0.0', 8000)
print(c('/embedding', 'Hello, Towhee '))

class towhee.serve.io.BYTES

Bases: IOBase

BYTES IO. Example:

from towhee import AutoPipes, api_service
from towhee.serve.io import TEXT, NDARRAY, JSON, BYTES

service = api_service.APIService(desc="Welcome")
img_emb = AutoPipes.pipeline('image-embedding')

@service.api(path='/embedding', input_model=BYTES(), output_model=NDARRAY())
def chat(image_bytes: bytes):
    return img_emb(image_bytes).to_list()[0][0]

Client:

HTTP:

import requests
with open(image_path, 'rb') as f:
    print(requests.post('http://127.0.0.1:8000/embedding', data=f.read()).json())

GRPC:

from towhee.serve.grpc.client import Client
c = Client('0.0.0.0', 8000)
with open('/home/junjie.jiangjjj/images/1.png', 'rb') as f:
    c('/embedding', r.read())

class towhee.serve.io.NDARRAY

Bases: IOBase

Numpy Ndarray

Example:

import typing as T
from pydantic import BaseModel
from towhee import AutoPipes, api_service
from towhee.serve.io import TEXT, NDARRAY

service = api_service.APIService(desc="Welcome")
stn = AutoPipes.pipeline('sentence_embedding')

@service.api(path='/embedding', input_model=TEXT(), output_model=NDARRAY())
def chat(text: str):
    return stn(text).to_list()[0][0]

class towhee.serve.http.server.HTTPServer(api_service: APIService)

Bases: object

An HTTP server implemented based on FastAPI

class towhee.serve.grpc.server.GRPCServer(api_service: APIService, max_workers: int = 20)

Bases: object