yomorun/yomo

🦖 Stateful Serverless Framework for building Geo-distributed Edge AI Infra

YoMo

YoMo is an open-source LLM Function Calling Framework for building Geo-distributed AI applications. Built atop QUIC Transport Protocol and Stateful Serverless architecture, makes your AI application low-latency, reliable, secure, and easy.

💚 We care about: Customer Experience in the Age of AI

🌶 Features

	Features
⚡️	Low-latency Guaranteed by implementing atop QUIC QUIC
🔐	Security TLS v1.3 on every data packet by design
📸	Stateful Serverless Make your GPU serverless 10x faster
🌎	Geo-Distributed Architecture Brings AI inference closer to end users
🚀	Y3 a faster than real-time codec

🚀 Getting Started

Let's implement a function calling with sfn-currency-converter:

Step 1. Install CLI

curl -fsSL https://get.yomo.run | sh

Verify if the CLI was installed successfully

yomo version

Step 2. Start the server

Prepare the configuration as my-agent.yaml

name: ai-zipper
host: 0.0.0.0
port: 9000

auth:
  type: token
  token: SECRET_TOKEN

bridge:
  ai:
    server:
      addr: 0.0.0.0:8000 ## Restful API endpoint
      provider: openai ## LLM API Service we will use

    providers:
      azopenai:
        api_endpoint: https://<RESOURCE>.openai.azure.com
        deployment_id: <DEPLOYMENT_ID>
        api_key: <API_KEY>
        api_version: <API_VERSION>

      openai:
        api_key: sk-xxxxxxxxxxxxxxxxxxxxxxxxxxx
        model: gpt-4-1106-preview

      gemini:
        api_key: <GEMINI_API_KEY>

      cloudflare_azure:
        endpoint: https://gateway.ai.cloudflare.com/v1/<CF_GATEWAY_ID>/<CF_GATEWAY_NAME>
        api_key: <AZURE_API_KEY>
        resource: <AZURE_OPENAI_RESOURCE>
        deployment_id: <AZURE_OPENAI_DEPLOYMENT_ID>
        api_version: 2023-12-01-preview

Start the server:

YOMO_LOG_LEVEL=debug yomo serve -c my-agent.yaml

Step 3. Write the function

First, let's define what this function do and how's the parameters required, these will be combined to prompt when invoking LLM.

type Parameter struct {
	Domain string `json:"domain" jsonschema:"description=Domain of the website,example=example.com"`
}

func Description() string {
	return `if user asks ip or network latency of a domain, you should return the result of the giving domain. try your best to dissect user expressions to infer the right domain names`
}

func InputSchema() any {
	return &Parameter{}
}

Create a Stateful Serverless Function to get the IP and Latency of a domain:

func handler(ctx serverless.Context) {
	fc, _ := ai.ParseFunctionCallContext(ctx)

	var msg Parameter
	fc.UnmarshalArguments(&msg)

	// get ip of the domain
	ips, _ := net.LookupIP(msg.Domain)

	// get ip[0] ping latency
	pinger, _ := ping.NewPinger(ips[0].String())
	pinger.Count = 3
	pinger.Run()
	stats := pinger.Statistics()

	val := fmt.Sprintf("domain %s has ip %s with average latency %s", msg.Domain, ips[0], stats.AvgRtt)
	fc.Write(val)
}

Finally, let's run it

$ go run main.go

time=2024-03-19T21:43:30.583+08:00 level=INFO msg="connected to zipper" component=StreamFunction sfn_id=B0ttNSEKLSgMjXidB11K1 sfn_name=fn-get-ip-from-domain zipper_addr=localhost:9000
time=2024-03-19T21:43:30.584+08:00 level=INFO msg="register ai function success" component=StreamFunction sfn_id=B0ttNSEKLSgMjXidB11K1 sfn_name=fn-get-ip-from-domain zipper_addr=localhost:9000 name=fn-get-ip-from-domain tag=16

Done, let's have a try

