LLM 应用示例：最佳实践示例

LLM 应用开发模式：轻量级 API 编排

在 LangChain 中使用了思维链的方式来选择合适的智能体（Agent），在 Co-mate 中，我们也是采取了类似的设计，在本地构建好函数，然后交由 LLM 来分析用户的输入适合调用哪个函数。

如下是我们的 prompt 示例：

  1. Answer the following questions as best you can. You have access to the following tools:
  3. introduce_system: introduce_system is a function to introduce a system.
  5. Use the following format:
  7. Question: the input question you must answer
  8. Thought: you should always think about what to do
  9. Action: the action to take, should be one of [introduce_system]
  10. Action Input: the input to the action
  11. Observation: the result of the action
  12. ... (this Thought/Action/Action Input/Observation can repeat N times)
  13. Thought: I now know the final answer
  14. Final Answer: the final answer to the original input question
  16. Begin!
  18. Question: Introduce the following system: https://github.com/archguard/ddd-monolithic-code-sample

这里的 Question 便是用户的输入，然后再调用对应的 introduce_system 函数进行分析。

LLM 应用开发模式：DSL 动态运行时

与事实能力相比，我们更信任 LLM 的编排能力，因此我们在 Co-mate 中采用了 DSL 的方式来编排函数，这样可以更加灵活的编排函数。

为了支撑这样的能力，我们在 Co-mate 中引入了 Kotlin 作为 DSL 的运行时：

  1. // 初始化运行时
  2. val repl = KotlinInterpreter()
  3. val mvcDslSpec = repl.evalCast<FoundationSpec>(InterpreterRequest(code = mvcFoundation))
  5. // 从用户的输入中获取 action
  6. val action = ComateToolingAction.from(action.lowercase())
  8. // 添加默认的 DSL spec
  9. if (action == ComateToolingAction.FOUNDATION_SPEC_GOVERNANCE) {
  10.     comateContext.spec = mvcDslSpec
  11. }

对应的 DSL 示例（由 ChatGPT 根据 DDD 版本 spec 生成）：

  1. foundation {
  2.     project_name {
  3.         pattern("^([a-z0-9-]+)-([a-z0-9-]+)(-common)?\${'$'}")
  4.         example("system1-webapp1")
  5.     }
  7.     layered {
  8.         layer("controller") {
  9.             pattern(".*\\.controller") { name shouldBe endsWith("Controller") }
  10.         }
  11.         layer("service") {
  12.             pattern(".*\\.service") {
  13.                 name shouldBe endsWith("DTO", "Request", "Response", "Factory", "Service")
  14.             }
  15.         }
  16.         layer("repository") {
  17.             pattern(".*\\.repository") { name shouldBe endsWith("Entity", "Repository", "Mapper") }
  18.         }
  20.         dependency {
  21.             "controller" dependedOn "service"
  22.             "controller" dependedOn "repository"
  23.             "service" dependedOn "repository"
  24.         }
  25.     }
  27.     naming {
  28.         class_level {
  29.             style("CamelCase")
  30.             pattern(".*") { name shouldNotBe contains("${'$'}") }
  31.         }
  32.         function_level {
  33.             style("CamelCase")
  34.             pattern(".*") { name shouldNotBe contains("${'$'}") }
  35.         }
  36.     }
  37. }

LLM 应用开发模式：本地小模型

在 Co-mate 中，我们在本地引入了 SentenceTransformer 来处理用户的输入，优在本地分析、匹配用户的输入，并处理。当匹配到结果后直接调用本地的函数，当匹配不到结果时调用远端的处理函数来处理。

HuggingFace: https://huggingface.co/sentence-transformers

在原理上主要是参考了 GitHub Copilot、 Bloop 的实现，通过本地的小模型来处理用户的输入，然后再通过远端的大模型来处理用户的输入。

Rust 实现示例

