Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions cli/cmd/geniex/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ go_library(
"main.go",
"model.go",
"run.go",
"run_history.go",
"serve.go",
"update.go",
"version.go",
Expand Down Expand Up @@ -46,6 +47,8 @@ go_cgo_test(
srcs = [
"infer_test.go",
"model_test.go",
"run_history_test.go",
"run_test.go",
],
embed = [":geniex_lib"],
gotags = [
Expand All @@ -54,6 +57,9 @@ go_cgo_test(
],
deps = [
"//cli/internal/testutil",
"@com_github_openai_openai_go_v3//:openai-go",
"@com_github_openai_openai_go_v3//option",
"@com_github_openai_openai_go_v3//packages/ssestream",
],
)

Expand Down
179 changes: 106 additions & 73 deletions cli/cmd/geniex/run.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,7 @@ func tagStreamError(err error) error {
var body struct {
Code *int32 `json:"code"`
}
// code < 0 means the server error was not an SDKError (SDKErrorCode
// returns -1); keep the original stream error in that case.
if sonic.Unmarshal(se.Event.Data, &body) == nil && body.Code != nil && *body.Code >= 0 {
if sonic.Unmarshal(se.Event.Data, &body) == nil && body.Code != nil && *body.Code != -1 {
return geniex_sdk.SDKError(*body.Code)
}
}
Expand Down Expand Up @@ -127,23 +125,31 @@ func runCompletions(ctx context.Context, name string, modelType geniex_sdk.Model

// repl
var history []openai.ChatCompletionMessageParamUnion
protected := 0
if systemPrompt != "" {
history = append(history, openai.SystemMessage(systemPrompt))
protected = 1
}

processor := &common.Processor{
ParseFile: modelType == geniex_sdk.ModelTypeVLM,
Verbose: verbose,
TestMode: testMode,
Reset: func() error {
history = nil
_, err := client.Chat.Completions.New(ctx, openai.ChatCompletionNewParams{
Messages: nil,
Model: name,
})
if err == nil {
history = history[:0]
if protected == 1 {
history = append(history, openai.SystemMessage(systemPrompt))
}
}
return err
},
Run: func(prompt string, images, audios []string, onToken func(string) bool) (string, geniex_sdk.ProfileData, error) {
var userMessage openai.ChatCompletionMessageParamUnion
if len(images) > 0 || len(audios) > 0 {
contents := make([]openai.ChatCompletionContentPartUnionParam, 0)
contents = append(contents, openai.ChatCompletionContentPartUnionParam{
Expand All @@ -169,87 +175,114 @@ func runCompletions(ctx context.Context, name string, modelType geniex_sdk.Model
},
})
}
history = append(history, openai.UserMessage(contents))
userMessage = openai.UserMessage(contents)
} else {
history = append(history, openai.UserMessage(prompt))
userMessage = openai.UserMessage(prompt)
}
pending := append(append([]openai.ChatCompletionMessageParamUnion(nil), history...), userMessage)
type attemptResult struct {
text string
profile geniex_sdk.ProfileData
complete bool
}
var emitted bool
result, kept, dropped, err := retryChatHistory(pending, protected,
func(messages []openai.ChatCompletionMessageParamUnion) (attemptResult, error) {
emitted = false

start := time.Now()
acc := openai.ChatCompletionAccumulator{}
stream := client.Chat.Completions.NewStreaming(ctx, openai.ChatCompletionNewParams{
Messages: history,
Model: name,
StreamOptions: openai.ChatCompletionStreamOptionsParam{IncludeUsage: openai.Opt(true)},
Temperature: openai.Float(float64(temperature)),
TopP: openai.Float(float64(topP)),
PresencePenalty: openai.Float(float64(presencePenalty)),
FrequencyPenalty: openai.Float(float64(frequencyPenalty)),
Seed: openai.Int(int64(seed)),
MaxCompletionTokens: openai.Int(int64(maxTokens)),
},
start := time.Now()
acc := openai.ChatCompletionAccumulator{}
stream := client.Chat.Completions.NewStreaming(ctx, openai.ChatCompletionNewParams{
Messages: messages,
Model: name,
StreamOptions: openai.ChatCompletionStreamOptionsParam{IncludeUsage: openai.Opt(true)},
Temperature: openai.Float(float64(temperature)),
TopP: openai.Float(float64(topP)),
PresencePenalty: openai.Float(float64(presencePenalty)),
FrequencyPenalty: openai.Float(float64(frequencyPenalty)),
Seed: openai.Int(int64(seed)),
MaxCompletionTokens: openai.Int(int64(maxTokens)),
},

option.WithJSONSet("enable_think", enableThink),
option.WithJSONSet("top_k", topK),
option.WithJSONSet("min_p", minP),
option.WithJSONSet("repetition_penalty", repetitionPenalty),
option.WithJSONSet("grammar_path", grammarPath),
option.WithJSONSet("grammar_string", grammarString),
option.WithJSONSet("ngl", ngl),
option.WithJSONSet("nctx", nctx),
option.WithJSONSet("compute", computeUnit),
option.WithJSONSet("vit_compute", vitComputeUnit),
option.WithJSONSet("spec_type", specType),
option.WithJSONSet("spec_draft_model", draftModel),
option.WithJSONSet("spec_n_max", draftTokens),
option.WithJSONSet("spec_n_min", draftMin),
option.WithJSONSet("spec_p_min", draftPMin),
option.WithJSONSet("power_mode", powerMode))
option.WithJSONSet("enable_think", enableThink),
option.WithJSONSet("top_k", topK),
option.WithJSONSet("min_p", minP),
option.WithJSONSet("repetition_penalty", repetitionPenalty),
option.WithJSONSet("grammar_path", grammarPath),
option.WithJSONSet("grammar_string", grammarString),
option.WithJSONSet("ngl", ngl),
option.WithJSONSet("nctx", nctx),
option.WithJSONSet("compute", computeUnit),
option.WithJSONSet("vit_compute", vitComputeUnit),
option.WithJSONSet("spec_type", specType),
option.WithJSONSet("spec_draft_model", draftModel),
option.WithJSONSet("spec_n_max", draftTokens),
option.WithJSONSet("spec_n_min", draftMin),
option.WithJSONSet("spec_p_min", draftPMin),
option.WithJSONSet("power_mode", powerMode))
defer stream.Close()

var firstToken time.Time
var profileData geniex_sdk.ProfileData
for stream.Next() {
if firstToken.IsZero() {
firstToken = time.Now()
}
var firstToken time.Time
var profileData geniex_sdk.ProfileData
for stream.Next() {
if firstToken.IsZero() {
firstToken = time.Now()
}

chunk := stream.Current()
acc.AddChunk(chunk)
if len(chunk.Choices) > 0 {
if !onToken(chunk.Choices[0].Delta.Content) {
stream.Close()
break
chunk := stream.Current()
acc.AddChunk(chunk)
if len(chunk.Choices) > 0 {
content := chunk.Choices[0].Delta.Content
if content != "" {
emitted = true
}
if !onToken(content) {
stream.Close()
break
}
}
if chunk.Usage.PromptTokens > 0 {
profileData.PromptTokens = chunk.Usage.PromptTokens
profileData.GeneratedTokens = chunk.Usage.CompletionTokens
det := chunk.Usage.CompletionTokensDetails
profileData.DraftNAccepted = det.AcceptedPredictionTokens
profileData.DraftNTotal = det.AcceptedPredictionTokens + det.RejectedPredictionTokens
}
}
}
if chunk.Usage.PromptTokens > 0 {
profileData.PromptTokens = chunk.Usage.PromptTokens
profileData.GeneratedTokens = chunk.Usage.CompletionTokens
det := chunk.Usage.CompletionTokensDetails
profileData.DraftNAccepted = det.AcceptedPredictionTokens
profileData.DraftNTotal = det.AcceptedPredictionTokens + det.RejectedPredictionTokens
}
}

// zero token generated
if firstToken.IsZero() {
firstToken = time.Now()
}
// zero token generated
if firstToken.IsZero() {
firstToken = time.Now()
}

end := time.Now()
profileData.TTFT = firstToken.Sub(start).Microseconds()
profileData.PromptTime = profileData.TTFT
profileData.DecodeTime = end.Sub(firstToken).Microseconds()
profileData.DecodingSpeed = float64(profileData.GeneratedTokens) / float64(end.Sub(firstToken).Seconds())
end := time.Now()
profileData.TTFT = firstToken.Sub(start).Microseconds()
profileData.PromptTime = profileData.TTFT
profileData.DecodeTime = end.Sub(firstToken).Microseconds()
profileData.DecodingSpeed = float64(profileData.GeneratedTokens) / float64(end.Sub(firstToken).Seconds())

if stream.Err() != nil {
return "", profileData, tagStreamError(stream.Err())
}
if stream.Err() != nil {
return attemptResult{profile: profileData}, tagStreamError(stream.Err())
}

if len(acc.Choices) > 0 {
history = append(history, openai.AssistantMessage(acc.Choices[0].Message.Content))
return acc.Choices[0].Message.Content, profileData, nil
}
if len(acc.Choices) > 0 {
return attemptResult{text: acc.Choices[0].Message.Content, profile: profileData, complete: true}, nil
}

return "", profileData, nil
return attemptResult{profile: profileData}, nil
}, func(err error) bool {
return !emitted && errors.Is(err, geniex_sdk.ErrLlmGenerationPromptTooLong)
})
if err != nil {
return result.text, result.profile, err
}
if dropped > 0 {
fmt.Println(render.GetTheme().Info.Sprint("Older messages were removed to fit the context window."))
}
if result.complete {
history = append(kept, openai.AssistantMessage(result.text))
}
return result.text, result.profile, nil
},
}
if len(prompt) > 0 || input != "" {
Expand Down
21 changes: 21 additions & 0 deletions cli/cmd/geniex/run_history.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
// Copyright 2024-2026 Qualcomm Technologies, Inc. and/or its subsidiaries.
// SPDX-License-Identifier: BSD-3-Clause

package main

import "github.com/openai/openai-go/v3"

func retryChatHistory[T any](messages []openai.ChatCompletionMessageParamUnion, protected int,
request func([]openai.ChatCompletionMessageParamUnion) (T, error), retryable func(error) bool,
) (T, []openai.ChatCompletionMessageParamUnion, int, error) {
dropped := 0
for {
result, err := request(messages)
if err == nil || !retryable(err) || len(messages)-protected < 3 ||
messages[protected].OfUser == nil || messages[protected+1].OfAssistant == nil {
return result, messages, dropped, err
}
messages = append(messages[:protected:protected], messages[protected+2:]...)
dropped++
}
}
94 changes: 94 additions & 0 deletions cli/cmd/geniex/run_history_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
// Copyright 2024-2026 Qualcomm Technologies, Inc. and/or its subsidiaries.
// SPDX-License-Identifier: BSD-3-Clause

package main

import (
"errors"
"testing"

"github.com/openai/openai-go/v3"
)

func TestRetryChatHistory(t *testing.T) {
tooLong := errors.New("prompt too long")
messages := []openai.ChatCompletionMessageParamUnion{
openai.SystemMessage("rules"),
openai.UserMessage("old"), openai.AssistantMessage("answer"),
openai.UserMessage("new"),
}
calls := 0
result, kept, dropped, err := retryChatHistory(messages, 1,
func(got []openai.ChatCompletionMessageParamUnion) (string, error) {
calls++
if calls == 1 {
return "", tooLong
}
if len(got) != 2 || got[0].OfSystem == nil || got[1].OfUser == nil {
t.Fatalf("retry lost system or current user: %+v", got)
}
return "ok", nil
}, func(err error) bool { return errors.Is(err, tooLong) })
if err != nil || result != "ok" || dropped != 1 || len(kept) != 2 || len(messages) != 4 {
t.Fatalf("result=%q kept=%d dropped=%d calls=%d err=%v", result, len(kept), dropped, calls, err)
}
}

func TestRetryChatHistoryStopsAtCurrentTurn(t *testing.T) {
tooLong := errors.New("prompt too long")
for _, protected := range []int{0, 1} {
messages := []openai.ChatCompletionMessageParamUnion{openai.UserMessage("current")}
if protected == 1 {
messages = append([]openai.ChatCompletionMessageParamUnion{openai.SystemMessage("rules")}, messages...)
}
calls := 0
_, kept, dropped, err := retryChatHistory(messages, protected,
func([]openai.ChatCompletionMessageParamUnion) (struct{}, error) {
calls++
return struct{}{}, tooLong
}, func(err error) bool { return errors.Is(err, tooLong) })
if !errors.Is(err, tooLong) || calls != 1 || dropped != 0 || len(kept) != len(messages) {
t.Fatalf("protected=%d calls=%d dropped=%d err=%v", protected, calls, dropped, err)
}
}
}

func TestRetryChatHistoryDropsMultipleTurns(t *testing.T) {
tooLong := errors.New("prompt too long")
messages := []openai.ChatCompletionMessageParamUnion{
openai.UserMessage("first"), openai.AssistantMessage("one"),
openai.UserMessage("second"), openai.AssistantMessage("two"),
openai.UserMessage("current"),
}
calls := 0
_, kept, dropped, err := retryChatHistory(messages, 0,
func(got []openai.ChatCompletionMessageParamUnion) (struct{}, error) {
calls++
if calls < 3 {
return struct{}{}, tooLong
}
if len(got) != 1 || got[0].OfUser == nil {
t.Fatalf("current user turn was lost: %+v", got)
}
return struct{}{}, nil
}, func(err error) bool { return errors.Is(err, tooLong) })
if err != nil || calls != 3 || dropped != 2 || len(kept) != 1 || len(messages) != 5 {
t.Fatalf("calls=%d dropped=%d kept=%d err=%v", calls, dropped, len(kept), err)
}
}

func TestRetryChatHistoryKeepsHistoryOnOtherErrors(t *testing.T) {
failure := errors.New("network failure")
messages := []openai.ChatCompletionMessageParamUnion{
openai.UserMessage("old"), openai.AssistantMessage("answer"), openai.UserMessage("new"),
}
calls := 0
_, kept, dropped, err := retryChatHistory(messages, 0,
func([]openai.ChatCompletionMessageParamUnion) (struct{}, error) {
calls++
return struct{}{}, failure
}, func(error) bool { return false })
if !errors.Is(err, failure) || calls != 1 || dropped != 0 || len(kept) != len(messages) {
t.Fatalf("calls=%d dropped=%d err=%v", calls, dropped, err)
}
}
Loading