$ curl -i -X POST -H "Content-Type: application/json" -d '{"prompt":"compare nike and puma website speed"}' http://127.0.0.1:8000/invoke
HTTP/1.1 200 OK
Content-Length: 944
Connection: keep-alive
Content-Type: application/json
Date: Tue, 19 Mar 2024 13:30:14 GMT
Keep-Alive: timeout=4
Proxy-Connection: keep-alive

{
  "Content": "Based on the data provided for the domains nike.com and puma.com which include IP addresses and average latencies, we can infer the following about their website speeds:
  - Nike.com has an IP address of 13.225.183.84 with an average latency of 65.568333 milliseconds.
  - Puma.com has an IP address of 151.101.194.132 with an average latency of 54.563666 milliseconds.
  
  Comparing these latencies, Puma.com is faster than Nike.com as it has a lower average latency. 
  
  Please be aware, however, that website speed can be influenced by many factors beyond latency, such as server processing time, content size, and delivery networks among others. To get a more comprehensive understanding of website speed, you would need to consider additional metrics and possibly conductreal-time speed tests.",
  "FinishReason": "stop"
}

Full Example Code

Full LLM Function Calling Codes

📚 Documentation

Read more about YoMo at yomo.run/docs.

YoMo ❤️ Vercel, our documentation website is

🎯 Focuses on Geo-distributed AI Inference Infra

It’s no secret that today’s users want instant AI inference, every AI application is more powerful when it response quickly. But, currently, when we talk about distribution, it represents distribution in data center. The AI model is far away from their users from all over the world.

If an application can be deployed anywhere close to their end users, solve the problem, this is Geo-distributed System Architecture:

🦸 Contributing

First off, thank you for considering making contributions. It's people like you that make YoMo better. There are many ways in which you can participate in the project, for example:

• File a bug report. Be sure to include information like what version of YoMo you are using, what your operating system is, and steps to recreate the bug.
• Suggest a new feature.
• Read our contributing guidelines to learn about what types of contributions we are looking for.
• We have also adopted a code of conduct that we expect project participants to adhere to.

License

Apache License 2.0

yomo's Issues

Refactor dev server for `yomo dev`, support yomo version matching

When developers write Rx code, logic can be implemented easily and quickly, when they run yomo dev, they get test data streams directly, especially for user onboarding. after v2.0, we consider provide a way for custom data types.

Count the amount of `data frame` in zipper

Count the amount of data frame when zipper receiving the data frame from source, and record it in a local file.

Log the connected/disconnected information in zipper

When source, sfn, and upstream zipper are connected/disconnected to zipper, we need to log them in Info level.

The absence of `yomo-function` configuration section can be accepted by `yomo-server`

As YoMo Server can be started by the -m parameter for loading a remote configuration, a YoMo Server w/o functions section should work.

When functions section present, only listed yomo-functions can be connected to.

failed to sufficiently increase receive buffer size when `yomo run` on Manjaro

Summary

Run yomo dev on Manjaro, failed to sufficiently increase receive buffer size error.

Steps To Reproduce

~/.../yomo/example >>> ../bin/yomo dev app.go
2020/12/21 13:01:37 Building the Serverless Function File...
2020/12/21 13:01:42 failed to sufficiently increase receive buffer size (was: 208 kiB, wanted: 2048 kiB, got: 416 kiB). See https://github.com/lucas-clemente/quic-go/wiki/UDP-Receive-Buffer-Size for details.
2020/12/21 13:01:42 QUIC Server listens on 0.0.0.0:4242
^C

Environments

~/.../yomo/example >>> cat /etc/lsb-release
DISTRIB_ID=ManjaroLinux
DISTRIB_RELEASE=20.2
DISTRIB_CODENAME=Nibia
DISTRIB_DESCRIPTION="Manjaro Linux"

Solution

Run sudo sysctl -w net.core.rmem_max=2500000 to solve this problem followed by quic-go wiki

`/docs/source` for how to write a `yomo-source`

Y3 Codec Draft-02

Simplify QUIC client & server