Rust 相关示例：https://github.com/unit-mesh/unit-agent

  1. #![allow(unused)]
  2. fn main() {
  3. pub fn embed(&self, sequence: &str) -> anyhow::Result<Embedding> {
  4.     let tokenizer_output = self.tokenizer.encode(sequence, true).unwrap();
  6.     let input_ids = tokenizer_output.get_ids();
  7.     let attention_mask = tokenizer_output.get_attention_mask();
  8.     let token_type_ids = tokenizer_output.get_type_ids();
  9.     let length = input_ids.len();
  10.     trace!("embedding {} tokens {:?}", length, sequence);
  12.     let inputs_ids_array = ndarray::Array::from_shape_vec(
  13.         (1, length),
  14.         input_ids.iter().map(|&x| x as i64).collect(),
  15.     )?;
  17.     let attention_mask_array = ndarray::Array::from_shape_vec(
  18.         (1, length),
  19.         attention_mask.iter().map(|&x| x as i64).collect(),
  20.     )?;
  22.     let token_type_ids_array = ndarray::Array::from_shape_vec(
  23.         (1, length),
  24.         token_type_ids.iter().map(|&x| x as i64).collect(),
  25.     )?;
  27.     let outputs = self.session.run([
  28.         InputTensor::from_array(inputs_ids_array.into_dyn()),
  29.         InputTensor::from_array(attention_mask_array.into_dyn()),
  30.         InputTensor::from_array(token_type_ids_array.into_dyn()),
  31.     ])?;
  33.     let output_tensor: OrtOwnedTensor<f32, _> = outputs[0].try_extract().unwrap();
  34.     let sequence_embedding = &*output_tensor.view();
  35.     let pooled = sequence_embedding.mean_axis(Axis(1)).unwrap();
  37.     Ok(pooled.to_owned().as_slice().unwrap().to_vec())
  38. }
  39. }

Kotlin 实现示例

  1. class Semantic(val tokenizer: HuggingFaceTokenizer, val session: OrtSession, val env: OrtEnvironment) {
  2.     fun embed(
  3.         sequence: String,
  4.     ): FloatArray {
  5.         val tokenized = tokenizer.encode(sequence, true)
  7.         val inputIds = tokenized.ids
  8.         val attentionMask = tokenized.attentionMask
  9.         val typeIds = tokenized.typeIds
  11.         val tensorInput = OrtUtil.reshape(inputIds, longArrayOf(1, inputIds.size.toLong()))
  12.         val tensorAttentionMask = OrtUtil.reshape(attentionMask, longArrayOf(1, attentionMask.size.toLong()))
  13.         val tensorTypeIds = OrtUtil.reshape(typeIds, longArrayOf(1, typeIds.size.toLong()))
  15.         val result = session.run(
  16.             mapOf(
  17.                 "input_ids" to OnnxTensor.createTensor(env, tensorInput),
  18.                 "attention_mask" to OnnxTensor.createTensor(env, tensorAttentionMask),
  19.                 "token_type_ids" to OnnxTensor.createTensor(env, tensorTypeIds),
  20.             ),
  21.         )
  23.         val outputTensor = result.get(0) as OnnxTensor
  24.         val output = outputTensor.floatBuffer.array()
  26.         return output
  27.     }
  30.     companion object {
  31.         fun create(): Semantic {
  32.             val classLoader = Thread.currentThread().getContextClassLoader()
  34.             val tokenizerPath = classLoader.getResource("model/tokenizer.json")!!.toURI()
  35.             val onnxPath =  classLoader.getResource("model/model.onnx")!!.toURI()
  37.             try {
  38.                 val env: Map<String, String> = HashMap()
  39.                 val array: List<String> = tokenizerPath.toString().split("!")
  40.                 FileSystems.newFileSystem(URI.create(array[0]), env)
  41.             } catch (e: Exception) {
  42. //                e.printStackTrace()
  43.             }
  45.             val tokenizer = HuggingFaceTokenizer.newInstance(Paths.get(tokenizerPath))
  46.             val ortEnv = OrtEnvironment.getEnvironment()
  47.             val sessionOptions = OrtSession.SessionOptions()
  49.             // load onnxPath as byte[]
  50.             val onnxPathAsByteArray = Files.readAllBytes(Paths.get(onnxPath))
  52.             val session = ortEnv.createSession(onnxPathAsByteArray, sessionOptions)
  54.             return Semantic(tokenizer, session, ortEnv)
  55.         }
  56.     }
  57. }

LLM 应用开发模式：Stream 封装

服务端 API 调用：Kotlin 实现

机制：结合 callbackFlow 来实现

  1. fun stream(text: String): Flow<String> {
  2.     val systemMessage = ChatMessage(ChatMessageRole.USER.value(), text)
  4.     messages.add(systemMessage)
  6.     val completionRequest = ChatCompletionRequest.builder()
  7.         .model(openAiVersion)
  8.         .temperature(0.0)
  9.         .messages(messages)
  10.         .build()
  12.     return callbackFlow {
  13.         withContext(Dispatchers.IO) {
  14.             service.streamChatCompletion(completionRequest)
  15.                 .doOnError(Throwable::printStackTrace)
  16.                 .blockingForEach { response ->
  17.                     val completion = response.choices[0].message
  18.                     if (completion != null && completion.content != null) {
  19.                         trySend(completion.content)
  20.                     }
  21.                 }
  23.             close()
  24.         }
  25.     }
  26. }

