Langchain支持很方便的OpenAI模型的調用，可以做到快速開發大模型應用。但是要使用Huggingface上的開源模型就沒有那么方便了，本文就詳細闡述如何用Langchain開發基于Huggingface上的模型，并實時返回生成結果。

實時返回生成結果是LLM很關鍵的一環，大模型的響應速度很大程度上會影響用戶的使用體驗，較長的等待時間會導致用戶流失。同時，Langchain可以很方便的調用openAI的接口，但是對于我們這種窮屌絲來說用不起GPT的接口，只能用huggingface上的開源模型。所以本文將詳細介紹如何使用Langchain調用Huggingface的模型并做到實時返回生成結果。

本文選用Qwen2.5-0.5B-Instruct作為部署的模型，同時我是下載到了本地，所以代碼中的路徑是本地路徑，如果不想下載到本地的話直接用Huggingface上的路徑即可。

1. Quick start

如果使用OpenAI的模型，要實現實施返回結果（即流式調用），只需要以下幾行代碼就可以快速實現：

from langchain_openai import ChatOpenAImodel = ChatOpenAI(model='gpt-4')
chunks = []
for chunk in model.stream('天空是什么顏色？'):chunks.append(chunk)print(chunk.content, end='|', flush=True)

但是鑒于我們是窮批，用不起GPT的API，所以我們只能借助transformers庫自己實現上述的功能。

首先加載模型及其 tokenizer：

from transformers import Qwen2Tokenizer, Qwen2ForCausalLMtokenizer = Qwen2Tokenizer.from_pretrained(r'D:\huggingface\Qwen2.5-0.5B-Instruct')
model = Qwen2ForCausalLM.from_pretrained(r'D:\huggingface\Qwen2.5-0.5B-Instruct')

接著我們采用以下代碼即可實現流式迭代返回生成結果：

import threading
from transformers import TextIteratorStreamer
from langchain_core.messages import AIMessageChunkdef hf_stream(prompt: str):inputs = tokenizer(prompt, return_tensors='pt')# 創建一個 “流式文本迭代器”, 每當模型生成一個新 token，就會立刻把它變成字符串，通過 streamer 吐出來（yield）streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)generation_kwargs = dict(**inputs,streamer=streamer,max_new_tokens=100,do_sample=True,temperature=0.95,)# 啟動模型的文本生成過程，但用新線程異步執行，讓主線程能實時處理輸出# 因為 model.generate() 是 阻塞函數（會等生成完才返回），我們要“邊生成邊取結果”，所以不能直接運行它，而是讓它在后臺線程里跑thread = threading.Thread(target=model.generate, kwargs=generation_kwargs)thread.start()# 流式迭代返回的 token 段for new_text in streamer:yield AIMessageChunk(content=new_text)

上述代碼中需要注意的有幾個地方：streamer 是一個關鍵點，這里是將 tokenizer 放入到流式處理中，這是因為 model.generate 是個同步操作，無法執行異步調用，所以采用多線程的方式將 model.generate 放入到多線程中，這樣就可以多線程執行模型的生成。同時，由于 generate 的結果是 token id 而非文字，這里 tokenzier 會將 token id decode 成為文字，并用 yield 流式輸出，所以 streamer 是將 tokenizer 放入到的流中。

接著就可以直接查看模型的輸出結果：

for chunk in hf_stream('請為我介紹夢幻西游的天宮門派'):print(chunk.content, end='|')

由于用了end='|'，所以在每個輸出字符后都會看到"|"，以下是運行結果：

2. 采用鏈式調用的方式執行代碼

Langchain 一個核心功能就是鏈式調用，即我們可以用諸如:

chain = prompt | llm | parser

的方式，同時執行提示模板、大模型生成、結構化輸出的功能。當然，由于窮批用不起API，所以我們這里依舊自己打造 Langchain 調用 huggingface 的鏈式調用。

首先我們定義如下的類：

