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