客户端 API 调用：TypeScript 实现

机制：依赖于 Vercel 的 AI 库，提供对于 Stream 的封装

  1. import { Message, OpenAIStream, StreamingTextResponse } from 'ai'
  2. import { Configuration, OpenAIApi } from 'openai-edge'
  4. export async function stream(apiKey: string, messages: Message[], isStream: boolean = true) {
  5.   let basePath = process.env.OPENAI_PROXY_URL
  6.   if (basePath == null) {
  7.     basePath = 'https://api.openai.com'
  8.   }
  10.   const configuration = new Configuration({
  11.     apiKey: apiKey || process.env.OPENAI_API_KEY,
  12.     basePath
  13.   })
  15.   const openai = new OpenAIApi(configuration)
  17.   const res = await openai.createChatCompletion({
  18.     model: 'gpt-3.5-turbo',
  19.     messages,
  20.     temperature: 0.7,
  21.     stream: isStream
  22.   })
  24.   if (!isStream) {
  25.     return res
  26.   }
  28.   const stream = OpenAIStream(res, {})
  30.   return new StreamingTextResponse(stream)
  31. }

客户端 UI 实现：Fetch

  1. const decoder = new TextDecoder()
  3. export function decodeAIStreamChunk(chunk: Uint8Array): string {
  4.   return decoder.decode(chunk)
  5. }
  7. await fetch("/api/action/tooling", {
  8.   method: "POST",
  9.   body: JSON.stringify(tooling),
  10. }).then(async response => {
  11.   onResult(await response.json())
  12.   let result = ""
  13.   const reader = response.body.getReader()
  14.   while (true) {
  15.     const { done, value } = await reader.read()
  16.     if (done) {
  17.       break
  18.     }
  20.     result += decodeAIStreamChunk(value)
  21.     onResult(result)
  22.   }
  24.   isPending = false
  25. });

服务端实现转发： Java + Spring

WebFlux + Spring Boot

  1. @RestController
  2. public class ChatController {
  4.     private WebClient webClient = WebClient.create();
  6.     @PostMapping(value = "/api/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
  7.     public SseEmitter chat(@RequestBody ChatInput input) throws IOException {
  8.         SseEmitter emitter = new SseEmitter();
  10.         webClient.post()
  11.                 .uri(REMOTE_URL)
  12.                 .bodyValue(input)
  13.                 .exchangeToFlux(response -> {
  14.                     if (response.statusCode().is2xxSuccessful()) {
  15.                         return response.bodyToFlux(byte[].class)
  16.                                 .map(String::new)
  17.                                 .doOnNext(string -> {
  18.                                     try {
  19.                                         emitter.send(string);
  20.                                     } catch (IOException e) {
  21.                                         logger.error("Error while sending data: {}", e.getMessage());
  22.                                         emitter.completeWithError(e);
  23.                                     }
  24.                                 })
  25.                                 .doOnComplete(emitter::complete)
  26.                                 .doOnError(emitter::completeWithError);
  27.                     } else {
  28.                         emitter.completeWithError(new RuntimeException("Error while calling remote service"));
  29.                     }
  30.                 })
  31.                 .subscribe();
  33.         return emitter;
  34.     }
  35. }

服务端转发：Python

FastAPI + OpenAI

  1. def generate_reply_stream(input_data: ChatInput):
  2.     prompt = input_data.message
  3.     try:
  4.         prompt = prompt
  5.         response = openai.ChatCompletion.create(
  6.             model=openai_model,
  7.             temperature=temperature,
  8.             max_tokens=max_tokens,
  9.             n=max_responses,
  10.             messages=[
  11.                 {"role": "user", "content": prompt},
  12.             ],
  13.             stream=True,
  14.         )
  15.     except Exception as e:
  16.         print("Error in creating campaigns from openAI:", str(e))
  17.         raise HTTPException(503, error503)
  18.     try:
  19.         for chunk in response:
  20.             current_content = chunk["choices"][0]["delta"].get("content", "")
  21.             yield current_content
  23.     except Exception as e:
  24.         print("OpenAI Response (Streaming) Error: " + str(e))
  25.         raise HTTPException(503, error503)
  28. @app.post("/api/chat", response_class=Response)
  29. async def chat(input_data: ChatInput):
  30.     return StreamingResponse(generate_reply_stream(input_data), media_type="text/event-stream")