FunctionModel

FunctionModel integrates your prompts and LLMs, and makes it available as a general function in your code. It can be used in 3 ways:

Fetch prompts from the cloud in OpenAI messages format
Run the FunctionModel locally (using your local API key), and get the result in OpenAI output format
Run the FunctionModel locally, then obtain the parsed outputs (with automatic type conversion) as a dictionary

We use LiteLLM (opens in a new tab) to integrate LLMs, but currently support only cloud based LLM providers. We will integrate more models, and even allow you to use your own models in the near future.

All of FunctionModel's methods except for get_prompts return either an LLMResponse or LLMStreamResponse object.

Methods

get_config fetches FunctionModelConfig from a specific FunctionModel, includes prompts and version_detail.
run, stream fetches published prompts, runs LLM API call locally, logs data to the cloud, and returns the result.
run_and_parse, stream_and_parse fetches published prompts, runs LLM API call locally, parses the result, logs data, then returns the parsed result.
Asynchronous methods like arun, astream, arun_and_parse, astream_and_parse are also available.
get_prompts fetches published prompts from a specific FunctionModel. (deprecated)

Error Handling

All methods of FunctionModel do not raise error.

Instead, error and error_log attributes will be set if error occurs.

If you use LLM API call with error handling, your previous code will be like this.

client = OpenAI()
max_retry = 3
trial = 0
while trial < max_retry:
	try:
		res = client.chat.completion.create(**kwargs)
		parsed_outputs = parser(res)
		break
	except Exception as e:
		print(e)
		trial += 1
print(parsed_outputs)

If you use FunctionModel, you can handle error like this.

max_retry = 3
trial = 0
while trial < max_retry:
	res = FunctionModel("example").run_and_parse(kwargs)
	if res.error:
		print(res.error_log)
		trial += 1
		continue
	break
print(res.parsed_outputs)

get_config

get_config() is a method for fetching FunctionModelConfig from a specific FunctionModel.

config: FunctionModelConfig = FunctionModel("say_hello").get_config()
print(config.prompts)
# [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Say hello"}]
print(config.version_detail)
# {"model" : "gpt-3.5-turbo", "uuid": "version_uuid", "parsing_type" : "colon", "output_keys" : ["response"]}

run

run() is a basic method for LLM calls.

You can use run() with inputs as Dictionary.

res = FunctionModel("extract_keyword").run({"text": "I like apple."})
print(res.raw_output)
# apple

You can also use arun() for asynchronous usage.

res = await FunctionModel("extract_keyword").arun({"text": "I like apple."})
print(res.raw_output)
# apple

stream

stream() is a method for streaming LLM calls. It is same with OpenAI API call with stream=True.

res = FunctionModel("start_conversation").stream()
for chunk in res:
	print(chunk.raw_output)
	# Hi
	# ,
	# how
	# are
	# you
	# ?

You can also use astream() for asynchronous call.

res = await FunctionModel("start_conversation").astream()
async for chunk in res:
	print(chunk.raw_output)
	# Hi
	# ,
	# how
	# are
	# you
	# ?

parse

FunctionModel can have parsing_type and output_keys attributes.

If you set parsing_type and output_keys when you engineer your FunctionModel, output keys can be parsed automatically.

Response of functionmodel will be parsed automatically and saved in parsed_outputs attribute as Dict[str, str].

run_and_parse

res = FunctionModel("get_keyword_and_score").run_and_parse({"text": text})
print(res.parsed_outputs)
# {'keyword': 'apple', 'score': 0.9}

stream_and_parse

res = FunctionModel("get_keyword_and_score").stream_and_parse({"text": text})
for chunk in res:
	print(chunk.parsed_outputs)
	# {'keyword': 'apple'}
	# {'score': 0}
	# {'score': .}
	# {'score': 9}

Response Format

All of FunctionModel's methods except for get_config return either an LLMResponse or LLMStreamResponse object.

LLMResponse

Here is the exact output dictionary and types of LLMResponse.

{
	'api_response' : {
		'choices': [
			{
				'finish_reason': Optional[str],
				'index': int,
				'message': {
					'role': Optional[str],
					'content': Optional[str],
					'function_call' : Optional[FunctionCall],
					'tool_calls' : Optional[ChatCompletionToolCall]
				}
			}
		],
		'created': str,
		'model': str,
		'usage': {
			'prompt_tokens': int,
			'completion_tokens': int,
			'total_tokens': int
		},
		'_response_ms' : float
	} or None,
	'raw_output' : Optional[str],
	'parsed_outputs' : Optional[Dict[str, Any]],
	'function_call' : Optional[FunctionCall],
	'tool_calls' : Optional[ChatCompletionToolCall]
	'error' : Optional[bool],
	'error_log' : Optional[str]
}