from langchain_core.runnables import Runnable
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParserclass HuggingFaceStreamWrapper(Runnable):def __init__(self, model, tokenizer, max_new_tokens=128):self.model = modelself.tokenizer = tokenizerself.max_new_tokens = max_new_tokensdef invoke(self, input_text, config=None) -> AIMessage:# Runnable 類要求強制實現抽象方法, 否則會報錯 TypeError: Can't instantiate abstract class with abstract method invokehuman_texts = [msg.content for msg in input_text.messages if isinstance(msg, HumanMessage)]prompt = "\n".join(human_texts)inputs = self.tokenizer(prompt, return_tensors='pt').to(self.model.device)outputs = self.model.generate(**inputs,max_new_tokens=self.max_new_tokens,do_sample=True,top_p=0.95,temperature=0.8,)decoded = self.tokenizer.decode(outputs[0], skip_special_tokens=True)return AIMessage(content=decoded)async def astream(self, input_text, config=None):human_texts = [msg.content for msg in input_text.messages if isinstance(msg, HumanMessage)]prompt = "\n".join(human_texts)# tokenizer encodeinputs = self.tokenizer(prompt, return_tensors='pt').to(self.model.device)# streamerstreamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True, skip_special_tokens=True)# 生成參數generation_kwargs = dict(**inputs,streamer=streamer,max_new_tokens=self.max_new_tokens,do_sample=True,top_p=0.95,temperature=0.8,)# 用線程調用生成（阻塞生成轉異步）thread = threading.Thread(target=self.model.generate, kwargs=generation_kwargs)thread.start()for token in streamer:# 手動觸發事件循環調度，允許其他任務執行。# 它的作用不是延時，而是調度讓步await asyncio.sleep(0)  # 允許事件循環切換# yield 是 Python 中的一個“進階但極其實用”的關鍵字，它的作用是讓函數變成一個生成器（generator），# 實現“邊計算邊返回”的效果，非常適合處理大數據、流式生成、異步 LLM 響應等場景。# yield 會暫停函數執行，返回一個值，但不會結束函數；yield AIMessageChunk(content=token)

這個是繼承了 langchain Runable 類，用于我們自定義開發 langchain 的鏈式調用。在這里有幾個要注意的點：Runable 中有個方法，名為 invoke，這個函數是必須要實現的函數，就算是

def invoke(self):pass

都沒問題，但是如果沒有該函數，那么會報錯：

TypeError: Can't instantiate abstract class with abstract method invoke

說明 invoke 這個函數是必須實現的。

而調用 invoke 函數也能夠返回結果，不過 invoke 是同步執行的，模型會一次性返回所有的結果，而非一個字一個字的蹦出來，比如執行下面代碼：

hf_model = HuggingFaceStreamWrapper(model, tokenizer)# 構建鏈
prompt = ChatPromptTemplate.from_template('請給我介紹夢幻西游中{topic}門派')
parser = StrOutputParser()
chain = prompt | hf_model | parser# 如果調用 invoke 方法, 必須實現 invoke 函數, 否則可以 pass
outputs = chain.invoke({'topic': '九黎城'})print(outputs)

會獲得：

就是這樣一次性輸出出來（前面我刪了三百多幀），這種用戶體驗感就會很差。

在上述代碼中，我們可以看到，采用 ChatPromptTemplate.from_template('請給我介紹夢幻西游中{topic}門派') 可以生成一個提示語，之后我們在調用的時候，傳入 topic 參數，就能夠直接將 九黎城 傳入進去，這樣在做實際開發的時候，就能夠給定一個提示模板，由用戶自行填充內容。

那么說完了 invoke 調用，鏈式調用流式輸出其實就很簡單了，只需要用如下代碼即可實現：

hf_model = HuggingFaceStreamWrapper(model, tokenizer)# 構建鏈
prompt = ChatPromptTemplate.from_template('請給我介紹夢幻西游中{topic}門派')
parser = StrOutputParser()
chain = prompt | hf_model | parserasync def async_stream():# 這里同樣, 如果要調用 astream 方法, 必須實現 astream 函數, 否則可以 passasync for chunk in chain.astream({'topic': '九黎城'}):print(chunk, end='|', flush=True)asyncio.run(async_stream())

以下是結果：