server.go 24 KB

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