The format of LLMResponse.api_response is equal to the format of OpenAI API response.

For example, the code blocks below will return the same output.

from openai import OpenAI
from promptmodel import FunctionModel
client = OpenAI()
 
openai_response = client.chat.completion.create(
	model="gpt-3.5-turbo",
	messages=[{"role" : "system", "content" : ""}, {"role" : "user", "content" : "Say Hello"}]
)
print(openai_response['choices'][0]['messages']['content'])
# Hello, my name is GPT-3.
 
functionmodel_response = FunctionModel("say_hello").run()
print(functionmodel_response.api_response['choices'][0]['messages']['content'])
# Hello, my name is GPT-3.

You can also use LLMResponse.raw_output to get the content from the LLM call's response.

functionmodel_response = FunctionModel("say_hello").run()
print(functionmodel_response.raw_output)
# Hello, my name is GPT-3.

When you use methods such as run_and_parse, the parsed results can be found in LLMResponse.parsed_outputs.

functionmodel_response = FunctionModel("say_hello_COT").run_and_parse()
print(functionmodel_response.parsed_outputs)
# {"think" : "I will say Hello", "response" : "Hello"}

LLMStreamResponse

Streaming methods like stream() or stream_and_parse() return LLMStreamResponse.

Here is the exact json output and type of LLMStreamResponse.

{
	'api_response' : {
		'choices': [
			{
				'finish_reason': Optional[str],
				'index': int,
				'delta': {
					'role': Optional[str],
					'content': Optional[str],
					'function_call' : Optional[ChoiceDeltaFunctionCall],
					'tool_calls' : Optional[ChoiceDeltaToolCall]
				},
				'message': {
					'role': Optional[str],
					'content': Optional[str],
					'function_call' : Optional[FunctionCall],
					'tool_calls' : Optional[ChatCompletionToolCall]
				},
			}
		],
		'created': str,
		'model': str,
		'usage': {
			'prompt_tokens': int,
			'completion_tokens': int,
			'total_tokens': int
		} or None,
		'_response_ms' : float
	} or None,
	'raw_output' : Optional[str],
	'parsed_outputs' : Optional[Dict[str, str]],
	'function_call' : Optional[ChoiceDeltaFunctionCall],
	'tool_calls' : Optional[ChoiceDeltaToolCall],
	'error' : Optional[bool],
	'error_log' : Optional[str]
}

The format of LLMStreamResponse.api_response is equal to the format of OpenAI API response.

However, the last chunk of LLMStreamResponse.api_response.choices[0] has message instead of delta. Every delta are converted to message in the last chunk.

For example, the code blocks below will return the same output.

from openai import OpenAI
from promptmodel import FunctionModel
client = OpenAI()
 
openai_response = client.chat.completion.create(
	model="gpt-3.5-turbo",
	messages=[{"role" : "system", "content" : ""}, {"role" : "user", "content" : "Say Hello"}],
	stream=True
)
for chunk in openai_response:
	print(chunk['choices'][0]['delta']['content'])
# Hello,
# my
# name
# is
# GPT-3.
 
functionmodel_response = FunctionModel("say_hello").stream()
for chunk in functionmodel_response:
	print(chunk.api_response['choices'][0]['delta']['content'])
# Hello,
# my
# name
# is
# GPT-3.

You can also use LLMStreamResponse.raw_output to get the content from the LLM call's response.

functionmodel_response = FunctionModel("say_hello").stream()
for chunk in functionmodel_response:
	print(chunk.raw_output)
# Hello,
# my
# name
# is
# GPT-3.

Using stream_and_parse will return the parsed results in LLMStreamResponse.parsed_outputs. The streamed LLM response will be parsed in real-time, streamed as Dictionary.

functionmodel_response = FunctionModel("say_hello_COT").stream_and_parse()
for chunk in functionmodel_response:
	print(chunk.parsed_outputs)
	# {"think" : "I"}
	# {"think" : "will"}
	# {"think" : "say"}
	# {"think" : "Hello."}
	# {"response" : "Hello"}

get_prompts (deprecated)

get_prompts() is a method for fetching published prompts from a specific FunctionModel.

prompts = FunctionModel("say_hello").get_prompts()
print(prompts)
# [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Say hello"}]

Python SDK ChatModel