Skip to content

Linear

SimpleLinearPipeline

Bases: BaseComponent

A simple pipeline for running a function with a prompt, a language model, and an optional post-processor.

Attributes:

Name Type Description
prompt BasePromptComponent

The prompt component used to generate the initial input.
llm Union[ChatLLM, LLM]

The language model component used to generate the output.
post_processor Union[BaseComponent, Callable[[IO_Type], IO_Type]]

An optional post-processor component or function.
Example Usage 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from kotaemon.llms import LCAzureChatOpenAI, BasePromptComponent

def identity(x):
    return x

llm = LCAzureChatOpenAI(
    openai_api_base="your openai api base",
    openai_api_key="your openai api key",
    openai_api_version="your openai api version",
    deployment_name="dummy-q2-gpt35",
    temperature=0,
    request_timeout=600,
)

pipeline = SimpleLinearPipeline(
    prompt=BasePromptComponent(template="what is {word} in Japanese ?"),
    llm=llm,
    post_processor=identity,
)
print(pipeline(word="lone"))
Source code in libs/kotaemon/kotaemon/llms/linear.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class SimpleLinearPipeline(BaseComponent):
    """
    A simple pipeline for running a function with a prompt, a language model, and an
        optional post-processor.

    Attributes:
        prompt (BasePromptComponent): The prompt component used to generate the initial
            input.
        llm (Union[ChatLLM, LLM]): The language model component used to generate the
            output.
        post_processor (Union[BaseComponent, Callable[[IO_Type], IO_Type]]): An optional
            post-processor component or function.

    Example Usage:
        ```python
        from kotaemon.llms import LCAzureChatOpenAI, BasePromptComponent

        def identity(x):
            return x

        llm = LCAzureChatOpenAI(
            openai_api_base="your openai api base",
            openai_api_key="your openai api key",
            openai_api_version="your openai api version",
            deployment_name="dummy-q2-gpt35",
            temperature=0,
            request_timeout=600,
        )

        pipeline = SimpleLinearPipeline(
            prompt=BasePromptComponent(template="what is {word} in Japanese ?"),
            llm=llm,
            post_processor=identity,
        )
        print(pipeline(word="lone"))
        ```
    """

    prompt: BasePromptComponent
    llm: Union[ChatLLM, LLM]
    post_processor: Union[BaseComponent, Callable[[IO_Type], IO_Type]]

    def run(
        self,
        *,
        llm_kwargs: Optional[dict] = {},
        post_processor_kwargs: Optional[dict] = {},
        **prompt_kwargs,
    ):
        """
        Run the function with the given arguments and return the final output as a
            Document object.

        Args:
            llm_kwargs (dict): Keyword arguments for the llm call.
            post_processor_kwargs (dict): Keyword arguments for the post_processor.
            **prompt_kwargs: Keyword arguments for populating the prompt.

        Returns:
            Document: The final output of the function as a Document object.
        """
        prompt = self.prompt(**prompt_kwargs)
        llm_output = self.llm(prompt.text, **llm_kwargs)
        if self.post_processor is not None:
            final_output = self.post_processor(llm_output, **post_processor_kwargs)[0]
        else:
            final_output = llm_output

        return Document(final_output)

run 

1
2
3
4
5
6
run(
    *,
    llm_kwargs={},
    post_processor_kwargs={},
    **prompt_kwargs
)

Run the function with the given arguments and return the final output as a Document object.

Parameters:

Name Type Description Default
llm_kwargs dict

Keyword arguments for the llm call.

 {}
post_processor_kwargs dict

Keyword arguments for the post_processor.

 {}
**prompt_kwargs

Keyword arguments for populating the prompt.

 {}

Returns:

Name Type Description
Document

The final output of the function as a Document object.
Source code in libs/kotaemon/kotaemon/llms/linear.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def run(
    self,
    *,
    llm_kwargs: Optional[dict] = {},
    post_processor_kwargs: Optional[dict] = {},
    **prompt_kwargs,
):
    """
    Run the function with the given arguments and return the final output as a
        Document object.

    Args:
        llm_kwargs (dict): Keyword arguments for the llm call.
        post_processor_kwargs (dict): Keyword arguments for the post_processor.
        **prompt_kwargs: Keyword arguments for populating the prompt.

    Returns:
        Document: The final output of the function as a Document object.
    """
    prompt = self.prompt(**prompt_kwargs)
    llm_output = self.llm(prompt.text, **llm_kwargs)
    if self.post_processor is not None:
        final_output = self.post_processor(llm_output, **post_processor_kwargs)[0]
    else:
        final_output = llm_output

    return Document(final_output)

GatedLinearPipeline

Bases: SimpleLinearPipeline

A pipeline that extends the SimpleLinearPipeline class and adds a condition attribute.

Attributes:

Name Type Description
condition Callable[[IO_Type], Any]

