Nanbeige4.1-3B Chainlit调用教程自定义system prompt、流式响应、历史会话管理1. 开篇为什么你需要这个教程如果你已经用vLLM部署好了Nanbeige4.1-3B模型也看到了那个简洁的Chainlit前端界面但总觉得还差点意思——比如想让它扮演一个专业的编程助手或者希望它能记住之前的对话又或者想看到文字一个字一个字蹦出来的效果——那么你来对地方了。这个教程就是要带你超越基础的“提问-回答”解锁Chainlit调用Nanbeige4.1-3B的进阶玩法。我们将一步步实现三个核心功能自定义System Prompt让模型拥有特定“人设”比如“你是一个Python专家”。流式响应告别等待让回复像打字一样实时显示。历史会话管理让模型拥有“记忆”进行连贯的多轮对话。整个过程不需要复杂的理论全是可运行的代码和清晰的步骤。让我们开始吧。2. 环境与基础准备在开始魔改之前我们需要确保基础环境是通的。假设你已经通过镜像成功部署了Nanbeige4.1-3B模型并且能看到类似下面的日志说明模型服务已经就绪。2.1 确认模型服务状态通常部署成功后我们需要检查模型服务是否正常运行。你可以通过查看日志文件来确认# 查看模型服务的运行日志 cat /root/workspace/llm.log如果看到日志中包含模型加载成功、服务启动在某个端口例如8000的信息就说明一切正常。这是后续所有操作的基础。2.2 理解基础调用流程目前最简单的调用方式就是通过Chainlit的默认界面。你打开Chainlit前端输入问题比如“Which number is bigger, 9.11 or 9.8?”模型会返回答案。这背后的简单流程是你在前端输入问题。Chainlit将问题发送给后端的vLLM服务。vLLM服务调用Nanbeige4.1-3B模型生成答案。答案返回并显示在Chainlit界面上。我们的目标就是深入这个流程的第二步和第三步加入我们想要的“调料”。3. 核心功能一自定义System PromptSystem Prompt就像是给AI模型的“角色设定”或“初始指令”。它能在对话开始前就悄悄告诉模型“嘿你现在应该以什么身份、什么风格来回答问题。”3.1 为什么要自定义默认情况下模型可能以一个通用的、中立的助手身份回应。但通过System Prompt我们可以专业化让它成为“代码审查员”、“创意写作教练”或“技术文档翻译”。控制输出格式要求它“始终用JSON格式回答”或“先总结再分点论述”。设定行为边界例如“不要编造你不知道的信息”。3.2 如何在Chainlit中实现Chainlit本身提供了优雅的方式来设置System Prompt。我们需要修改或创建一个Chainlit的应用文件通常是app.py。基础示例设置一个Python助手我们创建一个新的app.py文件# app.py import chainlit as cl import sys import os # 假设我们将调用逻辑放在另一个模块这里先导入 sys.path.append(os.path.dirname(__file__)) from llm_client import get_llm_response # 这是一个我们待会要写的函数 # 这是一个关键的装饰器用于设置每轮对话的初始提示 cl.on_chat_start async def on_chat_start(): # 设置系统提示词 (System Prompt) system_prompt 你是一个资深的Python开发专家专注于编写高效、可读且符合PEP 8规范的代码。你的回答应该专业、清晰并优先提供代码示例。如果用户的问题不明确请礼貌地请求澄清。 # 将system_prompt存入用户会话状态中方便后续调用 cl.user_session.set(system_prompt, system_prompt) # 可选发送一个欢迎消息告知用户AI的角色 await cl.Message( contentf你好我已准备好作为你的Python开发专家。有什么代码问题需要我帮忙吗 ).send() cl.on_message async def on_message(message: cl.Message): # 从会话中获取预设的system_prompt system_prompt cl.user_session.get(system_prompt) # 获取用户输入 user_input message.content # 构建发送给模型的完整消息。 # 通常对于遵循ChatML等格式的模型消息结构如下 full_prompt [ {role: system, content: system_prompt}, {role: user, content: user_input} ] # 调用我们后端的LLM服务并传入构建好的prompt # 注意这里的get_llm_response函数需要我们自己实现用于与vLLM API通信 response await get_llm_response(full_prompt) # 将模型的回复发送给前端 await cl.Message(contentresponse).send()代码解释cl.on_chat_start这个装饰器下的函数会在每次新聊天会话开始时运行。这是设置全局性System Prompt的理想位置。cl.user_session.set()Chainlit提供的会话状态管理可以存储和读取当前用户会话中的数据。这里我们把System Prompt存起来。cl.on_message这个装饰器下的函数会处理用户发送的每一条消息。我们在处理消息时从会话状态中取出system_prompt然后将其与用户的输入一起构造成一个包含“系统”和“用户”角色的消息列表。这种结构被大多数Chat模型所认可。最后将这个结构化的消息列表发送给后端的LLM服务。现在只要用户开始新对话模型就会以“Python开发专家”的身份来回答问题。4. 核心功能二实现流式响应流式响应能让用户体验发生质变。与其等待模型生成完整答案可能耗时几秒到十几秒不如让答案逐词或逐句实时显示出来就像有人在打字一样。4.1 流式响应的原理简单说就是不让模型一次性生成全部文本再返回而是让它“边想边说”。后端每生成一小段一个token或几个token就立刻推送给前端。Chainlit为这种模式提供了原生支持。4.2 修改代码以支持流式我们需要做两处关键修改让后端的LLM服务支持流式输出即返回一个可迭代的对象而不是一个完整的字符串。让Chainlit前端能够接收并显示这个流。首先更新我们的llm_client.py假设的与vLLM通信的客户端使其支持流式调用。这里我们以调用vLLM的OpenAI兼容API为例# llm_client.py import aiohttp import json from typing import AsyncGenerator # 你的vLLM服务地址根据你的部署情况修改 VLLM_SERVER_URL http://localhost:8000/v1/chat/completions async def get_llm_response_stream(messages: list) - AsyncGenerator[str, None]: 以流式方式获取LLM响应。 返回一个异步生成器每次yield一个文本块。 headers { Content-Type: application/json } payload { model: nanbeige-4.1-3b, # 模型名称根据实际修改 messages: messages, stream: True, # 关键参数开启流式输出 temperature: 0.7, max_tokens: 1024 } async with aiohttp.ClientSession() as session: async with session.post(VLLM_SERVER_URL, jsonpayload, headersheaders) as resp: if resp.status ! 200: error_text await resp.text() yield f请求后端服务出错: {resp.status}, {error_text} return # 流式读取响应 async for line in resp.content: if line: line_decoded line.decode(utf-8).strip() if line_decoded.startswith(data: ): data_str line_decoded[6:] # 去掉 data: 前缀 if data_str [DONE]: break try: data json.loads(data_str) # 从响应中提取增量内容 content_delta data.get(choices, [{}])[0].get(delta, {}).get(content, ) if content_delta: yield content_delta except json.JSONDecodeError: # 忽略非JSON行或解析错误 continue然后修改我们的app.py中的消息处理函数使用这个流式客户端# app.py (更新 on_message 函数) cl.on_message async def on_message(message: cl.Message): system_prompt cl.user_session.get(system_prompt) user_input message.content full_prompt [ {role: system, content: system_prompt}, {role: user, content: user_input} ] # 创建一个Chainlit的消息对象用于流式更新内容 msg cl.Message(content) await msg.send() # 先发送一个空消息 # 调用流式函数并逐步更新消息内容 async for chunk in get_llm_response_stream(full_prompt): await msg.stream_token(chunk) # 关键流式传输每个token # 流式传输完成 await msg.update()代码解释get_llm_response_stream函数它向vLLM服务器发送请求时设置了stream: True。然后它异步地读取服务器返回的每一行数据SSE格式解析出新增的文本内容(content_delta)并通过yield逐个返回。在on_message中我们不再等待完整响应。而是先发送一个空的cl.Message然后在一个异步循环中不断从流式生成器中获取文本块(chunk)并使用msg.stream_token(chunk)将其追加到前端显示的消息中。await msg.update()在所有内容流式传输完毕后更新消息状态。这样用户就能看到答案像打字一样逐渐出现了。5. 核心功能三管理历史会话多轮对话的核心是“记忆”。我们需要让模型知道之前聊过什么才能做出连贯的回应。这意味着我们需要在每次请求中不仅发送当前问题还要附带上之前的对话历史。5.1 在Chainlit中维护历史Chainlit的cl.user_session同样是我们管理历史的好帮手。我们可以把历史对话记录成一个消息列表。# app.py (进一步更新 on_message 函数并添加历史管理) cl.on_chat_start async def on_chat_start(): system_prompt 你是一个资深的Python开发专家... cl.user_session.set(system_prompt, system_prompt) # 初始化一个空列表用于存储对话历史 cl.user_session.set(message_history, []) await cl.Message(contentf你好Python专家已上线。我们可以进行多轮对话了。 ).send() cl.on_message async def on_message(message: cl.Message): system_prompt cl.user_session.get(system_prompt) user_input message.content # 1. 获取历史记录 history: list cl.user_session.get(message_history) # 2. 构建本次请求的消息列表 # 首先加入系统提示 messages_for_llm [{role: system, content: system_prompt}] # 然后加入历史对话假设历史中已经存储了交替的 user 和 assistant 消息 messages_for_llm.extend(history) # 最后加入当前用户的新消息 messages_for_llm.append({role: user, content: user_input}) # 3. 调用LLM这里以流式为例 msg cl.Message(content) await msg.send() full_response # 用于收集完整的响应以便存入历史 async for chunk in get_llm_response_stream(messages_for_llm): await msg.stream_token(chunk) full_response chunk await msg.update() # 4. 更新历史记录 # 将本次交互的“用户消息”和“助手回复”都存入历史 history.append({role: user, content: user_input}) history.append({role: assistant, content: full_response}) # 可选为了避免历史过长导致token超限可以设置一个最大长度 MAX_HISTORY_LENGTH 10 # 保留最近5轮对话10条消息 if len(history) MAX_HISTORY_LENGTH: # 移除最老的消息但保留系统提示后的最早几条或者移除最老的几对QA。 # 这里简单地从头部删除直到长度符合要求。 # 更复杂的策略可以移除中间某些轮次。 del history[:len(history) - MAX_HISTORY_LENGTH] cl.user_session.set(message_history, history)5.2 历史管理的注意事项Token限制模型有上下文长度限制。历史对话越长消耗的Token越多可能导致后续请求失败或截断。上面的代码提供了一个简单的截断机制。性能历史消息会随着每次请求发送给后端可能增加网络传输和模型处理的开销。重置历史你可以通过添加一个cl.on_chat_start中的重置按钮或者在会话中提供一个“清除历史”的功能来手动重置message_history。6. 完整示例与效果验证让我们把上面的所有功能整合到一个更完善的app.py示例中并看看效果。# app.py - 完整功能示例 import chainlit as cl import aiohttp import json from typing import AsyncGenerator VLLM_SERVER_URL http://localhost:8000/v1/chat/completions async def get_llm_response_stream(messages: list) - AsyncGenerator[str, None]: 流式调用LLM的客户端函数 headers {Content-Type: application/json} payload { model: nanbeige-4.1-3b, messages: messages, stream: True, temperature: 0.7, max_tokens: 1024 } async with aiohttp.ClientSession() as session: async with session.post(VLLM_SERVER_URL, jsonpayload, headersheaders) as resp: if resp.status ! 200: error_text await resp.text() yield fError: {resp.status}, {error_text} return async for line in resp.content: if line: line_decoded line.decode(utf-8).strip() if line_decoded.startswith(data: ): data_str line_decoded[6:] if data_str [DONE]: break try: data json.loads(data_str) content_delta data.get(choices, [{}])[0].get(delta, {}).get(content, ) if content_delta: yield content_delta except json.JSONDecodeError: continue cl.on_chat_start async def start(): # 1. 设置自定义System Prompt cl.user_session.set( system_prompt, 你是一个乐于助人且知识渊博的AI助手。你的回答应当准确、清晰、有条理。如果遇到不确定的问题请诚实说明。 ) # 2. 初始化对话历史 cl.user_session.set(message_history, []) await cl.Message(content会话已初始化。我可以记住我们的对话历史了。请问有什么可以帮您).send() cl.on_message async def main(message: cl.Message): system_prompt cl.user_session.get(system_prompt) history cl.user_session.get(message_history) user_msg message.content # 构建本次请求的完整消息列表 messages_to_send [{role: system, content: system_prompt}] messages_to_send.extend(history) messages_to_send.append({role: user, content: user_msg}) # 创建并发送一个空消息用于流式更新 msg cl.Message(content) await msg.send() full_response # 3. 流式获取响应 try: async for chunk in get_llm_response_stream(messages_to_send): await msg.stream_token(chunk) full_response chunk except Exception as e: full_response f调用模型服务时发生错误: {str(e)} await msg.stream_token(full_response) finally: await msg.update() # 4. 更新历史记录仅当成功获取响应时 if not full_response.startswith(Error): history.append({role: user, content: user_msg}) history.append({role: assistant, content: full_response}) # 简单历史长度管理保留最近6轮对话12条消息 if len(history) 12: del history[:2] # 删除最老的一轮对话一条user一条assistant cl.user_session.set(message_history, history) # 运行Chainlit应用 if __name__ __main__: # 假设这个文件叫 app.py运行命令是: chainlit run app.py pass如何运行和验证保存代码将上面的完整示例保存为app.py放在你的工作目录。确保服务运行确认你的Nanbeige4.1-3B的vLLM服务正在运行端口8000。启动Chainlit在终端运行chainlit run app.py。它会启动一个服务默认端口8000可能与vLLM冲突注意修改或使用--port参数指定新端口如chainlit run app.py --port 8080。打开浏览器访问Chainlit提供的地址如http://localhost:8080。验证功能自定义Prompt发送问题观察回复风格是否符合“乐于助人且知识渊博”的设定。流式响应输入一个稍复杂的问题你应该能看到文字逐词出现。历史管理先问“什么是Python”再基于它的回答追问“它有什么优点”看模型是否能基于上一轮的回答进行连贯对话。7. 总结与进阶思考通过本教程我们成功地为基于Chainlit的Nanbeige4.1-3B调用实现了三个关键增强功能自定义System Prompt让你能轻松为模型赋予不同的角色和任务指令极大地扩展了其应用场景。流式响应显著提升了交互体验让等待过程变得可感知更符合人类对话的自然节奏。历史会话管理通过维护一个上下文消息列表使模型具备了多轮对话的“记忆”能力能够进行连贯、深入的交流。更进一步前端定制Chainlit支持自定义前端UI元素按钮、侧边栏、文件上传等你可以打造更复杂的交互界面。复杂历史策略实现更智能的历史摘要或选择性记忆以应对超长对话。多模态集成如果你的模型支持可以在Chainlit中集成图片上传和分析功能。错误处理与超时为网络请求和模型调用添加更健壮的错误处理和超时机制。这些功能的实现将你的AI应用从简单的“问答机”升级为了一个更具交互性、个性化和实用性的智能对话伙伴。希望这篇教程能帮助你更好地驾驭Nanbeige4.1-3B和Chainlit构建出更强大的AI应用。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Nanbeige4.1-3B Chainlit调用教程：自定义system prompt、流式响应、历史会话管理
Nanbeige4.1-3B Chainlit调用教程自定义system prompt、流式响应、历史会话管理1. 开篇为什么你需要这个教程如果你已经用vLLM部署好了Nanbeige4.1-3B模型也看到了那个简洁的Chainlit前端界面但总觉得还差点意思——比如想让它扮演一个专业的编程助手或者希望它能记住之前的对话又或者想看到文字一个字一个字蹦出来的效果——那么你来对地方了。这个教程就是要带你超越基础的“提问-回答”解锁Chainlit调用Nanbeige4.1-3B的进阶玩法。我们将一步步实现三个核心功能自定义System Prompt让模型拥有特定“人设”比如“你是一个Python专家”。流式响应告别等待让回复像打字一样实时显示。历史会话管理让模型拥有“记忆”进行连贯的多轮对话。整个过程不需要复杂的理论全是可运行的代码和清晰的步骤。让我们开始吧。2. 环境与基础准备在开始魔改之前我们需要确保基础环境是通的。假设你已经通过镜像成功部署了Nanbeige4.1-3B模型并且能看到类似下面的日志说明模型服务已经就绪。2.1 确认模型服务状态通常部署成功后我们需要检查模型服务是否正常运行。你可以通过查看日志文件来确认# 查看模型服务的运行日志 cat /root/workspace/llm.log如果看到日志中包含模型加载成功、服务启动在某个端口例如8000的信息就说明一切正常。这是后续所有操作的基础。2.2 理解基础调用流程目前最简单的调用方式就是通过Chainlit的默认界面。你打开Chainlit前端输入问题比如“Which number is bigger, 9.11 or 9.8?”模型会返回答案。这背后的简单流程是你在前端输入问题。Chainlit将问题发送给后端的vLLM服务。vLLM服务调用Nanbeige4.1-3B模型生成答案。答案返回并显示在Chainlit界面上。我们的目标就是深入这个流程的第二步和第三步加入我们想要的“调料”。3. 核心功能一自定义System PromptSystem Prompt就像是给AI模型的“角色设定”或“初始指令”。它能在对话开始前就悄悄告诉模型“嘿你现在应该以什么身份、什么风格来回答问题。”3.1 为什么要自定义默认情况下模型可能以一个通用的、中立的助手身份回应。但通过System Prompt我们可以专业化让它成为“代码审查员”、“创意写作教练”或“技术文档翻译”。控制输出格式要求它“始终用JSON格式回答”或“先总结再分点论述”。设定行为边界例如“不要编造你不知道的信息”。3.2 如何在Chainlit中实现Chainlit本身提供了优雅的方式来设置System Prompt。我们需要修改或创建一个Chainlit的应用文件通常是app.py。基础示例设置一个Python助手我们创建一个新的app.py文件# app.py import chainlit as cl import sys import os # 假设我们将调用逻辑放在另一个模块这里先导入 sys.path.append(os.path.dirname(__file__)) from llm_client import get_llm_response # 这是一个我们待会要写的函数 # 这是一个关键的装饰器用于设置每轮对话的初始提示 cl.on_chat_start async def on_chat_start(): # 设置系统提示词 (System Prompt) system_prompt 你是一个资深的Python开发专家专注于编写高效、可读且符合PEP 8规范的代码。你的回答应该专业、清晰并优先提供代码示例。如果用户的问题不明确请礼貌地请求澄清。 # 将system_prompt存入用户会话状态中方便后续调用 cl.user_session.set(system_prompt, system_prompt) # 可选发送一个欢迎消息告知用户AI的角色 await cl.Message( contentf你好我已准备好作为你的Python开发专家。有什么代码问题需要我帮忙吗 ).send() cl.on_message async def on_message(message: cl.Message): # 从会话中获取预设的system_prompt system_prompt cl.user_session.get(system_prompt) # 获取用户输入 user_input message.content # 构建发送给模型的完整消息。 # 通常对于遵循ChatML等格式的模型消息结构如下 full_prompt [ {role: system, content: system_prompt}, {role: user, content: user_input} ] # 调用我们后端的LLM服务并传入构建好的prompt # 注意这里的get_llm_response函数需要我们自己实现用于与vLLM API通信 response await get_llm_response(full_prompt) # 将模型的回复发送给前端 await cl.Message(contentresponse).send()代码解释cl.on_chat_start这个装饰器下的函数会在每次新聊天会话开始时运行。这是设置全局性System Prompt的理想位置。cl.user_session.set()Chainlit提供的会话状态管理可以存储和读取当前用户会话中的数据。这里我们把System Prompt存起来。cl.on_message这个装饰器下的函数会处理用户发送的每一条消息。我们在处理消息时从会话状态中取出system_prompt然后将其与用户的输入一起构造成一个包含“系统”和“用户”角色的消息列表。这种结构被大多数Chat模型所认可。最后将这个结构化的消息列表发送给后端的LLM服务。现在只要用户开始新对话模型就会以“Python开发专家”的身份来回答问题。4. 核心功能二实现流式响应流式响应能让用户体验发生质变。与其等待模型生成完整答案可能耗时几秒到十几秒不如让答案逐词或逐句实时显示出来就像有人在打字一样。4.1 流式响应的原理简单说就是不让模型一次性生成全部文本再返回而是让它“边想边说”。后端每生成一小段一个token或几个token就立刻推送给前端。Chainlit为这种模式提供了原生支持。4.2 修改代码以支持流式我们需要做两处关键修改让后端的LLM服务支持流式输出即返回一个可迭代的对象而不是一个完整的字符串。让Chainlit前端能够接收并显示这个流。首先更新我们的llm_client.py假设的与vLLM通信的客户端使其支持流式调用。这里我们以调用vLLM的OpenAI兼容API为例# llm_client.py import aiohttp import json from typing import AsyncGenerator # 你的vLLM服务地址根据你的部署情况修改 VLLM_SERVER_URL http://localhost:8000/v1/chat/completions async def get_llm_response_stream(messages: list) - AsyncGenerator[str, None]: 以流式方式获取LLM响应。 返回一个异步生成器每次yield一个文本块。 headers { Content-Type: application/json } payload { model: nanbeige-4.1-3b, # 模型名称根据实际修改 messages: messages, stream: True, # 关键参数开启流式输出 temperature: 0.7, max_tokens: 1024 } async with aiohttp.ClientSession() as session: async with session.post(VLLM_SERVER_URL, jsonpayload, headersheaders) as resp: if resp.status ! 200: error_text await resp.text() yield f请求后端服务出错: {resp.status}, {error_text} return # 流式读取响应 async for line in resp.content: if line: line_decoded line.decode(utf-8).strip() if line_decoded.startswith(data: ): data_str line_decoded[6:] # 去掉 data: 前缀 if data_str [DONE]: break try: data json.loads(data_str) # 从响应中提取增量内容 content_delta data.get(choices, [{}])[0].get(delta, {}).get(content, ) if content_delta: yield content_delta except json.JSONDecodeError: # 忽略非JSON行或解析错误 continue然后修改我们的app.py中的消息处理函数使用这个流式客户端# app.py (更新 on_message 函数) cl.on_message async def on_message(message: cl.Message): system_prompt cl.user_session.get(system_prompt) user_input message.content full_prompt [ {role: system, content: system_prompt}, {role: user, content: user_input} ] # 创建一个Chainlit的消息对象用于流式更新内容 msg cl.Message(content) await msg.send() # 先发送一个空消息 # 调用流式函数并逐步更新消息内容 async for chunk in get_llm_response_stream(full_prompt): await msg.stream_token(chunk) # 关键流式传输每个token # 流式传输完成 await msg.update()代码解释get_llm_response_stream函数它向vLLM服务器发送请求时设置了stream: True。然后它异步地读取服务器返回的每一行数据SSE格式解析出新增的文本内容(content_delta)并通过yield逐个返回。在on_message中我们不再等待完整响应。而是先发送一个空的cl.Message然后在一个异步循环中不断从流式生成器中获取文本块(chunk)并使用msg.stream_token(chunk)将其追加到前端显示的消息中。await msg.update()在所有内容流式传输完毕后更新消息状态。这样用户就能看到答案像打字一样逐渐出现了。5. 核心功能三管理历史会话多轮对话的核心是“记忆”。我们需要让模型知道之前聊过什么才能做出连贯的回应。这意味着我们需要在每次请求中不仅发送当前问题还要附带上之前的对话历史。5.1 在Chainlit中维护历史Chainlit的cl.user_session同样是我们管理历史的好帮手。我们可以把历史对话记录成一个消息列表。# app.py (进一步更新 on_message 函数并添加历史管理) cl.on_chat_start async def on_chat_start(): system_prompt 你是一个资深的Python开发专家... cl.user_session.set(system_prompt, system_prompt) # 初始化一个空列表用于存储对话历史 cl.user_session.set(message_history, []) await cl.Message(contentf你好Python专家已上线。我们可以进行多轮对话了。 ).send() cl.on_message async def on_message(message: cl.Message): system_prompt cl.user_session.get(system_prompt) user_input message.content # 1. 获取历史记录 history: list cl.user_session.get(message_history) # 2. 构建本次请求的消息列表 # 首先加入系统提示 messages_for_llm [{role: system, content: system_prompt}] # 然后加入历史对话假设历史中已经存储了交替的 user 和 assistant 消息 messages_for_llm.extend(history) # 最后加入当前用户的新消息 messages_for_llm.append({role: user, content: user_input}) # 3. 调用LLM这里以流式为例 msg cl.Message(content) await msg.send() full_response # 用于收集完整的响应以便存入历史 async for chunk in get_llm_response_stream(messages_for_llm): await msg.stream_token(chunk) full_response chunk await msg.update() # 4. 更新历史记录 # 将本次交互的“用户消息”和“助手回复”都存入历史 history.append({role: user, content: user_input}) history.append({role: assistant, content: full_response}) # 可选为了避免历史过长导致token超限可以设置一个最大长度 MAX_HISTORY_LENGTH 10 # 保留最近5轮对话10条消息 if len(history) MAX_HISTORY_LENGTH: # 移除最老的消息但保留系统提示后的最早几条或者移除最老的几对QA。 # 这里简单地从头部删除直到长度符合要求。 # 更复杂的策略可以移除中间某些轮次。 del history[:len(history) - MAX_HISTORY_LENGTH] cl.user_session.set(message_history, history)5.2 历史管理的注意事项Token限制模型有上下文长度限制。历史对话越长消耗的Token越多可能导致后续请求失败或截断。上面的代码提供了一个简单的截断机制。性能历史消息会随着每次请求发送给后端可能增加网络传输和模型处理的开销。重置历史你可以通过添加一个cl.on_chat_start中的重置按钮或者在会话中提供一个“清除历史”的功能来手动重置message_history。6. 完整示例与效果验证让我们把上面的所有功能整合到一个更完善的app.py示例中并看看效果。# app.py - 完整功能示例 import chainlit as cl import aiohttp import json from typing import AsyncGenerator VLLM_SERVER_URL http://localhost:8000/v1/chat/completions async def get_llm_response_stream(messages: list) - AsyncGenerator[str, None]: 流式调用LLM的客户端函数 headers {Content-Type: application/json} payload { model: nanbeige-4.1-3b, messages: messages, stream: True, temperature: 0.7, max_tokens: 1024 } async with aiohttp.ClientSession() as session: async with session.post(VLLM_SERVER_URL, jsonpayload, headersheaders) as resp: if resp.status ! 200: error_text await resp.text() yield fError: {resp.status}, {error_text} return async for line in resp.content: if line: line_decoded line.decode(utf-8).strip() if line_decoded.startswith(data: ): data_str line_decoded[6:] if data_str [DONE]: break try: data json.loads(data_str) content_delta data.get(choices, [{}])[0].get(delta, {}).get(content, ) if content_delta: yield content_delta except json.JSONDecodeError: continue cl.on_chat_start async def start(): # 1. 设置自定义System Prompt cl.user_session.set( system_prompt, 你是一个乐于助人且知识渊博的AI助手。你的回答应当准确、清晰、有条理。如果遇到不确定的问题请诚实说明。 ) # 2. 初始化对话历史 cl.user_session.set(message_history, []) await cl.Message(content会话已初始化。我可以记住我们的对话历史了。请问有什么可以帮您).send() cl.on_message async def main(message: cl.Message): system_prompt cl.user_session.get(system_prompt) history cl.user_session.get(message_history) user_msg message.content # 构建本次请求的完整消息列表 messages_to_send [{role: system, content: system_prompt}] messages_to_send.extend(history) messages_to_send.append({role: user, content: user_msg}) # 创建并发送一个空消息用于流式更新 msg cl.Message(content) await msg.send() full_response # 3. 流式获取响应 try: async for chunk in get_llm_response_stream(messages_to_send): await msg.stream_token(chunk) full_response chunk except Exception as e: full_response f调用模型服务时发生错误: {str(e)} await msg.stream_token(full_response) finally: await msg.update() # 4. 更新历史记录仅当成功获取响应时 if not full_response.startswith(Error): history.append({role: user, content: user_msg}) history.append({role: assistant, content: full_response}) # 简单历史长度管理保留最近6轮对话12条消息 if len(history) 12: del history[:2] # 删除最老的一轮对话一条user一条assistant cl.user_session.set(message_history, history) # 运行Chainlit应用 if __name__ __main__: # 假设这个文件叫 app.py运行命令是: chainlit run app.py pass如何运行和验证保存代码将上面的完整示例保存为app.py放在你的工作目录。确保服务运行确认你的Nanbeige4.1-3B的vLLM服务正在运行端口8000。启动Chainlit在终端运行chainlit run app.py。它会启动一个服务默认端口8000可能与vLLM冲突注意修改或使用--port参数指定新端口如chainlit run app.py --port 8080。打开浏览器访问Chainlit提供的地址如http://localhost:8080。验证功能自定义Prompt发送问题观察回复风格是否符合“乐于助人且知识渊博”的设定。流式响应输入一个稍复杂的问题你应该能看到文字逐词出现。历史管理先问“什么是Python”再基于它的回答追问“它有什么优点”看模型是否能基于上一轮的回答进行连贯对话。7. 总结与进阶思考通过本教程我们成功地为基于Chainlit的Nanbeige4.1-3B调用实现了三个关键增强功能自定义System Prompt让你能轻松为模型赋予不同的角色和任务指令极大地扩展了其应用场景。流式响应显著提升了交互体验让等待过程变得可感知更符合人类对话的自然节奏。历史会话管理通过维护一个上下文消息列表使模型具备了多轮对话的“记忆”能力能够进行连贯、深入的交流。更进一步前端定制Chainlit支持自定义前端UI元素按钮、侧边栏、文件上传等你可以打造更复杂的交互界面。复杂历史策略实现更智能的历史摘要或选择性记忆以应对超长对话。多模态集成如果你的模型支持可以在Chainlit中集成图片上传和分析功能。错误处理与超时为网络请求和模型调用添加更健壮的错误处理和超时机制。这些功能的实现将你的AI应用从简单的“问答机”升级为了一个更具交互性、个性化和实用性的智能对话伙伴。希望这篇教程能帮助你更好地驾驭Nanbeige4.1-3B和Chainlit构建出更强大的AI应用。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
