runner.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932
  1. package main
  2. import (
  3. "context"
  4. "encoding/json"
  5. "errors"
  6. "flag"
  7. "fmt"
  8. "log"
  9. "log/slog"
  10. "net"
  11. "net/http"
  12. "os"
  13. "path/filepath"
  14. "regexp"
  15. "runtime"
  16. "strconv"
  17. "strings"
  18. "sync"
  19. "time"
  20. "unicode/utf8"
  21. "github.com/ollama/ollama/api"
  22. "github.com/ollama/ollama/llama"
  23. )
  24. // input is an element of the prompt to process, either
  25. // a token or an image embedding (generated from a vision projector)
  26. type input struct {
  27. token int
  28. // embed is an image embedding
  29. embed []float32
  30. }
  31. type Sequence struct {
  32. // number of inputs evaluated
  33. numPast int
  34. // batch index
  35. iBatch int
  36. // number of tokens predicted so far
  37. numPredicted int
  38. // prompt inputs left to evaluate
  39. inputs []input
  40. // tokens that have been generated but not returned yet (e.g. for stop sequences)
  41. pendingResponses []string
  42. // input cache being used by this sequence
  43. cache *InputCacheSlot
  44. // channel to send responses over
  45. responses chan string
  46. // channel to stop decoding (such as if the remote connection is closed)
  47. quit chan bool
  48. // number of tokens to predict
  49. numPredict int
  50. samplingCtx *llama.SamplingContext
  51. // channel to send back the embedding if embedding only
  52. embedding chan []float32
  53. // stop sequences
  54. stop []string
  55. // number of inputs to keep at the beginning when shifting context window
  56. numKeep int
  57. // true if an embedding are to be returned instead of text generation
  58. embeddingOnly bool
  59. doneReason string
  60. // Metrics
  61. startProcessingTime time.Time
  62. startGenerationTime time.Time
  63. numDecoded int
  64. numPromptInputs int
  65. }
  66. type NewSequenceParams struct {
  67. numPredict int
  68. stop []string
  69. numKeep int
  70. samplingParams *llama.SamplingParams
  71. embedding bool
  72. }
  73. func (s *Server) NewSequence(prompt string, images []ImageData, params NewSequenceParams) (*Sequence, error) {
  74. s.ready.Wait()
  75. startTime := time.Now()
  76. inputs, err := s.inputs(prompt, images)
  77. if err != nil {
  78. return nil, fmt.Errorf("failed to process inputs: %w", err)
  79. } else if len(inputs) == 0 {
  80. return nil, errors.New("no input provided")
  81. }
  82. if params.numKeep < 0 {
  83. params.numKeep = len(inputs)
  84. }
  85. if !params.embedding {
  86. // Subtracting 4 ensures that at least 1 input can be discarded during shift
  87. params.numKeep = min(params.numKeep, s.cache.numCtx-4)
  88. params.numKeep += s.bosToken
  89. } else {
  90. // Embeddings are 1 shot - just truncate to the context window, without ever shifting
  91. params.numKeep = min(params.numKeep, s.cache.numCtx)
  92. }
  93. // truncate to fit in context window
  94. if len(inputs) > s.cache.numCtx {
  95. slog.Warn("truncating input prompt", "limit", s.cache.numCtx, "prompt", len(inputs), "numKeep", params.numKeep)
  96. newInputs := inputs[:params.numKeep]
  97. newInputs = append(newInputs, inputs[len(inputs)-s.cache.numCtx+params.numKeep:]...)
  98. inputs = newInputs
  99. }
  100. var sc *llama.SamplingContext
  101. if params.samplingParams != nil {
  102. sc = llama.NewSamplingContext(s.model, *params.samplingParams)
  103. for _, input := range inputs {
  104. if input.embed == nil {
  105. sc.Accept(input.token, false)
  106. }
  107. }
  108. }
  109. return &Sequence{
  110. inputs: inputs,
  111. numPromptInputs: len(inputs),
  112. startProcessingTime: startTime,
  113. numPredict: params.numPredict,
  114. pendingResponses: make([]string, 0),
  115. responses: make(chan string, 100),
  116. quit: make(chan bool, 1),
  117. embedding: make(chan []float32, 1),
  118. samplingCtx: sc,
  119. embeddingOnly: params.embedding,
  120. stop: params.stop,
  121. numKeep: params.numKeep,
  122. }, nil
  123. }
  124. // inputs processes the prompt and images into a list of inputs
  125. // by splitting the prompt on [img-<n>] tags, tokenizing text and
  126. // generating image embeddings for each image
  127. func (s *Server) inputs(prompt string, images []ImageData) ([]input, error) {
  128. var inputs []input
  129. re := regexp.MustCompile(`\[img-(\d+)\]`)
  130. parts := re.Split(prompt, -1)
  131. matches := re.FindAllStringSubmatch(prompt, -1)
  132. for i, part := range parts {
  133. // text - tokenize
  134. if strings.TrimSpace(part) != "" {
  135. tokens, err := s.lc.Model().Tokenize(part, i == 0, true)
  136. if err != nil {
  137. return nil, err
  138. }
  139. for _, t := range tokens {
  140. inputs = append(inputs, input{token: t})
  141. }
  142. }
  143. // image - generate image embedding
  144. if i < len(matches) {
  145. n, _ := strconv.Atoi(matches[i][1])
  146. imageIndex := -1
  147. for j := range images {
  148. if images[j].ID == n {
  149. imageIndex = j
  150. break
  151. }
  152. }
  153. if imageIndex < 0 {
  154. return nil, fmt.Errorf("invalid image index: %d", n)
  155. }
  156. hash := s.cache.HashImage(images[imageIndex].Data)
  157. // Vision models cannot be accessed concurrently
  158. s.clip.mu.Lock()
  159. embed, err := s.cache.FindImage(hash)
  160. if err != nil {
  161. embed = llama.NewLlavaImageEmbed(s.lc, s.clip.cc, images[imageIndex].Data)
  162. s.cache.AddImage(hash, embed)
  163. }
  164. s.clip.mu.Unlock()
  165. for _, e := range embed {
  166. inputs = append(inputs, input{embed: e})
  167. }
  168. }
  169. }
  170. if s.clip.cc != nil {
  171. var embed [][]float32
  172. if s.clip.cc.IsMllama && len(images) >= 1 {
  173. hash := s.cache.HashImage(images[0].Data)
  174. s.clip.mu.Lock()
  175. var err error
  176. embed, err = s.cache.FindImage(hash)
  177. if err != nil {
  178. embed = llama.NewMllamaImageEmbed(s.lc, s.clip.cc, images[0].Data, images[0].AspectRatioID)
  179. s.cache.AddImage(hash, embed)
  180. }
  181. s.clip.mu.Unlock()
  182. }
  183. s.mu.Lock()
  184. llama.MllamaSetCrossAttn(s.lc, s.clip.cc, embed)
  185. s.mu.Unlock()
  186. }
  187. return inputs, nil
  188. }
  189. type clip struct {
  190. cc *llama.ClipContext
  191. mu sync.Mutex
  192. }
  193. type Server struct {
  194. model *llama.Model
  195. lc *llama.Context
  196. // required for image embeddings
  197. clip clip
  198. batchSize int
  199. // parallel is the number of parallel requests to handle
  200. parallel int
  201. // seqs is the list of parallel sequences being evaluated
  202. // TODO (jmorganca): this can probably be moved into run()
  203. seqs []*Sequence
  204. // KV cache
  205. cache *InputCache
  206. // does this model require a beginning of sequence token?
  207. bosToken int
  208. // next sequence for prompt processing to avoid starvation
  209. nextSeq int
  210. // is the server ready to process requests?
  211. ready sync.WaitGroup
  212. mu sync.Mutex
  213. cond *sync.Cond
  214. progress float32
  215. status ServerStatus
  216. }
  217. func (s *Server) allNil() bool {
  218. for _, item := range s.seqs {
  219. if item != nil {
  220. return false
  221. }
  222. }
  223. return true
  224. }
  225. func (s *Server) shiftContext(seq *Sequence) {
  226. numLeft := seq.numPast - seq.numKeep
  227. numDiscard := numLeft / 2
  228. slog.Debug("context limit hit - shifting", "limit", s.cache.numCtx, "numPast", seq.numPast,
  229. "numKeep", seq.numKeep, "numLeft", numLeft, "numDiscard", numDiscard)
  230. s.cache.ShiftCacheSlot(seq.cache, seq.numKeep, numDiscard, seq.numPast)
  231. seq.numPast -= numDiscard
  232. }
  233. func flushPending(seq *Sequence) bool {
  234. joined := strings.Join(seq.pendingResponses, "")
  235. seq.pendingResponses = []string{}
  236. // Check if there are any partial UTF-8 characters remaining.
  237. // We already check and queue as we are generating but some may
  238. // still make it here:
  239. // - Sequence is ending, e.g. generation limit has been hit
  240. // - Invalid characters in the middle of a string
  241. // This is a stricter check to ensure we never output invalid Unicode.
  242. for !utf8.ValidString(joined) {
  243. joined = joined[:len(joined)-1]
  244. }
  245. if len(joined) == 0 {
  246. return true
  247. }
  248. select {
  249. case seq.responses <- joined:
  250. return true
  251. case <-seq.quit:
  252. return false
  253. }
  254. }
  255. func (s *Server) removeSequence(seqIndex int, reason string) {
  256. seq := s.seqs[seqIndex]
  257. flushPending(seq)
  258. seq.doneReason = reason
  259. close(seq.responses)
  260. close(seq.embedding)
  261. seq.cache.InUse = false
  262. if s.clip.cc != nil {
  263. llama.MllamaSetCrossAttn(s.lc, s.clip.cc, nil)
  264. }
  265. s.seqs[seqIndex] = nil
  266. }
  267. func (s *Server) run(ctx context.Context) {
  268. s.ready.Wait()
  269. // logically these batches are used only within the context of processBatch
  270. // but it is better for performance to allocate them once here
  271. tokenBatch := llama.NewBatch(s.batchSize*len(s.seqs), 0, len(s.seqs))
  272. defer tokenBatch.Free()
  273. embedBatch := llama.NewBatch(s.batchSize*len(s.seqs), s.lc.Model().NEmbd(), len(s.seqs))
  274. defer embedBatch.Free()
  275. for {
  276. select {
  277. case <-ctx.Done():
  278. return
  279. default:
  280. s.processBatch(tokenBatch, embedBatch)
  281. tokenBatch.Clear()
  282. embedBatch.Clear()
  283. }
  284. }
  285. }
  286. // TODO (jmorganca): processBatch should be simplified, removing:
  287. // * sampling
  288. // * stop token checking
  289. // * metrics
  290. // these should instead be handled by the handlers
  291. // it should only be responsible for accepting tokens or embeddings and
  292. // processing batches as fast as possible
  293. func (s *Server) processBatch(tokenBatch *llama.Batch, embedBatch *llama.Batch) {
  294. s.mu.Lock()
  295. for s.allNil() {
  296. s.cond.Wait() // Wait until an item is added
  297. }
  298. defer s.mu.Unlock()
  299. var batch *llama.Batch
  300. seqIdx := s.nextSeq - 1
  301. for range s.seqs {
  302. seqIdx = (seqIdx + 1) % len(s.seqs)
  303. seq := s.seqs[seqIdx]
  304. if seq == nil {
  305. continue
  306. }
  307. // if past the num predict limit
  308. if seq.numPredict > 0 && seq.numPredicted > seq.numPredict {
  309. s.removeSequence(seqIdx, "limit")
  310. continue
  311. }
  312. if seq.numPast+len(seq.inputs) > s.cache.numCtx {
  313. s.shiftContext(seq)
  314. }
  315. var numInputsProcessed int
  316. for i, input := range seq.inputs {
  317. embedding := input.embed != nil
  318. // If we don't currently have a batch, use one of the correct type and
  319. // fill it up as much as possible across all sequences. If we encounter an
  320. // input of the opppsite type, stop for that sequence but then pick up from
  321. // there for the next batch, ensuring that we alternate types
  322. if batch == nil {
  323. if !embedding {
  324. batch = tokenBatch
  325. } else {
  326. batch = embedBatch
  327. }
  328. } else if embedding != batch.IsEmbedding() {
  329. s.nextSeq = seqIdx
  330. break
  331. }
  332. // todo: make this n_batch
  333. if i >= s.batchSize {
  334. break
  335. }
  336. batch.Add(input.token, input.embed, seq.numPast, []int{seq.cache.Id}, numInputsProcessed+1 == len(seq.inputs))
  337. seq.numPast++
  338. numInputsProcessed++
  339. }
  340. if numInputsProcessed > 0 {
  341. seq.cache.Inputs = append(seq.cache.Inputs, seq.inputs[:numInputsProcessed]...)
  342. seq.inputs = seq.inputs[numInputsProcessed:]
  343. seq.iBatch = batch.NumTokens() - 1
  344. }
  345. }
  346. if batch == nil || batch.NumTokens() == 0 {
  347. return
  348. }
  349. err := s.lc.Decode(batch)
  350. if err != nil {
  351. slog.Error("failed to decode batch", "error", err)
  352. return
  353. }
  354. for i, seq := range s.seqs {
  355. if seq == nil {
  356. continue
  357. }
  358. // don't sample prompt processing
  359. if len(seq.inputs) != 0 {
  360. continue
  361. }
  362. seq.numDecoded += 1
  363. if seq.numDecoded == 1 {
  364. seq.startGenerationTime = time.Now()
  365. }
  366. // if done processing the prompt, generate an embedding and return
  367. if seq.embeddingOnly {
  368. embed := s.lc.GetEmbeddingsSeq(i)
  369. if embed == nil {
  370. embed = s.lc.GetEmbeddingsIth(seq.iBatch)
  371. }
  372. seq.embedding <- embed
  373. s.removeSequence(i, "")
  374. continue
  375. }
  376. // sample a token
  377. token := seq.samplingCtx.Sample(s.lc, seq.iBatch)
  378. seq.samplingCtx.Accept(token, true)
  379. piece := s.model.TokenToPiece(token)
  380. seq.numPredicted++
  381. // if it's an end of sequence token, break
  382. if s.model.TokenIsEog(token) {
  383. // TODO (jmorganca): we should send this back
  384. // as it's important for the /api/generate context
  385. // seq.responses <- piece
  386. s.removeSequence(i, "stop")
  387. continue
  388. }
  389. seq.inputs = []input{{token: token}}
  390. seq.pendingResponses = append(seq.pendingResponses, piece)
  391. sequence := strings.Join(seq.pendingResponses, "")
  392. if ok, stop := findStop(sequence, seq.stop); ok {
  393. slog.Debug("hit stop token", "pending", seq.pendingResponses, "stop", stop)
  394. var tokenTruncated bool
  395. origLen := len(seq.pendingResponses)
  396. seq.pendingResponses, tokenTruncated = truncateStop(seq.pendingResponses, stop)
  397. newLen := len(seq.pendingResponses)
  398. // Update the cache based on the tokens that will be returned:
  399. // - We have 1 token more than is currently in the cache because
  400. // the last one generated wasn't submitted to Decode
  401. // - Remove any stop sequences that we stripped out
  402. // - If truncateStop removed a portion of a token, drop that
  403. // - As defense-in-depth, if truncatedToken didn't find a stop token
  404. // remove the extra one that we added to the cache len
  405. tokenLen := len(seq.cache.Inputs) + 1
  406. tokenLen -= origLen - newLen
  407. if tokenTruncated || origLen == newLen {
  408. tokenLen--
  409. }
  410. seq.cache.Inputs = seq.cache.Inputs[:tokenLen]
  411. s.removeSequence(i, "stop")
  412. continue
  413. }
  414. if containsStopSuffix(sequence, seq.stop) {
  415. continue
  416. }
  417. if incompleteUnicode(sequence) {
  418. continue
  419. }
  420. if !flushPending(seq) {
  421. s.removeSequence(i, "connection")
  422. }
  423. }
  424. }
  425. // TODO (jmorganca): use structs from the api package to avoid duplication
  426. // this way the api acts as a proxy instead of using a different api for the
  427. // runner
  428. type Options struct {
  429. api.Runner
  430. NumKeep int `json:"n_keep"`
  431. Seed int `json:"seed"`
  432. NumPredict int `json:"n_predict"`
  433. TopK int `json:"top_k"`
  434. TopP float32 `json:"top_p"`
  435. MinP float32 `json:"min_p"`
  436. TFSZ float32 `json:"tfs_z"`
  437. TypicalP float32 `json:"typical_p"`
  438. RepeatLastN int `json:"repeat_last_n"`
  439. Temperature float32 `json:"temperature"`
  440. RepeatPenalty float32 `json:"repeat_penalty"`
  441. PresencePenalty float32 `json:"presence_penalty"`
  442. FrequencyPenalty float32 `json:"frequency_penalty"`
  443. Mirostat int `json:"mirostat"`
  444. MirostatTau float32 `json:"mirostat_tau"`
  445. MirostatEta float32 `json:"mirostat_eta"`
  446. PenalizeNewline bool `json:"penalize_nl"`
  447. Stop []string `json:"stop"`
  448. }
  449. type ImageData struct {
  450. Data []byte `json:"data"`
  451. ID int `json:"id"`
  452. AspectRatioID int `json:"aspect_ratio_id"`
  453. }
  454. type CompletionRequest struct {
  455. Prompt string `json:"prompt"`
  456. Images []ImageData `json:"image_data"`
  457. Grammar string `json:"grammar"`
  458. CachePrompt bool `json:"cache_prompt"`
  459. Options
  460. }
  461. type Timings struct {
  462. PredictedN int `json:"predicted_n"`
  463. PredictedMS float64 `json:"predicted_ms"`
  464. PromptN int `json:"prompt_n"`
  465. PromptMS float64 `json:"prompt_ms"`
  466. }
  467. type CompletionResponse struct {
  468. Content string `json:"content"`
  469. Stop bool `json:"stop"`
  470. Model string `json:"model,omitempty"`
  471. Prompt string `json:"prompt,omitempty"`
  472. StoppedLimit bool `json:"stopped_limit,omitempty"`
  473. PredictedN int `json:"predicted_n,omitempty"`
  474. PredictedMS float64 `json:"predicted_ms,omitempty"`
  475. PromptN int `json:"prompt_n,omitempty"`
  476. PromptMS float64 `json:"prompt_ms,omitempty"`
  477. Timings Timings `json:"timings"`
  478. }
  479. func (s *Server) completion(w http.ResponseWriter, r *http.Request) {
  480. var req CompletionRequest
  481. req.Options = Options(api.DefaultOptions())
  482. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  483. http.Error(w, "Bad request", http.StatusBadRequest)
  484. return
  485. }
  486. // Set the headers to indicate streaming
  487. w.Header().Set("Content-Type", "application/json")
  488. w.Header().Set("Transfer-Encoding", "chunked")
  489. flusher, ok := w.(http.Flusher)
  490. if !ok {
  491. http.Error(w, "Streaming not supported", http.StatusInternalServerError)
  492. return
  493. }
  494. var samplingParams llama.SamplingParams
  495. samplingParams.TopK = req.TopK
  496. samplingParams.TopP = req.TopP
  497. samplingParams.MinP = req.MinP
  498. samplingParams.TfsZ = req.TFSZ
  499. samplingParams.TypicalP = req.TypicalP
  500. samplingParams.Temp = req.Temperature
  501. samplingParams.RepeatLastN = req.RepeatLastN
  502. samplingParams.PenaltyRepeat = req.RepeatPenalty
  503. samplingParams.PenaltyFreq = req.FrequencyPenalty
  504. samplingParams.PenaltyPresent = req.PresencePenalty
  505. samplingParams.Mirostat = req.Mirostat
  506. samplingParams.MirostatTau = req.MirostatTau
  507. samplingParams.MirostatEta = req.MirostatEta
  508. samplingParams.PenalizeNl = req.PenalizeNewline
  509. samplingParams.Seed = uint32(req.Seed)
  510. samplingParams.Grammar = req.Grammar
  511. seq, err := s.NewSequence(req.Prompt, req.Images, NewSequenceParams{
  512. numPredict: req.NumPredict,
  513. stop: req.Stop,
  514. numKeep: req.NumKeep,
  515. samplingParams: &samplingParams,
  516. embedding: false,
  517. })
  518. if err != nil {
  519. http.Error(w, fmt.Sprintf("Failed to create new sequence: %v", err), http.StatusInternalServerError)
  520. return
  521. }
  522. // TODO (jmorganca): add to sequence queue instead of
  523. // failing if a slot isn't available
  524. s.mu.Lock()
  525. for i, sq := range s.seqs {
  526. if sq == nil {
  527. seq.cache, seq.inputs, seq.numPast, err = s.cache.LoadCacheSlot(seq.inputs, req.CachePrompt)
  528. if err != nil {
  529. s.mu.Unlock()
  530. http.Error(w, fmt.Sprintf("Failed to load cache: %v", err), http.StatusInternalServerError)
  531. return
  532. }
  533. s.seqs[i] = seq
  534. s.cond.Signal()
  535. break
  536. }
  537. }
  538. s.mu.Unlock()
  539. for {
  540. select {
  541. case <-r.Context().Done():
  542. close(seq.quit)
  543. return
  544. case content, ok := <-seq.responses:
  545. if ok {
  546. if err := json.NewEncoder(w).Encode(&CompletionResponse{
  547. Content: content,
  548. }); err != nil {
  549. http.Error(w, fmt.Sprintf("failed to encode response: %v", err), http.StatusInternalServerError)
  550. close(seq.quit)
  551. return
  552. }
  553. flusher.Flush()
  554. } else {
  555. // Send the final response
  556. if err := json.NewEncoder(w).Encode(&CompletionResponse{
  557. Stop: true,
  558. StoppedLimit: seq.doneReason == "limit",
  559. Timings: Timings{
  560. PromptN: seq.numPromptInputs,
  561. PromptMS: float64(seq.startGenerationTime.Sub(seq.startProcessingTime).Milliseconds()),
  562. PredictedN: seq.numDecoded,
  563. PredictedMS: float64(time.Since(seq.startGenerationTime).Milliseconds()),
  564. },
  565. }); err != nil {
  566. http.Error(w, fmt.Sprintf("failed to encode final response: %v", err), http.StatusInternalServerError)
  567. }
  568. return
  569. }
  570. }
  571. }
  572. }
  573. type EmbeddingRequest struct {
  574. Content string `json:"content"`
  575. CachePrompt bool `json:"cache_prompt"`
  576. }
  577. type EmbeddingResponse struct {
  578. Embedding []float32 `json:"embedding"`
  579. }
  580. func (s *Server) embeddings(w http.ResponseWriter, r *http.Request) {
  581. var req EmbeddingRequest
  582. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  583. http.Error(w, fmt.Sprintf("bad request: %s", err), http.StatusBadRequest)
  584. return
  585. }
  586. w.Header().Set("Content-Type", "application/json")
  587. slog.Debug("embedding request", "content", req.Content)
  588. seq, err := s.NewSequence(req.Content, nil, NewSequenceParams{embedding: true})
  589. if err != nil {
  590. http.Error(w, fmt.Sprintf("Failed to create new sequence: %v", err), http.StatusInternalServerError)
  591. return
  592. }
  593. // TODO (jessegross): Wait for a free slot instead of failing and blocking forever
  594. s.mu.Lock()
  595. for i, sq := range s.seqs {
  596. if sq == nil {
  597. seq.cache, seq.inputs, seq.numPast, err = s.cache.LoadCacheSlot(seq.inputs, req.CachePrompt)
  598. if err != nil {
  599. s.mu.Unlock()
  600. http.Error(w, fmt.Sprintf("Failed to load cache: %v", err), http.StatusInternalServerError)
  601. return
  602. }
  603. s.seqs[i] = seq
  604. s.cond.Signal()
  605. break
  606. }
  607. }
  608. s.mu.Unlock()
  609. embedding := <-seq.embedding
  610. if err := json.NewEncoder(w).Encode(&EmbeddingResponse{
  611. Embedding: embedding,
  612. }); err != nil {
  613. http.Error(w, fmt.Sprintf("failed to encode response: %v", err), http.StatusInternalServerError)
  614. }
  615. }
  616. type HealthResponse struct {
  617. Status string `json:"status"`
  618. Progress float32 `json:"progress"`
  619. }
  620. type ServerStatus int
  621. const (
  622. ServerStatusReady ServerStatus = iota
  623. ServerStatusLoadingModel
  624. ServerStatusError
  625. )
  626. func (s ServerStatus) ToString() string {
  627. switch s {
  628. case ServerStatusReady:
  629. return "ok"
  630. case ServerStatusLoadingModel:
  631. return "loading model"
  632. default:
  633. return "server error"
  634. }
  635. }
  636. func (s *Server) health(w http.ResponseWriter, r *http.Request) {
  637. w.Header().Set("Content-Type", "application/json")
  638. if err := json.NewEncoder(w).Encode(&HealthResponse{
  639. Status: s.status.ToString(),
  640. Progress: s.progress,
  641. }); err != nil {
  642. http.Error(w, fmt.Sprintf("failed to encode response: %v", err), http.StatusInternalServerError)
  643. }
  644. }
  645. func (s *Server) loadModel(
  646. params llama.ModelParams,
  647. mpath string,
  648. lpath string,
  649. ppath string,
  650. kvSize int,
  651. flashAttention bool,
  652. threads int,
  653. multiUserCache bool,
  654. ) {
  655. llama.BackendInit()
  656. s.model = llama.LoadModelFromFile(mpath, params)
  657. ctxParams := llama.NewContextParams(kvSize, s.batchSize*s.parallel, s.parallel, threads, flashAttention)
  658. s.lc = llama.NewContextWithModel(s.model, ctxParams)
  659. if lpath != "" {
  660. err := s.model.ApplyLoraFromFile(s.lc, lpath, 1.0, threads)
  661. if err != nil {
  662. panic(err)
  663. }
  664. }
  665. if s.model.AddBOSToken() {
  666. s.bosToken = 1
  667. }
  668. if ppath != "" {
  669. var err error
  670. s.clip.cc, err = llama.NewClipContext(ppath)
  671. if err != nil {
  672. panic(err)
  673. }
  674. }
  675. s.cache = NewInputCache(s.lc, kvSize, s.parallel, multiUserCache)
  676. s.status = ServerStatusReady
  677. s.ready.Done()
  678. }
  679. func main() {
  680. mpath := flag.String("model", "", "Path to model binary file")
  681. ppath := flag.String("mmproj", "", "Path to projector binary file")
  682. parallel := flag.Int("parallel", 1, "Number of sequences to handle simultaneously")
  683. batchSize := flag.Int("batch-size", 512, "Batch size")
  684. nGpuLayers := flag.Int("n-gpu-layers", 0, "Number of layers to offload to GPU")
  685. mainGpu := flag.Int("main-gpu", 0, "Main GPU")
  686. flashAttention := flag.Bool("flash-attn", false, "Enable flash attention")
  687. kvSize := flag.Int("ctx-size", 2048, "Context (or KV cache) size")
  688. lpath := flag.String("lora", "", "Path to lora layer file")
  689. port := flag.Int("port", 8080, "Port to expose the server on")
  690. threads := flag.Int("threads", runtime.NumCPU(), "Number of threads to use during generation")
  691. verbose := flag.Bool("verbose", false, "verbose output (default: disabled)")
  692. noMmap := flag.Bool("no-mmap", false, "do not memory-map model (slower load but may reduce pageouts if not using mlock)")
  693. mlock := flag.Bool("mlock", false, "force system to keep model in RAM rather than swapping or compressing")
  694. tensorSplit := flag.String("tensor-split", "", "fraction of the model to offload to each GPU, comma-separated list of proportions")
  695. multiUserCache := flag.Bool("multiuser-cache", false, "optimize input cache algorithm for multiple users")
  696. // Expose requirements as a JSON output to stdout
  697. requirements := flag.Bool("requirements", false, "print json requirement information")
  698. // These are either ignored by llama.cpp or have no significance to us
  699. _ = flag.Bool("embedding", false, "enable embedding vector output (default: disabled)")
  700. _ = flag.Bool("log-disable", false, "disables logging to a file")
  701. _ = flag.Bool("memory-f32", false, "use f32 instead of f16 for memory key+value (default: disabled) not recommended: doubles context memory required and no measurable increase in quality")
  702. flag.Parse()
  703. if *requirements {
  704. printRequirements(os.Stdout)
  705. return
  706. }
  707. level := slog.LevelInfo
  708. if *verbose {
  709. level = slog.LevelDebug
  710. }
  711. handler := slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
  712. Level: level,
  713. AddSource: true,
  714. ReplaceAttr: func(_ []string, attr slog.Attr) slog.Attr {
  715. if attr.Key == slog.SourceKey {
  716. source := attr.Value.Any().(*slog.Source)
  717. source.File = filepath.Base(source.File)
  718. }
  719. return attr
  720. },
  721. })
  722. slog.SetDefault(slog.New(handler))
  723. slog.Info("starting go runner")
  724. slog.Debug("system info", "cpu", llama.PrintSystemInfo(), "threads", *threads)
  725. server := &Server{
  726. batchSize: *batchSize,
  727. parallel: *parallel,
  728. seqs: make([]*Sequence, *parallel),
  729. status: ServerStatusLoadingModel,
  730. }
  731. var tensorSplitFloats []float32
  732. if *tensorSplit != "" {
  733. stringFloats := regexp.MustCompile(",").Split(*tensorSplit, -1)
  734. tensorSplitFloats = make([]float32, 0, len(stringFloats))
  735. for _, s := range stringFloats {
  736. f, _ := strconv.ParseFloat(s, 32)
  737. tensorSplitFloats = append(tensorSplitFloats, float32(f))
  738. }
  739. }
  740. params := llama.ModelParams{
  741. NumGpuLayers: *nGpuLayers,
  742. MainGpu: *mainGpu,
  743. UseMmap: !*noMmap && *lpath == "",
  744. UseMlock: *mlock,
  745. TensorSplit: tensorSplitFloats,
  746. Progress: func(progress float32) {
  747. server.progress = progress
  748. },
  749. }
  750. server.ready.Add(1)
  751. go server.loadModel(params, *mpath, *lpath, *ppath, *kvSize, *flashAttention, *threads, *multiUserCache)
  752. server.cond = sync.NewCond(&server.mu)
  753. ctx, cancel := context.WithCancel(context.Background())
  754. go server.run(ctx)
  755. addr := "127.0.0.1:" + strconv.Itoa(*port)
  756. listener, err := net.Listen("tcp", addr)
  757. if err != nil {
  758. fmt.Println("Listen error:", err)
  759. return
  760. }
  761. defer listener.Close()
  762. mux := http.NewServeMux()
  763. mux.HandleFunc("/embedding", server.embeddings)
  764. mux.HandleFunc("/completion", server.completion)
  765. mux.HandleFunc("/health", server.health)
  766. httpServer := http.Server{
  767. Handler: mux,
  768. }
  769. log.Println("Server listening on", addr)
  770. if err := httpServer.Serve(listener); err != nil {
  771. log.Fatal("server error:", err)
  772. }
  773. cancel()
  774. }