(Deprecated) Generation API and parameters
under construction
The English version of this document is under construction and will be available soon.
Generate (請優先使用 Conversation)
一般使用
export API_KEY={API_KEY}
export API_URL={API_URL}
# model: ffm-mixtral-8x7b-32k-instruct
curl "${API_URL}/models/generate" \
-H "X-API-KEY:${API_KEY}" \
-H "X-API-HOST: afs-inference" \
-H "content-type: application/json" \
-d '{"model":"ffm-mixtral-8x7b-32k-instruct",
"inputs":"從前從前，有位老太太去河邊",
"parameters":{
"max_new_tokens":200,
"temperature":0.5,
"top_k":50,
"top_p":1,
"frequence_penalty":1}}'
輸出：包括生成的文字、token 個數以及所花費的時間秒數。
{
"generated_text": "，她洗完衣服後，要把衣服晾在河邊的一棵樹上。但老太太又老又弱，她爬不上樹，於是她決定把衣服掛在樹枝上。老太太拿起衣服，開始往樹枝上掛衣服，但她掛了幾件衣服後，樹枝斷了，所有的衣服都掉到河裡去了。老太太看到這一幕，非常傷心，她說：「我的衣服都掉到河裡去了！」老太太的孫女看到這一幕，心想：「我可以幫助我的祖母，我可以幫助她把衣服掛在樹枝上。」於是，這位孫女走到河邊，開始幫助她的祖母把衣服掛在樹枝上",<br/>
"function_call": null,
"details": null,
"total_time_taken": "18.88 sec",
"prompt_tokens": 17,
"generated_tokens": 200,
"total_tokens": 217,
"finish_reason": "length"
}
Python 範例
import json
import requests
MODEL_NAME = "ffm-mixtral-8x7b-32k-instruct"
API_KEY = "{API_KEY}"
API_URL = "{API_URL}"
API_HOST = "afs-inference"
# parameters
max_new_tokens = 200
temperature = 0.5
top_k = 50
top_p = 1.0
frequence_penalty = 1.0
def generate(prompt):
headers = {
"content-type": "application/json",
"X-API-Key": API_KEY,
"X-API-Host": API_HOST}
data = {
"model": MODEL_NAME,
"inputs": prompt,
"parameters": {
"max_new_tokens": max_new_tokens,
"temperature": temperature,
"top_k": top_k,
"top_p": top_p,
"frequence_penalty": frequence_penalty
}
}
result = ''
try:
response = requests.post(
API_URL + "/models/generate", json=data, headers=headers)
if response.status_code == 200:
result = json.loads(response.text, strict=False)['generated_text']
else:
print("error")
except:
print("error")
return result.strip("\n")
result = generate("從前從前，有位老太太去河邊")
print(result)
輸出：
，她洗完衣服後，要把衣服晾在河邊的一棵樹上。但老太太又老又弱，她爬不上樹，於是她決定把衣服掛在樹枝上。老太太拿起衣服，開始往樹枝上掛衣服，但她掛了幾件衣服後，樹枝斷了，所有的衣服都掉到河裡去了。老太太看到這一幕，非常傷心，她說：「我的衣服都掉到河裡去了！」老太太的孫女看到這一幕，心想：「我可以幫助我的祖母，我可以幫助她把衣服掛在樹枝上。」於是，這位孫女走到河邊，開始幫助她的祖母把衣服掛在樹枝上
export API_KEY={API_KEY}
export API_URL={API_URL}
# model: ffm-mixtral-8x7b-32k-instruct
curl "${API_URL}/models/generate" \
-H "X-API-KEY:${API_KEY}" \
-H "X-API-HOST: afs-inference" \
-H "content-type: application/json" \
-d '{"model":"ffm-mixtral-8x7b-32k-instruct",
"inputs":"可以幫我規劃台北兩日遊，並推薦每天的景點及說明其特色嗎？",
"parameters":{
"max_new_tokens":350,
"temperature":0.5,
"top_k":50,
"top_p":1,
"frequence_penalty":1}}'
輸出：包括生成的文字、token 個數以及所花費的時間秒數。
{
"generated_text": "答案是： 第一天： 1. 台北101觀景台 - 這是台北最受歡迎的景點之一，提供城市天際線的壯觀景色。 2. 國立故宮博物院 - 這是世界上最著名的藝術和文物收藏之一，展示了中國豐富的文化遺產。 3. 台北中正紀念堂 - 這是一座宏偉的紀念堂，致力於紀念中華民國前總統蔣中正。 4. 台北國立台灣博物館 - 這是一個展示台灣歷史、文化和藝術的博物館。 5. 台北夜市 - 這是一個熱鬧的夜市，提供各種街頭美食、購物和娛樂。 第二天： 1. 陽明山國家公園 - 這是一個美麗的國家公園，提供令人驚嘆的台北市區景色。 2. 台北101觀景台 - 這是另一個觀景台，提供城市天際線的壯觀景色。 3. 台北國立台灣博物館 - 這是另一個博物館，展示台灣歷史、文化和藝術。 4. 台北故宮 - 這是另一個展示中國豐富文化遺產的",<br/>
"function_call": null,
"details": null,
"total_time_taken": "33.08 sec",
"prompt_tokens": 31,
"generated_tokens": 350,
"total_tokens": 381,
"finish_reason": "length"
}
Python 範例
import json
import requests
MODEL_NAME = "ffm-mixtral-8x7b-32k-instruct"
API_KEY = "{API_KEY}"
API_URL = "{API_URL}"
API_HOST = "afs-inference"
# parameters
max_new_tokens = 350
temperature = 0.5
top_k = 50
top_p = 1.0
frequence_penalty = 1.0
def generate(prompt):
headers = {
"content-type": "application/json",
"X-API-Key": API_KEY,
"X-API-Host": API_HOST}
data = {
"model": MODEL_NAME,
"inputs": prompt,
"parameters": {
"max_new_tokens": max_new_tokens,
"temperature": temperature,
"top_k": top_k,
"top_p": top_p,
"frequence_penalty": frequence_penalty
}
}
result = ''
try:
response = requests.post(
API_URL + "/models/generate", json=data, headers=headers)
if response.status_code == 200:
result = json.loads(response.text, strict=False)['generated_text']
else:
print("error")
except:
print("error")
return result.strip("\n")
result = generate("可以幫我規劃台北兩日遊，並推薦每天的景點及說明其特色嗎？")
print(result)
輸出：
答案是： 第一天： 1. 台北101觀景台 - 這是台北最受歡迎的景點之一，提供城市天際線的壯觀景色。 2. 國立故宮博物院 - 這是世界上最著名的藝術和文物收藏之一，展示了中國豐富的文化遺產。 3. 台北中正紀念堂 - 這是一座宏偉的紀念堂，致力於紀念中華民國前總統蔣中正。 4. 台北國立台灣博物館 - 這是一個展示台灣歷史、文化和藝術的博物館。 5. 台北夜市 - 這是一個熱鬧的夜市，提供各種街頭美食、購物和娛樂。 第二天： 1. 陽明山國家公園 - 這是一個美麗的國家公園，提供令人驚嘆的台北市區景色。 2. 台北101觀景台 - 這是另一個觀景台，提供城市天際線的壯觀景色。 3. 台北國立台灣博物館 - 這是另一個博物館，展示台灣歷史、文化和藝術。 4. 台北故宮 - 這是另一個展示中國豐富文化遺產的
使用 Stream 模式
Server-sent event (SSE)：伺服器主動向客戶端推送資料，連線建立後，在一步步生成字句的同時也將資料往客戶端拋送，和先前的一次性回覆不同，可加強使用者體驗。
export API_KEY={API_KEY}
export API_URL={API_URL}
# model: ffm-mixtral-8x7b-32k-instruct
curl "${API_URL}/models/generate" \
-H "X-API-KEY:${API_KEY}" \
-H "X-API-HOST: afs-inference" \
-H "content-type: application/json" \
-d '{"model":"ffm-mixtral-8x7b-32k-instruct",
"inputs":"台灣最高峰是",
"stream":true,
"parameters":{
"max_new_tokens":2,
"temperature":0.5,
"top_k":50,
"top_p":1,
"frequence_penalty":1}}'
輸出：每個 token 會輸出一筆資料，最末筆則是會將先前生成的文字串成一筆、以及描述 token 個數和所花費的時間秒數。
data: {"generated_text": "玉", "function_call": null, "details": null, "total_time_taken": null, "prompt_tokens": 0, "generated_tokens": 0, "total_tokens": 0, "finish_reason": null}
data: {"generated_text": "山", "function_call": null, "details": null, "total_time_taken": "0.25 sec", "prompt_tokens": 7, "generated_tokens": 2, "total_tokens": 9, "finish_reason": "length"}
注意事項
- 每筆 token 並不一定能解碼成合適的文字，如果遇到該種情況，該筆 generated_text 欄位會顯示空字串，該 token 會結合下一筆資料再來解碼，直接能呈現為止。
- 本案例採用 sse-starlette，在 SSE 過程中約 15 秒就會收到 ping event，目前在程式中如果連線大於該時間就會收到以下資訊 (非 JSON 格式)，在資料處理時需特別注意，下列 Python 範例已經包含此資料處理。
event: ping
data: 2023-09-26 04:25:08.978531
Python 範例
import json
import requests
MODEL_NAME = "ffm-mixtral-8x7b-32k-instruct"
API_KEY = "{API_KEY}"
API_URL = "{API_URL}"
API_HOST = "afs-inference"
# parameters
max_new_tokens = 2
temperature = 0.5
top_k = 50
top_p = 1.0
frequence_penalty = 1.0
def generate(prompt):
headers = {
"content-type": "application/json",
"X-API-Key": API_KEY,
"X-API-Host": API_HOST}
data = {
"model": MODEL_NAME,
"inputs": prompt,
"parameters": {
"max_new_tokens": max_new_tokens,
"temperature": temperature,
"top_k": top_k,
"top_p": top_p,
"frequence_penalty": frequence_penalty
},
"stream": True
}
messages = []
result = ""
try:
response = requests.post(API_URL + "/models/generate", json=data, headers=headers, stream=True)
if response.status_code == 200:
for chunk in response.iter_lines():
chunk = chunk.decode('utf-8')
if chunk == "":
continue
try:
record = json.loads(chunk[5:], strict=False)
if "status_code" in record:
print("{:d}, {}".format(record["status_code"], record["error"]))
break
elif record["total_time_taken"] is not None or ("finish_reason" in record and record["finish_reason"] is not None) :
message = record["generated_text"]
messages.append(message)
print(">>> " + message)
result = ''.join(messages)
break
elif record["generated_text"] is not None:
message = record["generated_text"]
messages.append(message)
print(">>> " + message)
else:
print("error")
break
except:
pass
except:
print("error")
return result.strip("\n")
result = generate("台灣最高峰是")
print(result)
輸出：
>>> 玉
>>> 山
玉山
LangChain 使用方式
Custom LLM Model Wrapper
from typing import Any, Dict, List, Mapping, Optional, Tuple
from langchain.llms.base import BaseLLM
import requests
from langchain.callbacks.manager import CallbackManagerForLLMRun
from langchain.schema.language_model import BaseLanguageModel
from langchain.schema import Generation, LLMResult
from pydantic import Field
import json
import os
class _FormosaFoundationCommon(BaseLanguageModel):
base_url: str = "http://localhost:12345"
"""Base url the model is hosted under."""
model: str = "ffm-mixtral-8x7b-32k-instruct"
"""Model name to use."""
temperature: Optional[float]
"""The temperature of the model. Increasing the temperature will
make the model answer more creatively."""
stop: Optional[List[str]]
"""Sets the stop tokens to use."""
top_k: int = 50
"""Reduces the probability of generating nonsense. A higher value (e.g. 100)
will give more diverse answers, while a lower value (e.g. 10)
will be more conservative. (Default: 50)"""
top_p: float = 1
"""Works together with top-k. A higher value (e.g., 0.95) will lead
to more diverse text, while a lower value (e.g., 0.5) will
generate more focused and conservative text. (Default: 1)"""
max_new_tokens: int = 350
"""The maximum number of tokens to generate in the completion.
-1 returns as many tokens as possible given the prompt and
the models maximal context size."""
frequence_penalty: float = 1
"""Penalizes repeated tokens according to frequency."""
model_kwargs: Dict[str, Any] = Field(default_factory=dict)
"""Holds any model parameters valid for `create` call not explicitly specified."""
ffm_api_key: Optional[str] = None
@property
def _default_params(self) -> Dict[str, Any]:
"""Get the default parameters for calling FFM API."""
normal_params = {
"temperature": self.temperature,
"max_new_tokens": self.max_new_tokens,
"top_p": self.top_p,
"frequence_penalty": self.frequence_penalty,
"top_k": self.top_k,
}
return {**normal_params, **self.model_kwargs}
def _call(
self,
prompt,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> str:
if self.stop is not None and stop is not None:
raise ValueError("`stop` found in both the input and default params.")
elif self.stop is not None:
stop = self.stop
elif stop is None:
stop = []
params = {**self._default_params, "stop": stop, **kwargs}
parameter_payload = {"parameters": params, "inputs": prompt, "model": self.model}
# HTTP headers for authorization
headers = {
"X-API-KEY": self.ffm_api_key,
"Content-Type": "application/json",
"X-API-HOST": "afs-inference"
}
endpoint_url = f"{self.base_url}/models/generate"
# send request
try:
response = requests.post(
url=endpoint_url,
headers=headers,
data=json.dumps(parameter_payload, ensure_ascii=False).encode("utf8"),
stream=False,
)
response.encoding = "utf-8"
generated_text = response.json()
if response.status_code != 200:
detail = generated_text.get("detail")
raise ValueError(
f"FormosaFoundationModel endpoint_url: {endpoint_url}\n"
f"error raised with status code {response.status_code}\n"
f"Details: {detail}\n"
)
except requests.exceptions.RequestException as e: # This is the correct syntax
raise ValueError(f"FormosaFoundationModel error raised by inference endpoint: {e}\n")
if generated_text.get("detail") is not None:
detail = generated_text["detail"]
raise ValueError(
f"FormosaFoundationModel endpoint_url: {endpoint_url}\n"
f"error raised by inference API: {detail}\n"
)
if generated_text.get("generated_text") is None:
raise ValueError(
f"FormosaFoundationModel endpoint_url: {endpoint_url}\n"
f"Response format error: {generated_text}\n"
)
return generated_text
class FormosaFoundationModel(BaseLLM, _FormosaFoundationCommon):
"""Formosa Foundation Model
Example:
.. code-block:: python
ffm = FormosaFoundationModel(model_name="llama2-7b-chat-meta")
"""
@property
def _llm_type(self) -> str:
return "FormosaFoundationModel"
@property
def _identifying_params(self) -> Mapping[str, Any]:
"""Get the identifying parameters."""
return {
**{
"model": self.model,
"base_url": self.base_url
},
**self._default_params
}
def _generate(
self,
prompts: List[str],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> LLMResult:
"""Call out to FormosaFoundationModel's generate endpoint.
Args:
prompt: The prompt to pass into the model.
stop: Optional list of stop words to use when generating.
Returns:
The string generated by the model.
Example:
.. code-block:: python
response = FormosaFoundationModel("Tell me a joke.")
"""
generations = []
token_usage = 0
for prompt in prompts:
final_chunk = super()._call(
prompt,
stop=stop,
**kwargs,
)
generations.append(
[
Generation(
text = final_chunk["generated_text"],
generation_info=dict(
finish_reason = final_chunk["finish_reason"]
)
)
]
)
token_usage += final_chunk["generated_tokens"]
llm_output = {"token_usage": token_usage, "model": self.model}
return LLMResult(generations=generations, llm_output=llm_output)
- 完成以上封裝後，就可以在 LangChain 中使用 FFM 大語言模型。
info
更多資訊，請參考 LangChain Custom LLM 文件。
MODEL_NAME = "ffm-mixtral-8x7b-32k-instruct"
API_KEY = "{API_KEY}"
API_URL = "{API_URL}"
ffm = FormosaFoundationModel(
base_url = API_URL,
max_new_tokens = 350,
temperature = 0.5,
top_k = 50,
top_p = 1.0,
frequence_penalty = 1.0,
ffm_api_key = API_KEY,
model = MODEL_NAME
)
print(ffm("請問台灣最高的山是？"))
輸出：
答案是：玉山。
玉山，也被稱為玉山國家公園，位於台灣南部，是該國最高的山，海拔3952米（12966英尺）。它是台灣阿里山山脈的一部分，以其崎嶇的地形、翠綠的森林和多種植物和動物而聞名。玉山是徒步旅行者和自然愛好者的熱門目的地，被認為是台灣最美麗和最具挑戰性的山之一。