In order to decrease the code complexity, we need to remove the customized QUIC interfaces in core/quic, and directly use quic-go as QUIC client & server.
See skeleton project for reference.

Adding a description of contributing

Add contributing file about yomo project

Update documentation

Update the documentations on yomo.run to fit the new changes in #173 #174

update codec to v1.1.0

• Solve the problem of base type encoding without wrapper node

use data frame between zipper and sfn

Currently, zipper and sfn are communicating with raw bytes, we need to replace it with Data Frame.

`/docs` show the index page of docs

update codec to v1.0.0

• optimising performance
• sticky packet handling

`/docs/flow` for how to write a streaming serverless application

zip 3 or more streams

I thought it would be straightforward, but I'm facing some difficulties...

I have a data source app that sends 5 packets encoded with 5 different keys:

func generateAndSendData(stream quic.Stream) {
    keys := []byte{0x10, 0x11, 0x13, 0x7E, 0x7F}

    for i, key := range keys {
        time.Sleep(100 * time.Millisecond)
        
        codec := y3.NewCodec(key)

        sendingBuf, _ := codec.Marshal(int64(i))

        _, err := stream.Write(sendingBuf)
        if err != nil {
            log.Printf("Couldn't send buffer with i=%v", i)
        } else {
            fmt.Print(".")
        }
    }
}

And the flow app that has 5 streams that are subscribed to each individual key and then zipped together:

var zipper = func(_ context.Context, a interface{}, b interface{}) (interface{}, error) {
    accumulator, ok := a.([]interface{})
    if !ok {
        fmt.Printf("No accumulator: %v + %v\n", a, b)
        return []interface{}{a, b}, nil
    }

    fmt.Printf("With accumulator: %v + %v\n", accumulator, b)
    accumulator = append(accumulator, b)
    return accumulator, nil
}

var convert = func(v []byte) (interface{}, error) {
    fmt.Printf("Got: %v\n", v)
	return y3.ToInt64(v)
}

// Handler will handle data in Rx way
func Handler(rxstream rx.RxStream) rx.RxStream {
	streamA:= rxstream.Subscribe(0x10).OnObserve(convert)
	streamB:= rxstream.Subscribe(0x11).OnObserve(convert)
	streamC:= rxstream.Subscribe(0x13).OnObserve(convert)
	streamD:= rxstream.Subscribe(0x7E).OnObserve(convert)
	streamE:= rxstream.Subscribe(0x7F).OnObserve(convert)

    return streamA.
        ZipFromIterable(streamB, zipper).
        ZipFromIterable(streamC, zipper).
        ZipFromIterable(streamD, zipper).
        ZipFromIterable(streamE, zipper).
        StdOut().
	Encode(0x11)
}

The problem, is that only the first two packets sent by the source are received on the flow. The order of zipping doesn't matter, the order of sending by the source matters only, for what I can see...

What can I do? (Also, I'm zipping sequentially, because there's no Merge operator... Is it really missing?)

Originally posted by @kuredoro in #158

Introduce frames to YoMo

• Frames will be the minimal element transferred between services built on YoMo, frames built on Y3 codec for fast decoding
• Introduce HandshakeFrame
• Introduce PayloadFrame with MetaFrame and DataFrame
• Introduce TransactionID to MetaFrame, improve traceability

Print `UDP Receive Buffer Size` in CLI

When running the CLI, we need to print UDP Receive Buffer Size in the current OS and validate if it doesn't fit the recommended size.

Streaming-serverless arch branch for community feedbacks

Feedbacks from the community, developers consider writing Serverless functions rather than implementing & maintaining microservices on the edge. Consider build a version on feat/sl branch for this.

Simplify workflow.yaml

We will rename source/flow/sink in #173, and it's also not needed to place sinks in workflow.yaml.

• Remove sinks
• Rename flows to functions
• Connect output connectors to yomo and receive the output data in parallel.

name: Service
host: localhost
port: 9000
functions:
  - name: func1
  - name: func2

[BUG] plugin different version of package