A callable function that represents the condition.
Usage
Example Usage
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from kotaemon.llms import LCAzureChatOpenAI, BasePromptComponent
from kotaemon.parsers import RegexExtractor

def identity(x):
    return x

llm = LCAzureChatOpenAI(
    openai_api_base="your openai api base",
    openai_api_key="your openai api key",
    openai_api_version="your openai api version",
    deployment_name="dummy-q2-gpt35",
    temperature=0,
    request_timeout=600,
)

pipeline = GatedLinearPipeline(
    prompt=BasePromptComponent(template="what is {word} in Japanese ?"),
    condition=RegexExtractor(pattern="some pattern"),
    llm=llm,
    post_processor=identity,
)
print(pipeline(condition_text="some pattern", word="lone"))
print(pipeline(condition_text="other pattern", word="lone"))
Source code in libs/kotaemon/kotaemon/llms/linear.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class GatedLinearPipeline(SimpleLinearPipeline):
    """
    A pipeline that extends the SimpleLinearPipeline class and adds a condition
        attribute.

    Attributes:
        condition (Callable[[IO_Type], Any]): A callable function that represents the
            condition.

    Usage:
        ```{.py3 title="Example Usage"}
        from kotaemon.llms import LCAzureChatOpenAI, BasePromptComponent
        from kotaemon.parsers import RegexExtractor

        def identity(x):
            return x

        llm = LCAzureChatOpenAI(
            openai_api_base="your openai api base",
            openai_api_key="your openai api key",
            openai_api_version="your openai api version",
            deployment_name="dummy-q2-gpt35",
            temperature=0,
            request_timeout=600,
        )

        pipeline = GatedLinearPipeline(
            prompt=BasePromptComponent(template="what is {word} in Japanese ?"),
            condition=RegexExtractor(pattern="some pattern"),
            llm=llm,
            post_processor=identity,
        )
        print(pipeline(condition_text="some pattern", word="lone"))
        print(pipeline(condition_text="other pattern", word="lone"))
        ```
    """

    condition: Callable[[IO_Type], Any]

    def run(
        self,
        *,
        condition_text: Optional[str] = None,
        llm_kwargs: Optional[dict] = {},
        post_processor_kwargs: Optional[dict] = {},
        **prompt_kwargs,
    ) -> Document:
        """
        Run the pipeline with the given arguments and return the final output as a
            Document object.

        Args:
            condition_text (str): The condition text to evaluate. Default to None.
            llm_kwargs (dict): Additional keyword arguments for the language model call.
            post_processor_kwargs (dict): Additional keyword arguments for the
                post-processor.
            **prompt_kwargs: Keyword arguments for populating the prompt.

        Returns:
            Document: The final output of the pipeline as a Document object.

        Raises:
            ValueError: If condition_text is None
        """
        if condition_text is None:
            raise ValueError("`condition_text` must be provided")

        if self.condition(condition_text)[0]:
            return super().run(
                llm_kwargs=llm_kwargs,
                post_processor_kwargs=post_processor_kwargs,
                **prompt_kwargs,
            )

        return Document(None)

run 

1
2
3
4
5
6
7
run(
    *,
    condition_text=None,
    llm_kwargs={},
    post_processor_kwargs={},
    **prompt_kwargs
)

Run the pipeline with the given arguments and return the final output as a Document object.

Parameters:

Name Type Description Default
condition_text str

The condition text to evaluate. Default to None.

 None
llm_kwargs dict

Additional keyword arguments for the language model call.

 {}
post_processor_kwargs dict

Additional keyword arguments for the post-processor.

 {}
**prompt_kwargs

Keyword arguments for populating the prompt.

 {}

Returns:

Name Type Description
Document Document

The final output of the pipeline as a Document object.

Raises:

Type Description
ValueError

If condition_text is None
Source code in libs/kotaemon/kotaemon/llms/linear.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def run(
    self,
    *,
    condition_text: Optional[str] = None,
    llm_kwargs: Optional[dict] = {},
    post_processor_kwargs: Optional[dict] = {},
    **prompt_kwargs,
) -> Document:
    """
    Run the pipeline with the given arguments and return the final output as a
        Document object.

    Args:
        condition_text (str): The condition text to evaluate. Default to None.
        llm_kwargs (dict): Additional keyword arguments for the language model call.
        post_processor_kwargs (dict): Additional keyword arguments for the
            post-processor.
        **prompt_kwargs: Keyword arguments for populating the prompt.

    Returns:
        Document: The final output of the pipeline as a Document object.

    Raises:
        ValueError: If condition_text is None
    """
    if condition_text is None:
        raise ValueError("`condition_text` must be provided")

    if self.condition(condition_text)[0]:
        return super().run(
            llm_kwargs=llm_kwargs,
            post_processor_kwargs=post_processor_kwargs,
            **prompt_kwargs,
        )

    return Document(None)