server.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896
  1. package llm
  2. import (
  3. "bufio"
  4. "bytes"
  5. "context"
  6. "encoding/json"
  7. "errors"
  8. "fmt"
  9. "io"
  10. "log"
  11. "log/slog"
  12. "math/rand"
  13. "net"
  14. "net/http"
  15. "os"
  16. "os/exec"
  17. "path/filepath"
  18. "runtime"
  19. "strconv"
  20. "strings"
  21. "time"
  22. "golang.org/x/sync/semaphore"
  23. "github.com/ollama/ollama/api"
  24. "github.com/ollama/ollama/format"
  25. "github.com/ollama/ollama/gpu"
  26. )
  27. type LlamaServer interface {
  28. Ping(ctx context.Context) error
  29. WaitUntilRunning(ctx context.Context) error
  30. Completion(ctx context.Context, req CompletionRequest, fn func(CompletionResponse)) error
  31. Embedding(ctx context.Context, prompt string) ([]float64, error)
  32. Tokenize(ctx context.Context, content string) ([]int, error)
  33. Detokenize(ctx context.Context, tokens []int) (string, error)
  34. Close() error
  35. EstimatedVRAM() uint64
  36. }
  37. // llmServer is an instance of the llama.cpp server
  38. type llmServer struct {
  39. port int
  40. cmd *exec.Cmd
  41. done chan error // Channel to signal when the process exits
  42. status *StatusWriter
  43. options api.Options
  44. // TODO - this should be broken down by GPU
  45. estimatedVRAM uint64 // Estimated usage of VRAM by the loaded model
  46. sem *semaphore.Weighted
  47. }
  48. func LoadModel(model string) (*GGML, error) {
  49. if _, err := os.Stat(model); err != nil {
  50. return nil, err
  51. }
  52. f, err := os.Open(model)
  53. if err != nil {
  54. return nil, err
  55. }
  56. defer f.Close()
  57. ggml, _, err := DecodeGGML(f)
  58. return ggml, err
  59. }
  60. // NewLlamaServer will run a server for the given GPUs
  61. // The gpu list must be a single family.
  62. func NewLlamaServer(gpus gpu.GpuInfoList, model string, ggml *GGML, adapters, projectors []string, opts api.Options) (LlamaServer, error) {
  63. var err error
  64. if opts.NumCtx > int(ggml.KV().ContextLength()) {
  65. slog.Warn("requested context length is greater than model max context length", "requested", opts.NumCtx, "model", ggml.KV().ContextLength())
  66. opts.NumCtx = int(ggml.KV().ContextLength())
  67. }
  68. if opts.NumCtx < 4 {
  69. opts.NumCtx = 4
  70. }
  71. cpuRunner := ""
  72. var estimatedVRAM uint64
  73. var systemMemory uint64
  74. if (len(gpus) == 1 && gpus[0].Library == "cpu") || opts.NumGPU == 0 {
  75. // TODO evaluate system memory to see if we should block the load, or force an unload of another CPU runner
  76. cpuRunner = serverForCpu()
  77. } else {
  78. if gpus[0].Library == "metal" {
  79. memInfo, err := gpu.GetCPUMem()
  80. if err != nil {
  81. slog.Error("failed to lookup system memory", "error", err)
  82. } else {
  83. systemMemory = memInfo.TotalMemory
  84. slog.Debug("system memory", "total", format.HumanBytes2(systemMemory))
  85. }
  86. }
  87. var layers int
  88. layers, estimatedVRAM = EstimateGPULayers(gpus, ggml, projectors, opts)
  89. if gpus[0].Library == "metal" && estimatedVRAM > systemMemory {
  90. // disable partial offloading when model is greater than total system memory as this
  91. // can lead to locking up the system
  92. opts.NumGPU = 0
  93. } else if opts.NumGPU < 0 && layers > 0 && gpus[0].Library != "cpu" {
  94. opts.NumGPU = layers
  95. }
  96. }
  97. // Loop through potential servers
  98. finalErr := fmt.Errorf("no suitable llama servers found")
  99. if len(adapters) > 1 {
  100. return nil, errors.New("ollama supports only one lora adapter, but multiple were provided")
  101. }
  102. availableServers := availableServers()
  103. var servers []string
  104. if cpuRunner != "" {
  105. servers = []string{cpuRunner}
  106. } else {
  107. servers = serversForGpu(gpus[0]) // All GPUs in the list are matching Library and Variant
  108. }
  109. demandLib := strings.Trim(os.Getenv("OLLAMA_LLM_LIBRARY"), "\"' ")
  110. if demandLib != "" {
  111. serverPath := availableServers[demandLib]
  112. if serverPath == "" {
  113. slog.Info(fmt.Sprintf("Invalid OLLAMA_LLM_LIBRARY %s - not found", demandLib))
  114. } else {
  115. slog.Info("user override", "OLLAMA_LLM_LIBRARY", demandLib, "path", serverPath)
  116. servers = []string{demandLib}
  117. }
  118. }
  119. if len(servers) == 0 {
  120. return nil, fmt.Errorf("no servers found for %v", gpus)
  121. }
  122. params := []string{
  123. "--model", model,
  124. "--ctx-size", fmt.Sprintf("%d", opts.NumCtx),
  125. "--batch-size", fmt.Sprintf("%d", opts.NumBatch),
  126. "--embedding",
  127. }
  128. if debug := os.Getenv("OLLAMA_DEBUG"); debug != "" {
  129. params = append(params, "--log-format", "json")
  130. } else {
  131. params = append(params, "--log-disable")
  132. }
  133. if opts.NumGPU >= 0 {
  134. params = append(params, "--n-gpu-layers", fmt.Sprintf("%d", opts.NumGPU))
  135. }
  136. if debug := os.Getenv("OLLAMA_DEBUG"); debug != "" {
  137. params = append(params, "--verbose")
  138. }
  139. if opts.MainGPU > 0 {
  140. params = append(params, "--main-gpu", fmt.Sprintf("%d", opts.MainGPU))
  141. }
  142. if len(adapters) > 0 {
  143. // TODO: applying multiple adapters is not supported by the llama.cpp server yet
  144. params = append(params, "--lora", adapters[0])
  145. }
  146. if len(projectors) > 0 {
  147. // TODO: applying multiple projectors is not supported by the llama.cpp server yet
  148. params = append(params, "--mmproj", projectors[0])
  149. }
  150. if opts.NumThread > 0 {
  151. params = append(params, "--threads", fmt.Sprintf("%d", opts.NumThread))
  152. }
  153. if !opts.F16KV {
  154. params = append(params, "--memory-f32")
  155. }
  156. if opts.UseMLock {
  157. params = append(params, "--mlock")
  158. }
  159. if !opts.UseMMap {
  160. params = append(params, "--no-mmap")
  161. }
  162. if opts.UseNUMA {
  163. params = append(params, "--numa")
  164. }
  165. // "--cont-batching", // TODO - doesn't seem to have any noticeable perf change for multiple requests
  166. numParallel := 1
  167. if onp := os.Getenv("OLLAMA_NUM_PARALLEL"); onp != "" {
  168. numParallel, err = strconv.Atoi(onp)
  169. if err != nil || numParallel <= 0 {
  170. err = fmt.Errorf("invalid OLLAMA_NUM_PARALLEL=%s must be greater than zero - %w", onp, err)
  171. slog.Error("misconfiguration", "error", err)
  172. return nil, err
  173. }
  174. }
  175. params = append(params, "--parallel", fmt.Sprintf("%d", numParallel))
  176. for i := 0; i < len(servers); i++ {
  177. dir := availableServers[servers[i]]
  178. // Find an availableServers port, retry on each iterration in case the failure was a port conflict race
  179. port := 0
  180. if a, err := net.ResolveTCPAddr("tcp", "localhost:0"); err == nil {
  181. var l *net.TCPListener
  182. if l, err = net.ListenTCP("tcp", a); err == nil {
  183. port = l.Addr().(*net.TCPAddr).Port
  184. l.Close()
  185. }
  186. }
  187. if port == 0 {
  188. slog.Debug("ResolveTCPAddr failed ", "error", err)
  189. port = rand.Intn(65535-49152) + 49152 // get a random port in the ephemeral range
  190. }
  191. finalParams := append(params, "--port", strconv.Itoa(port))
  192. pathEnv := "LD_LIBRARY_PATH"
  193. if runtime.GOOS == "windows" {
  194. pathEnv = "PATH"
  195. }
  196. // append the server directory to LD_LIBRARY_PATH/PATH
  197. libraryPaths := []string{dir}
  198. if libraryPath, ok := os.LookupEnv(pathEnv); ok {
  199. // Append our runner directory to the path
  200. // This will favor system libraries over our bundled library dependencies
  201. libraryPaths = append(filepath.SplitList(libraryPath), libraryPaths...)
  202. }
  203. // Note: we always put the dependency path first
  204. // since this was the exact version we verified for AMD GPUs
  205. // and we favor what the user had in their path
  206. if gpus[0].DependencyPath != "" {
  207. // TODO refine for multi-gpu support
  208. libraryPaths = append([]string{gpus[0].DependencyPath}, libraryPaths...)
  209. }
  210. server := filepath.Join(dir, "ollama_llama_server")
  211. if runtime.GOOS == "windows" {
  212. server = server + ".exe"
  213. }
  214. s := &llmServer{
  215. port: port,
  216. cmd: exec.Command(server, finalParams...),
  217. status: NewStatusWriter(os.Stderr),
  218. options: opts,
  219. estimatedVRAM: estimatedVRAM,
  220. sem: semaphore.NewWeighted(int64(numParallel)),
  221. }
  222. libEnv := fmt.Sprintf("%s=%s", pathEnv, strings.Join(libraryPaths, string(filepath.ListSeparator)))
  223. s.cmd.Env = append(os.Environ(), libEnv)
  224. s.cmd.Stdout = os.Stdout
  225. s.cmd.Stderr = s.status
  226. // TODO - multiple GPU selection logic...
  227. key, val := gpu.GpuInfoList(gpus).GetVisibleDevicesEnv()
  228. if key != "" {
  229. s.cmd.Env = append(s.cmd.Env, key+"="+val)
  230. }
  231. slog.Info("starting llama server", "cmd", s.cmd.String())
  232. // Log at debug as the environment is inherited and might contain sensitive information
  233. slog.Debug("subprocess", "environment", s.cmd.Env)
  234. if err = s.cmd.Start(); err != nil {
  235. msg := ""
  236. if s.status != nil && s.status.LastErrMsg != "" {
  237. msg = s.status.LastErrMsg
  238. }
  239. err = fmt.Errorf("error starting the external llama server: %v %s", err, msg)
  240. finalErr = err
  241. continue
  242. }
  243. // reap subprocess when it exits
  244. go func() {
  245. // Exit status managed via getServerStatus
  246. _ = s.cmd.Wait()
  247. }()
  248. // TODO - make sure this is all wired up correctly
  249. // if err = s.WaitUntilRunning(); err != nil {
  250. // slog.Error("error starting llama server", "server", servers[i], "error", err)
  251. // s.Close()
  252. // finalErr = err
  253. // continue
  254. // }
  255. return s, nil
  256. }
  257. slog.Error("unable to load any llama server", "error", finalErr)
  258. return nil, finalErr
  259. }
  260. func projectorMemoryRequirements(filename string) uint64 {
  261. file, err := os.Open(filename)
  262. if err != nil {
  263. return 0
  264. }
  265. defer file.Close()
  266. ggml, _, err := DecodeGGML(file)
  267. if err != nil {
  268. return 0
  269. }
  270. var mem uint64
  271. for _, layer := range ggml.Tensors().Layers() {
  272. mem += layer.size()
  273. }
  274. return mem
  275. }
  276. type ServerStatus int
  277. const ( // iota is reset to 0
  278. ServerStatusReady ServerStatus = iota
  279. ServerStatusNoSlotsAvaialble
  280. ServerStatusLoadingModel
  281. ServerStatusNotResponding
  282. ServerStatusError
  283. )
  284. func (s ServerStatus) ToString() string {
  285. switch s {
  286. case ServerStatusReady:
  287. return "llm server ready"
  288. case ServerStatusNoSlotsAvaialble:
  289. return "llm busy - no slots available"
  290. case ServerStatusLoadingModel:
  291. return "llm server loading model"
  292. case ServerStatusNotResponding:
  293. return "llm server not responding"
  294. default:
  295. return "llm server error"
  296. }
  297. }
  298. type ServerStatusResp struct {
  299. Status string `json:"status"`
  300. SlotsIdle int `json:"slots_idle"`
  301. SlotsProcessing int `json:"slots_processing"`
  302. Error string `json:"error"`
  303. }
  304. func (s *llmServer) getServerStatus(ctx context.Context) (ServerStatus, error) {
  305. // Fail fast if its exited
  306. if s.cmd.ProcessState != nil {
  307. msg := ""
  308. if s.status != nil && s.status.LastErrMsg != "" {
  309. msg = s.status.LastErrMsg
  310. }
  311. return ServerStatusError, fmt.Errorf("llama runner process no longer running: %d %s", s.cmd.ProcessState.ExitCode(), msg)
  312. }
  313. req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/health", s.port), nil)
  314. if err != nil {
  315. return ServerStatusError, fmt.Errorf("error creating GET request: %v", err)
  316. }
  317. req.Header.Set("Content-Type", "application/json")
  318. resp, err := http.DefaultClient.Do(req)
  319. if err != nil {
  320. if errors.Is(err, context.DeadlineExceeded) {
  321. return ServerStatusNotResponding, fmt.Errorf("server not responding")
  322. }
  323. return ServerStatusError, fmt.Errorf("health resp: %w", err)
  324. }
  325. defer resp.Body.Close()
  326. body, err := io.ReadAll(resp.Body)
  327. if err != nil {
  328. return ServerStatusError, fmt.Errorf("read health request: %w", err)
  329. }
  330. var status ServerStatusResp
  331. if err := json.Unmarshal(body, &status); err != nil {
  332. return ServerStatusError, fmt.Errorf("health unmarshal encode response: %w", err)
  333. }
  334. switch status.Status {
  335. case "ok":
  336. return ServerStatusReady, nil
  337. case "no slot available":
  338. return ServerStatusNoSlotsAvaialble, nil
  339. case "loading model":
  340. return ServerStatusLoadingModel, nil
  341. default:
  342. return ServerStatusError, fmt.Errorf("server error: %+v", status)
  343. }
  344. }
  345. func (s *llmServer) Ping(ctx context.Context) error {
  346. _, err := s.getServerStatus(ctx)
  347. if err != nil {
  348. slog.Debug("server unhealthy", "error", err)
  349. return err
  350. }
  351. return nil
  352. }
  353. func (s *llmServer) WaitUntilRunning(ctx context.Context) error {
  354. start := time.Now()
  355. // TODO we need to wire up a better way to detect hangs during model load and startup of the server
  356. expiresAt := time.Now().Add(10 * time.Minute) // be generous with timeout, large models can take a while to load
  357. ticker := time.NewTicker(50 * time.Millisecond)
  358. defer ticker.Stop()
  359. slog.Info("waiting for llama runner to start responding")
  360. var lastStatus ServerStatus = -1
  361. for {
  362. select {
  363. case <-ctx.Done():
  364. slog.Info("context expired before server started")
  365. return fmt.Errorf("timed out waiting for llama runner to start")
  366. case err := <-s.done:
  367. msg := ""
  368. if s.status != nil && s.status.LastErrMsg != "" {
  369. msg = s.status.LastErrMsg
  370. }
  371. return fmt.Errorf("llama runner process has terminated: %v %s", err, msg)
  372. case <-ticker.C:
  373. if time.Now().After(expiresAt) {
  374. // timeout
  375. msg := ""
  376. if s.status != nil && s.status.LastErrMsg != "" {
  377. msg = s.status.LastErrMsg
  378. }
  379. return fmt.Errorf("timed out waiting for llama runner to start: %s", msg)
  380. }
  381. if s.cmd.ProcessState != nil {
  382. msg := ""
  383. if s.status != nil && s.status.LastErrMsg != "" {
  384. msg = s.status.LastErrMsg
  385. }
  386. return fmt.Errorf("llama runner process no longer running: %d %s", s.cmd.ProcessState.ExitCode(), msg)
  387. }
  388. c, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
  389. defer cancel()
  390. status, err := s.getServerStatus(c)
  391. if err != nil && lastStatus != status {
  392. slog.Debug("server not yet available", "error", err)
  393. lastStatus = status
  394. continue
  395. }
  396. switch status {
  397. case ServerStatusLoadingModel:
  398. // TODO - this state never seems to happen with the current server.cpp code (bug?)
  399. // it doesn't respond to the health endpoint until after the model is loaded
  400. slog.Debug("loading model")
  401. case ServerStatusReady:
  402. slog.Debug(fmt.Sprintf("llama runner started in %f seconds", time.Since(start).Seconds()))
  403. return nil
  404. }
  405. }
  406. }
  407. }
  408. const jsonGrammar = `
  409. root ::= object
  410. value ::= object | array | string | number | ("true" | "false" | "null") ws
  411. object ::=
  412. "{" ws (
  413. string ":" ws value
  414. ("," ws string ":" ws value)*
  415. )? "}" ws
  416. array ::=
  417. "[" ws (
  418. value
  419. ("," ws value)*
  420. )? "]" ws
  421. string ::=
  422. "\"" (
  423. [^"\\] |
  424. "\\" (["\\/bfnrt] | "u" [0-9a-fA-F] [0-9a-fA-F] [0-9a-fA-F] [0-9a-fA-F]) # escapes
  425. )* "\"" ws
  426. number ::= ("-"? ([0-9] | [1-9] [0-9]*)) ("." [0-9]+)? ([eE] [-+]? [0-9]+)? ws
  427. # Optional space: by convention, applied in this grammar after literal chars when allowed
  428. ws ::= ([ \t\n] ws)?
  429. `
  430. const maxBufferSize = 512 * format.KiloByte
  431. const maxRetries = 3
  432. type ImageData struct {
  433. Data []byte `json:"data"`
  434. ID int `json:"id"`
  435. }
  436. type completion struct {
  437. Content string `json:"content"`
  438. Model string `json:"model"`
  439. Prompt string `json:"prompt"`
  440. Stop bool `json:"stop"`
  441. Timings struct {
  442. PredictedN int `json:"predicted_n"`
  443. PredictedMS float64 `json:"predicted_ms"`
  444. PromptN int `json:"prompt_n"`
  445. PromptMS float64 `json:"prompt_ms"`
  446. }
  447. }
  448. type CompletionRequest struct {
  449. Prompt string
  450. Format string
  451. Images []ImageData
  452. Options api.Options
  453. }
  454. type CompletionResponse struct {
  455. Content string
  456. Done bool
  457. PromptEvalCount int
  458. PromptEvalDuration time.Duration
  459. EvalCount int
  460. EvalDuration time.Duration
  461. }
  462. func (s *llmServer) Completion(ctx context.Context, req CompletionRequest, fn func(CompletionResponse)) error {
  463. if err := s.sem.Acquire(ctx, 1); err != nil {
  464. slog.Error("Failed to acquire semaphore", "error", err)
  465. return err
  466. }
  467. defer s.sem.Release(1)
  468. request := map[string]any{
  469. "prompt": req.Prompt,
  470. "stream": true,
  471. "n_predict": req.Options.NumPredict,
  472. "n_keep": req.Options.NumKeep,
  473. "main_gpu": req.Options.MainGPU,
  474. "temperature": req.Options.Temperature,
  475. "top_k": req.Options.TopK,
  476. "top_p": req.Options.TopP,
  477. "tfs_z": req.Options.TFSZ,
  478. "typical_p": req.Options.TypicalP,
  479. "repeat_last_n": req.Options.RepeatLastN,
  480. "repeat_penalty": req.Options.RepeatPenalty,
  481. "presence_penalty": req.Options.PresencePenalty,
  482. "frequency_penalty": req.Options.FrequencyPenalty,
  483. "mirostat": req.Options.Mirostat,
  484. "mirostat_tau": req.Options.MirostatTau,
  485. "mirostat_eta": req.Options.MirostatEta,
  486. "penalize_nl": req.Options.PenalizeNewline,
  487. "seed": req.Options.Seed,
  488. "stop": req.Options.Stop,
  489. "image_data": req.Images,
  490. "cache_prompt": true,
  491. }
  492. // Make sure the server is ready
  493. status, err := s.getServerStatus(ctx)
  494. if err != nil {
  495. return err
  496. } else if status != ServerStatusReady {
  497. return fmt.Errorf("unexpected server status: %s", status.ToString())
  498. }
  499. if req.Format == "json" {
  500. request["grammar"] = jsonGrammar
  501. if !strings.Contains(strings.ToLower(req.Prompt), "json") {
  502. slog.Warn("Prompt does not specify that the LLM should response in JSON, but JSON format is expected. For best results specify that JSON is expected in the system prompt.")
  503. }
  504. }
  505. retryDelay := 100 * time.Microsecond
  506. for retries := 0; retries < maxRetries; retries++ {
  507. if retries > 0 {
  508. time.Sleep(retryDelay) // wait before retrying
  509. retryDelay *= 2 // exponential backoff
  510. }
  511. // Handling JSON marshaling with special characters unescaped.
  512. buffer := &bytes.Buffer{}
  513. enc := json.NewEncoder(buffer)
  514. enc.SetEscapeHTML(false)
  515. if err := enc.Encode(request); err != nil {
  516. return fmt.Errorf("failed to marshal data: %v", err)
  517. }
  518. endpoint := fmt.Sprintf("http://127.0.0.1:%d/completion", s.port)
  519. req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, buffer)
  520. if err != nil {
  521. return fmt.Errorf("error creating POST request: %v", err)
  522. }
  523. req.Header.Set("Content-Type", "application/json")
  524. resp, err := http.DefaultClient.Do(req)
  525. if err != nil {
  526. return fmt.Errorf("POST predict: %v", err)
  527. }
  528. defer resp.Body.Close()
  529. if resp.StatusCode >= 400 {
  530. bodyBytes, err := io.ReadAll(resp.Body)
  531. if err != nil {
  532. return fmt.Errorf("failed reading llm error response: %w", err)
  533. }
  534. log.Printf("llm predict error: %s", bodyBytes)
  535. return fmt.Errorf("%s", bodyBytes)
  536. }
  537. scanner := bufio.NewScanner(resp.Body)
  538. buf := make([]byte, 0, maxBufferSize)
  539. scanner.Buffer(buf, maxBufferSize)
  540. retryNeeded := false
  541. // keep track of the last token generated, this is used to abort if the model starts looping
  542. var lastToken string
  543. var tokenRepeat int
  544. for scanner.Scan() {
  545. select {
  546. case <-ctx.Done():
  547. // This handles the request cancellation
  548. return ctx.Err()
  549. default:
  550. line := scanner.Bytes()
  551. if len(line) == 0 {
  552. continue
  553. }
  554. // try again on slot unavailable
  555. if bytes.Contains(line, []byte("slot unavailable")) {
  556. retryNeeded = true
  557. break
  558. }
  559. evt, ok := bytes.CutPrefix(line, []byte("data: "))
  560. if !ok {
  561. return fmt.Errorf("error parsing llm response stream: %s", line)
  562. }
  563. var c completion
  564. if err := json.Unmarshal(evt, &c); err != nil {
  565. return fmt.Errorf("error unmarshaling llm prediction response: %v", err)
  566. }
  567. switch {
  568. case strings.TrimSpace(c.Content) == lastToken:
  569. tokenRepeat++
  570. default:
  571. lastToken = strings.TrimSpace(c.Content)
  572. tokenRepeat = 0
  573. }
  574. // 30 picked as an arbitrary max token repeat limit, modify as needed
  575. if tokenRepeat > 30 {
  576. slog.Debug("prediction aborted, token repeat limit reached")
  577. return ctx.Err()
  578. }
  579. if c.Content != "" {
  580. fn(CompletionResponse{
  581. Content: c.Content,
  582. })
  583. }
  584. if c.Stop {
  585. fn(CompletionResponse{
  586. Done: true,
  587. PromptEvalCount: c.Timings.PromptN,
  588. PromptEvalDuration: parseDurationMs(c.Timings.PromptMS),
  589. EvalCount: c.Timings.PredictedN,
  590. EvalDuration: parseDurationMs(c.Timings.PredictedMS),
  591. })
  592. return nil
  593. }
  594. }
  595. }
  596. if err := scanner.Err(); err != nil {
  597. if strings.Contains(err.Error(), "unexpected EOF") {
  598. s.Close()
  599. msg := ""
  600. if s.status != nil && s.status.LastErrMsg != "" {
  601. msg = s.status.LastErrMsg
  602. }
  603. return fmt.Errorf("an unknown error was encountered while running the model %s", msg)
  604. }
  605. return fmt.Errorf("error reading llm response: %v", err)
  606. }
  607. if !retryNeeded {
  608. return nil // success
  609. }
  610. }
  611. // should never reach here ideally
  612. return fmt.Errorf("max retries exceeded")
  613. }
  614. type EmbeddingRequest struct {
  615. Content string `json:"content"`
  616. }
  617. type EmbeddingResponse struct {
  618. Embedding []float64 `json:"embedding"`
  619. }
  620. func (s *llmServer) Embedding(ctx context.Context, prompt string) ([]float64, error) {
  621. if err := s.sem.Acquire(ctx, 1); err != nil {
  622. slog.Error("Failed to acquire semaphore", "error", err)
  623. return nil, err
  624. }
  625. defer s.sem.Release(1)
  626. // Make sure the server is ready
  627. status, err := s.getServerStatus(ctx)
  628. if err != nil {
  629. return nil, err
  630. } else if status != ServerStatusReady {
  631. return nil, fmt.Errorf("unexpected server status: %s", status.ToString())
  632. }
  633. data, err := json.Marshal(TokenizeRequest{Content: prompt})
  634. if err != nil {
  635. return nil, fmt.Errorf("error marshaling embed data: %w", err)
  636. }
  637. req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("http://127.0.0.1:%d/embedding", s.port), bytes.NewBuffer(data))
  638. if err != nil {
  639. return nil, fmt.Errorf("error creating embed request: %w", err)
  640. }
  641. req.Header.Set("Content-Type", "application/json")
  642. resp, err := http.DefaultClient.Do(req)
  643. if err != nil {
  644. return nil, fmt.Errorf("do embedding request: %w", err)
  645. }
  646. defer resp.Body.Close()
  647. body, err := io.ReadAll(resp.Body)
  648. if err != nil {
  649. return nil, fmt.Errorf("error reading embed response: %w", err)
  650. }
  651. if resp.StatusCode >= 400 {
  652. log.Printf("llm encode error: %s", body)
  653. return nil, fmt.Errorf("%s", body)
  654. }
  655. var embedding EmbeddingResponse
  656. if err := json.Unmarshal(body, &embedding); err != nil {
  657. return nil, fmt.Errorf("unmarshal tokenize response: %w", err)
  658. }
  659. return embedding.Embedding, nil
  660. }
  661. type TokenizeRequest struct {
  662. Content string `json:"content"`
  663. }
  664. type TokenizeResponse struct {
  665. Tokens []int `json:"tokens"`
  666. }
  667. func (s *llmServer) Tokenize(ctx context.Context, content string) ([]int, error) {
  668. // Make sure the server is ready
  669. status, err := s.getServerStatus(ctx)
  670. if err != nil {
  671. return nil, err
  672. } else if status != ServerStatusReady && status != ServerStatusNoSlotsAvaialble {
  673. return nil, fmt.Errorf("unexpected server status: %s", status.ToString())
  674. }
  675. data, err := json.Marshal(TokenizeRequest{Content: content})
  676. if err != nil {
  677. return nil, fmt.Errorf("marshaling encode data: %w", err)
  678. }
  679. req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("http://127.0.0.1:%d/tokenize", s.port), bytes.NewBuffer(data))
  680. if err != nil {
  681. return nil, fmt.Errorf("encode request: %w", err)
  682. }
  683. req.Header.Set("Content-Type", "application/json")
  684. resp, err := http.DefaultClient.Do(req)
  685. if err != nil {
  686. return nil, fmt.Errorf("do encode request: %w", err)
  687. }
  688. defer resp.Body.Close()
  689. body, err := io.ReadAll(resp.Body)
  690. if err != nil {
  691. return nil, fmt.Errorf("read encode request: %w", err)
  692. }
  693. if resp.StatusCode >= 400 {
  694. log.Printf("llm encode error: %s", body)
  695. return nil, fmt.Errorf("%s", body)
  696. }
  697. var encoded TokenizeResponse
  698. if err := json.Unmarshal(body, &encoded); err != nil {
  699. return nil, fmt.Errorf("unmarshal encode response: %w", err)
  700. }
  701. return encoded.Tokens, nil
  702. }
  703. type DetokenizeRequest struct {
  704. Tokens []int `json:"tokens"`
  705. }
  706. type DetokenizeResponse struct {
  707. Content string `json:"content"`
  708. }
  709. func (s *llmServer) Detokenize(ctx context.Context, tokens []int) (string, error) {
  710. // Make sure the server is ready
  711. status, err := s.getServerStatus(ctx)
  712. if err != nil {
  713. return "", err
  714. } else if status != ServerStatusReady && status != ServerStatusNoSlotsAvaialble {
  715. return "", fmt.Errorf("unexpected server status: %s", status.ToString())
  716. }
  717. data, err := json.Marshal(DetokenizeRequest{Tokens: tokens})
  718. if err != nil {
  719. return "", fmt.Errorf("marshaling decode data: %w", err)
  720. }
  721. req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("http://127.0.0.1:%d/detokenize", s.port), bytes.NewBuffer(data))
  722. if err != nil {
  723. return "", fmt.Errorf("decode request: %w", err)
  724. }
  725. req.Header.Set("Content-Type", "application/json")
  726. resp, err := http.DefaultClient.Do(req)
  727. if err != nil {
  728. return "", fmt.Errorf("do decode request: %w", err)
  729. }
  730. defer resp.Body.Close()
  731. body, err := io.ReadAll(resp.Body)
  732. if err != nil {
  733. return "", fmt.Errorf("read decode request: %w", err)
  734. }
  735. if resp.StatusCode >= 400 {
  736. log.Printf("llm decode error: %s", body)
  737. return "", fmt.Errorf("%s", body)
  738. }
  739. var decoded DetokenizeResponse
  740. if err := json.Unmarshal(body, &decoded); err != nil {
  741. return "", fmt.Errorf("unmarshal encode response: %w", err)
  742. }
  743. return decoded.Content, nil
  744. }
  745. func (s *llmServer) Close() error {
  746. if s.cmd != nil {
  747. slog.Debug("stopping llama server")
  748. return s.cmd.Process.Kill()
  749. }
  750. return nil
  751. }
  752. func (s *llmServer) EstimatedVRAM() uint64 {
  753. return s.estimatedVRAM
  754. }
  755. func parseDurationMs(ms float64) time.Duration {
  756. dur, err := time.ParseDuration(fmt.Sprintf("%fms", ms))
  757. if err != nil {
  758. panic(err)
  759. }
  760. return dur
  761. }