Summary (概要)

when I use cmd "yomo dev", I got some problem

Steps To Reproduce (复现的步骤)

1.mkdir demo
2.open my IDE goland to open my demo project
3.go mod tidy
4.yomo dev

Additional Details & Logs (附加的信息和日志)

• Yomo version (版本)1.0.0-3gf175e41 (2021-03-10)
• Yomo error logs (错误日志)Load handle from .so file failure with error: plugin.Open("sl"): plugin was built with a different version of package github.com/yomorun/yomo/pkg/yy3
• Yomo configuration (配置) I got the issue when I enter the cmd "yomo dev",so there`s probably is yomo init config
• Operating system (操作系统)MacOS Catalina v10.15.7

`/docs/tutorial` for `source->flow->sink(faunaDB)`

[BUG] Multiple streams between zipper and flow

In some cases, when the connection was interrupted between yomo-zipper and yomo-flow, a new stream will be created. And currently yomo-zipper keeps multiple streams for all yomo-flow, yomo-sink and yomo-source.
It causes a performance issue when there are too many streams between yomo-zipper and yomo-flow/sink.

But actually, yomo-zipper should only support multiple streams for yomo-source, and one stream for yomo-flow/ yomo-sink.

high cpu

Summary (概要)

After running yomo dev for 2 minutes, the cpu is 100%

Separate CLI and Framework

By getting rid of Go plugins, is friendly to developers:

• never need clone whole project to $GOPATH
• Windows developers can use yomo dev xxx.go

Replace logger from skeleton project

It's inconvenient to use the current logger, we need to replace it with the one from skeleton project.

Reusable Server-Sent Events connector for frontend

about Server-Sent Events: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events

Rename source/flow/sink

It is difficult and confused for users to understand. the source/flow/sink concepts in YoMo.
It's better to simplify the naming of source/flow/sink/zipper.

Connectors

Connectors provides the common and reusable services for input/output data in YoMo.
The users can import the YoMo SDK to implement the input/output connector and add the main() function. Then you can also run the connectors by Go cli, f.e. go run main.go.

Input Connector

Input connector sends the input data from the external system (f.e. IoT Sensors) to yomo.

• Rename source to input connector.

Output Connector

Output connector sends the output data from yomo to an external system (f.e. database).

• Rename sink to output connector.

Stream Functions

Stream Functions are the streaming Serverless functions in YoMo, it contains a Handler function which the users can write the business logic there.
The users can use yomo cli to run the function locally or deploy it to YCloud.

• Rename flow to stream function.

YoMo Server

YoMo Server is an orchestrator which receives the data from sources (input connectors) and routes the data to Stream Functions one by one. The result of real-time stream computing will be sent to Output Connectors.

• Rename zipper to yomo server.

[BUG] CopyN 修改参数配置可引起消息阻塞

Summary (概要)

流处理使用io.CopyN，当设置较少的拷贝字节数时，会出现消息阻塞和消息切断的现象。

Steps To Reproduce (复现的步骤)

1. Prj/Yomo >> run.go >> RunDev >> 修改io.CopyN参数值:
   io.CopyN(pluginStream.Writer, deStream1.Reader, 64)

2. Prj/yomo-echo-plugin >> go run echo.go, 出现消息阻塞和消息切断：
   name:yomo!-new
   name:yom

3. 设置的值越小则情况越严重，甚至一开始就出现阻塞

4. 后续测试: 默认1024的值也会出现消息阻塞和消息切断，应该是处理消息量达到一定程度就会出现该现象。

Additional Details & Logs (附加的信息和日志)

• Yomo version (版本): v0.1.0
• Yomo error logs (错误日志): 无
• Yomo configuration (配置): io.CopyN(pluginStream.Writer, deStream1.Reader, 64)
• Operating system (操作系统) macOS

The stream should be immutable, `stream function` can only append new data to stream

• The stream from yomo-zipper should be immutable, the stream functions shouldn't change the raw stream.
  In case the upstream function needs to generate new data for downstream, it can only append the new data to the stream.
• Combine flow and sink as stream function, it will be executed one by one by zipper.

`/docs/rx` for `rxstream` methods supported

One QUIC stream between zipper and sfn

Currently, zipper and sfn creates individual QUIC streams for each data, we need to replace it with one persistent QUIC stream.

The stream.Write() should also add lock to prevent concurrent writing in the same QUIC stream.

Use 0-RTT in QUIC client

on 170ca40

replace Codec with yomo-codec-golang

• using yomo-codec-golang
• RunDev support OutputFormatter: HexString/PacketPrinter/EchoData/ThermometerData

Remove Rx in zipper

Currently, zipper is using Rx to dispatch the data from source to stream-functions, we need to remove Rx in zipper and use a clean way to dispatch the data.

CLI can lower down the difficulty when writing YoMo applications

also from the community, related to #58 , developers can save more time when they can yomo run app.go

Integrate steam-zipper to CLI

also from the community, related to #58. Developers can chain-up their yomo-flows by yomo run workflow.yaml

RxStream is better than raw QUICStream when writing streaming-processing codes.

Streaming is not like HTTP's Request/Response, by introducing ReactiveX, developers could write concise codes.

[BUG]无法成功执行执行demo中echo.go

Summary (概要)

无法成功执行执行demo中echo.go

Steps To Reproduce (复现的步骤)

1.在$HOME/go/yomo中执行go mod init yomo
2.执行go get -u github.com/yomorun/yomo
3.按文档创建echo.go
4.执行go run echo.go

Additional Details & Logs (附加的信息和日志)

• Yomo version: latest
• Yomo error logs:

command-line-arguments

./echo.go:16:14: cannot use &EchoPlugin literal (type *EchoPlugin) as type plugin.YomoObjectPlugin in argument to yomo.RunDev:
*EchoPlugin does not implement plugin.YomoObjectPlugin (missing Mold method)

• Operating system: Debian 10

Could install YoMo by curl

As many CLI can do like this, developers love copy & past 😄

Rewrite Official Website

Consider rewrite YoMo official website by Nextra.

• i18n support, English and Chinese at first

Support basic JSON codec support

on 9a46852: for WebSocket / Crawler scenario, applications always running on JSON codec, implement basic support and give examples for users get there easily.

Connectors

The input/output connectors in YoMo.

cpu 100% after ran quick-start 3~5 mins later

Summary (概要)

quick start demo make cpu 100%

Steps To Reproduce (复现的步骤)

1. write quick-start project, use yomo.Run
2. QUIC_GO_LOG_LEVEL=debug go run echo.go
3. wait 3~5 mins
4. cpu goes to 100%

Additional Details & Logs (附加的信息和日志)

• Yomo version (版本)
  v0.1.1
• Yomo error logs (错误日志)
  no error
• Yomo configuration (配置)
• Operating system (操作系统)
  macOS Big Sur 11.0 Beta (20A4300b)

Remove Rx in sfn

There are two scenarios in sfn:

1. All data are independent, the users will handle the raw data immediately when the sfn receives data from zipper. In this case, we don't need to use Rx.
2. The data is sequential, the users might need to handle a collection of data (f.e. the average value in a time window). In this case, it is suitable to use Rx.

We need to remove Rx in sfn and use a different way to support both raw bytes and rx stream.

YoMo-Zipper communication

For edge-mesh feature, YoMo-Zipper will be geo-distributed and closer to end users, all YoMo-Zippers need to communicate with each other.

Each YoMo-Zipper is both a Sender and a Receiver:

• Sender: send the output stream to other yomp-zippers in edge-mesh network.
• Receiver: receive the input stream from other yomp-zippers.

`/docs/sink` for how to write a `yomo-sink`

Add yomo to homebrew formula

Developers can install yomo by brew install yomo

Support `yomo version` to get current yomo version

Allow to set the quic-go debug level in CLI

In order to debug the QUIC transport information, we need to set QUIC_GO_LOG_LEVEL environment variable in CLI.