Skip to content

LLMs

GatedBranchingPipeline

Bases: SimpleBranchingPipeline

A simple gated branching pipeline for executing multiple branches based on a condition.

This class extends the SimpleBranchingPipeline class and adds the ability to execute the branches until a branch returns a non-empty output based on a condition.

Attributes:

Name Type Description
branches List[BaseComponent]

The list of branches to be executed.
Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from kotaemon.llms import (
    LCAzureChatOpenAI,
    BasePromptComponent,
    GatedLinearPipeline,
)
from kotaemon.parsers import RegexExtractor

def identity(x):
    return x

pipeline = GatedBranchingPipeline()
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,
)

for i in range(3):
    pipeline.add_branch(
        GatedLinearPipeline(
            prompt=BasePromptComponent(template=f"what is {i} in Japanese ?"),
            condition=RegexExtractor(pattern=f"{i}"),
            llm=llm,
            post_processor=identity,
        )
    )
print(pipeline(condition_text="1"))
print(pipeline(condition_text="2"))
Source code in libs\kotaemon\kotaemon\llms\branching.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
class GatedBranchingPipeline(SimpleBranchingPipeline):
    """
    A simple gated branching pipeline for executing multiple branches based on a
        condition.

    This class extends the SimpleBranchingPipeline class and adds the ability to execute
        the branches until a branch returns a non-empty output based on a condition.

    Attributes:
        branches (List[BaseComponent]): The list of branches to be executed.

    Example:
        ```python
        from kotaemon.llms import (
            LCAzureChatOpenAI,
            BasePromptComponent,
            GatedLinearPipeline,
        )
        from kotaemon.parsers import RegexExtractor

        def identity(x):
            return x

        pipeline = GatedBranchingPipeline()
        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,
        )

        for i in range(3):
            pipeline.add_branch(
                GatedLinearPipeline(
                    prompt=BasePromptComponent(template=f"what is {i} in Japanese ?"),
                    condition=RegexExtractor(pattern=f"{i}"),
                    llm=llm,
                    post_processor=identity,
                )
            )
        print(pipeline(condition_text="1"))
        print(pipeline(condition_text="2"))
        ```
    """

    def run(self, *, condition_text: Optional[str] = None, **prompt_kwargs):
        """
        Execute the pipeline by running each branch and return the output of the first
            branch that returns a non-empty output based on the provided condition.

        Args:
            condition_text (str): The condition text to evaluate for each branch.
                Default to None.
            **prompt_kwargs: Keyword arguments for the branches.

        Returns:
            Union[OutputType, None]: The output of the first branch that satisfies the
            condition, or None if no branch satisfies the condition.

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

        for i, branch in enumerate(self.branches):
            self._prepare_child(branch, name=f"branch-{i}")
            output = branch(condition_text=condition_text, **prompt_kwargs)
            if output:
                return output

        return Document(None)

run 

1
run(*, condition_text=None, **prompt_kwargs)

Execute the pipeline by running each branch and return the output of the first branch that returns a non-empty output based on the provided condition.

Parameters:

Name Type Description Default
condition_text str

The condition text to evaluate for each branch. Default to None.

 None
**prompt_kwargs

Keyword arguments for the branches.

 {}

Returns:

Type Description

Union[OutputType, None]: The output of the first branch that satisfies the

condition, or None if no branch satisfies the condition.

Raises:

Type Description
ValueError

If condition_text is None
Source code in libs\kotaemon\kotaemon\llms\branching.py 
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
def run(self, *, condition_text: Optional[str] = None, **prompt_kwargs):
    """
    Execute the pipeline by running each branch and return the output of the first
        branch that returns a non-empty output based on the provided condition.

    Args:
        condition_text (str): The condition text to evaluate for each branch.
            Default to None.
        **prompt_kwargs: Keyword arguments for the branches.

    Returns:
        Union[OutputType, None]: The output of the first branch that satisfies the
        condition, or None if no branch satisfies the condition.

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

    for i, branch in enumerate(self.branches):
        self._prepare_child(branch, name=f"branch-{i}")
        output = branch(condition_text=condition_text, **prompt_kwargs)
        if output:
            return output

    return Document(None)

SimpleBranchingPipeline

Bases: BaseComponent

A simple branching pipeline for executing multiple branches.

Attributes:

Name Type Description
branches List[BaseComponent]

The list of branches to be executed.
Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from kotaemon.llms import (
    LCAzureChatOpenAI,
    BasePromptComponent,
    GatedLinearPipeline,
)
from kotaemon.parsers import RegexExtractor

def identity(x):
    return x

pipeline = SimpleBranchingPipeline()
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,
)

for i in range(3):
    pipeline.add_branch(
        GatedLinearPipeline(
            prompt=BasePromptComponent(template=f"what is {i} in Japanese ?"),
            condition=RegexExtractor(pattern=f"{i}"),
            llm=llm,
            post_processor=identity,
        )
    )
print(pipeline(condition_text="1"))
print(pipeline(condition_text="2"))
print(pipeline(condition_text="12"))
Source code in libs\kotaemon\kotaemon\llms\branching.py 
 8
 9
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 SimpleBranchingPipeline(BaseComponent):
    """
    A simple branching pipeline for executing multiple branches.

    Attributes:
        branches (List[BaseComponent]): The list of branches to be executed.

    Example:
        ```python
        from kotaemon.llms import (
            LCAzureChatOpenAI,
            BasePromptComponent,
            GatedLinearPipeline,
        )
        from kotaemon.parsers import RegexExtractor

        def identity(x):
            return x

        pipeline = SimpleBranchingPipeline()
        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,
        )

        for i in range(3):
            pipeline.add_branch(
                GatedLinearPipeline(
                    prompt=BasePromptComponent(template=f"what is {i} in Japanese ?"),
                    condition=RegexExtractor(pattern=f"{i}"),
                    llm=llm,
                    post_processor=identity,
                )
            )
        print(pipeline(condition_text="1"))
        print(pipeline(condition_text="2"))
        print(pipeline(condition_text="12"))
        ```
    """

    branches: List[BaseComponent] = Param(default_callback=lambda *_: [])

    def add_branch(self, component: BaseComponent):
        """
        Add a new branch to the pipeline.

        Args:
            component (BaseComponent): The branch component to be added.
        """
        self.branches.append(component)

    def run(self, **prompt_kwargs):
        """
        Execute the pipeline by running each branch and return the outputs as a list.

        Args:
            **prompt_kwargs: Keyword arguments for the branches.

        Returns:
            List: The outputs of each branch as a list.
        """
        output = []
        for i, branch in enumerate(self.branches):
            self._prepare_child(branch, name=f"branch-{i}")
            output.append(branch(**prompt_kwargs))

        return output

add_branch 

1
add_branch(component)

Add a new branch to the pipeline.

Parameters:

Name Type Description Default
component BaseComponent

The branch component to be added.

 required
Source code in libs\kotaemon\kotaemon\llms\branching.py 
54
55
56
57
58
59
60
61
def add_branch(self, component: BaseComponent):
    """
    Add a new branch to the pipeline.

    Args:
        component (BaseComponent): The branch component to be added.
    """
    self.branches.append(component)

run 

1
run(**prompt_kwargs)

Execute the pipeline by running each branch and return the outputs as a list.

Parameters:

Name Type Description Default
**prompt_kwargs

Keyword arguments for the branches.

 {}

Returns:

Name Type Description
List

The outputs of each branch as a list.
Source code in libs\kotaemon\kotaemon\llms\branching.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def run(self, **prompt_kwargs):
    """
    Execute the pipeline by running each branch and return the outputs as a list.

    Args:
        **prompt_kwargs: Keyword arguments for the branches.

    Returns:
        List: The outputs of each branch as a list.
    """
    output = []
    for i, branch in enumerate(self.branches):
        self._prepare_child(branch, name=f"branch-{i}")
        output.append(branch(**prompt_kwargs))

    return output

AzureChatOpenAI

Bases: BaseChatOpenAI

OpenAI chat model provided by Microsoft Azure

Source code in libs\kotaemon\kotaemon\llms\chats\openai.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
class AzureChatOpenAI(BaseChatOpenAI):
    """OpenAI chat model provided by Microsoft Azure"""

    azure_endpoint: str = Param(
        help=(
            "HTTPS endpoint for the Azure OpenAI model. The azure_endpoint, "
            "azure_deployment, and api_version parameters are used to construct "
            "the full URL for the Azure OpenAI model."
        ),
        required=True,
    )
    azure_deployment: str = Param(help="Azure deployment name", required=True)
    api_version: str = Param(help="Azure model version", required=True)
    azure_ad_token: Optional[str] = Param(None, help="Azure AD token")
    azure_ad_token_provider: Optional[str] = Param(None, help="Azure AD token provider")

    @Param.auto(depends_on=["azure_ad_token_provider"])
    def azure_ad_token_provider_(self):
        if isinstance(self.azure_ad_token_provider, str):
            return import_dotted_string(self.azure_ad_token_provider, safe=False)

    def prepare_client(self, async_version: bool = False):
        """Get the OpenAI client

        Args:
            async_version (bool): Whether to get the async version of the client
        """
        params = {
            "azure_endpoint": self.azure_endpoint,
            "api_version": self.api_version,
            "api_key": self.api_key,
            "azure_ad_token": self.azure_ad_token,
            "azure_ad_token_provider": self.azure_ad_token_provider_,
            "timeout": self.timeout,
            "max_retries": self.max_retries_,
        }
        if async_version:
            from openai import AsyncAzureOpenAI

            return AsyncAzureOpenAI(**params)

        from openai import AzureOpenAI

        return AzureOpenAI(**params)

    def openai_response(self, client, **kwargs):
        """Get the openai response"""
        params_ = {
            "model": self.azure_deployment,
            "temperature": self.temperature,
            "max_tokens": self.max_tokens,
            "n": self.n,
            "stop": self.stop,
            "frequency_penalty": self.frequency_penalty,
            "presence_penalty": self.presence_penalty,
            "tool_choice": self.tool_choice,
            "tools": self.tools,
            "logprobs": self.logprobs,
            "logit_bias": self.logit_bias,
            "top_logprobs": self.top_logprobs,
            "top_p": self.top_p,
        }
        params = {k: v for k, v in params_.items() if v is not None}
        params.update(kwargs)

        return client.chat.completions.create(**params)

prepare_client 

1
prepare_client(async_version=False)

Get the OpenAI client

Parameters:

Name Type Description Default
async_version bool

Whether to get the async version of the client

 False
Source code in libs\kotaemon\kotaemon\llms\chats\openai.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def prepare_client(self, async_version: bool = False):
    """Get the OpenAI client

    Args:
        async_version (bool): Whether to get the async version of the client
    """
    params = {
        "azure_endpoint": self.azure_endpoint,
        "api_version": self.api_version,
        "api_key": self.api_key,
        "azure_ad_token": self.azure_ad_token,
        "azure_ad_token_provider": self.azure_ad_token_provider_,
        "timeout": self.timeout,
        "max_retries": self.max_retries_,
    }
    if async_version:
        from openai import AsyncAzureOpenAI

        return AsyncAzureOpenAI(**params)

    from openai import AzureOpenAI

    return AzureOpenAI(**params)

openai_response 

1
openai_response(client, **kwargs)

Get the openai response

Source code in libs\kotaemon\kotaemon\llms\chats\openai.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
def openai_response(self, client, **kwargs):
    """Get the openai response"""
    params_ = {
        "model": self.azure_deployment,
        "temperature": self.temperature,
        "max_tokens": self.max_tokens,
        "n": self.n,
        "stop": self.stop,
        "frequency_penalty": self.frequency_penalty,
        "presence_penalty": self.presence_penalty,
        "tool_choice": self.tool_choice,
        "tools": self.tools,
        "logprobs": self.logprobs,
        "logit_bias": self.logit_bias,
        "top_logprobs": self.top_logprobs,
        "top_p": self.top_p,
    }
    params = {k: v for k, v in params_.items() if v is not None}
    params.update(kwargs)

    return client.chat.completions.create(**params)

ChatOpenAI

Bases: BaseChatOpenAI

OpenAI chat model

Source code in libs\kotaemon\kotaemon\llms\chats\openai.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
class ChatOpenAI(BaseChatOpenAI):
    """OpenAI chat model"""

    base_url: Optional[str] = Param(None, help="OpenAI base URL")
    organization: Optional[str] = Param(None, help="OpenAI organization")
    model: str = Param(help="OpenAI model", required=True)

    def prepare_client(self, async_version: bool = False):
        """Get the OpenAI client

        Args:
            async_version (bool): Whether to get the async version of the client
        """
        params = {
            "api_key": self.api_key,
            "organization": self.organization,
            "base_url": self.base_url,
            "timeout": self.timeout,
            "max_retries": self.max_retries_,
        }
        if async_version:
            from openai import AsyncOpenAI

            return AsyncOpenAI(**params)

        from openai import OpenAI

        return OpenAI(**params)

    def openai_response(self, client, **kwargs):
        """Get the openai response"""
        params_ = {
            "model": self.model,
            "temperature": self.temperature,
            "max_tokens": self.max_tokens,
            "n": self.n,
            "stop": self.stop,
            "frequency_penalty": self.frequency_penalty,
            "presence_penalty": self.presence_penalty,
            "tool_choice": self.tool_choice,
            "tools": self.tools,
            "logprobs": self.logprobs,
            "logit_bias": self.logit_bias,
            "top_logprobs": self.top_logprobs,
            "top_p": self.top_p,
        }
        params = {k: v for k, v in params_.items() if v is not None}
        params.update(kwargs)

        return client.chat.completions.create(**params)

prepare_client 

1
prepare_client(async_version=False)

Get the OpenAI client

Parameters:

Name Type Description Default
async_version bool

Whether to get the async version of the client

 False
Source code in libs\kotaemon\kotaemon\llms\chats\openai.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def prepare_client(self, async_version: bool = False):
    """Get the OpenAI client

    Args:
        async_version (bool): Whether to get the async version of the client
    """
    params = {
        "api_key": self.api_key,
        "organization": self.organization,
        "base_url": self.base_url,
        "timeout": self.timeout,
        "max_retries": self.max_retries_,
    }
    if async_version:
        from openai import AsyncOpenAI

        return AsyncOpenAI(**params)

    from openai import OpenAI

    return OpenAI(**params)

openai_response 

1
openai_response(client, **kwargs)

Get the openai response

Source code in libs\kotaemon\kotaemon\llms\chats\openai.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def openai_response(self, client, **kwargs):
    """Get the openai response"""
    params_ = {
        "model": self.model,
        "temperature": self.temperature,
        "max_tokens": self.max_tokens,
        "n": self.n,
        "stop": self.stop,
        "frequency_penalty": self.frequency_penalty,
        "presence_penalty": self.presence_penalty,
        "tool_choice": self.tool_choice,
        "tools": self.tools,
        "logprobs": self.logprobs,
        "logit_bias": self.logit_bias,
        "top_logprobs": self.top_logprobs,
        "top_p": self.top_p,
    }
    params = {k: v for k, v in params_.items() if v is not None}
    params.update(kwargs)

    return client.chat.completions.create(**params)

EndpointChatLLM

Bases: ChatLLM

A ChatLLM that uses an endpoint to generate responses. This expects an OpenAI API compatible endpoint.

Attributes:

Name Type Description
endpoint_url str

The url of a OpenAI API compatible endpoint.
Source code in libs\kotaemon\kotaemon\llms\chats\endpoint_based.py 
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
79
80
81
82
83
84
85
86
87
88
class EndpointChatLLM(ChatLLM):
    """
    A ChatLLM that uses an endpoint to generate responses. This expects an OpenAI API
    compatible endpoint.

    Attributes:
        endpoint_url (str): The url of a OpenAI API compatible endpoint.
    """

    endpoint_url: str = Param(
        help="URL of the OpenAI API compatible endpoint", required=True
    )

    def run(
        self, messages: str | BaseMessage | list[BaseMessage], **kwargs
    ) -> LLMInterface:
        """
        Generate response from messages
        Args:
            messages (str | BaseMessage | list[BaseMessage]): history of messages to
                generate response from
            **kwargs: additional arguments to pass to the OpenAI API
        Returns:
            LLMInterface: generated response
        """
        if isinstance(messages, str):
            input_ = [HumanMessage(content=messages)]
        elif isinstance(messages, BaseMessage):
            input_ = [messages]
        else:
            input_ = messages

        def decide_role(message: BaseMessage):
            if isinstance(message, SystemMessage):
                return "system"
            elif isinstance(message, AIMessage):
                return "assistant"
            else:
                return "user"

        request_json = {
            "messages": [{"content": m.text, "role": decide_role(m)} for m in input_]
        }

        response = requests.post(self.endpoint_url, json=request_json).json()

        content = ""
        candidates = []
        if response["choices"]:
            candidates = [
                each["message"]["content"]
                for each in response["choices"]
                if each["message"]["content"]
            ]
            content = candidates[0]

        return LLMInterface(
            content=content,
            candidates=candidates,
            completion_tokens=response["usage"]["completion_tokens"],
            total_tokens=response["usage"]["total_tokens"],
            prompt_tokens=response["usage"]["prompt_tokens"],
        )

    def invoke(
        self, messages: str | BaseMessage | list[BaseMessage], **kwargs
    ) -> LLMInterface:
        """Same as run"""
        return self.run(messages, **kwargs)

    async def ainvoke(
        self, messages: str | BaseMessage | list[BaseMessage], **kwargs
    ) -> LLMInterface:
        return self.invoke(messages, **kwargs)

run 

1
run(messages, **kwargs)

Generate response from messages Args: messages (str | BaseMessage | list[BaseMessage]): history of messages to generate response from **kwargs: additional arguments to pass to the OpenAI API Returns: LLMInterface: generated response

Source code in libs\kotaemon\kotaemon\llms\chats\endpoint_based.py 
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
def run(
    self, messages: str | BaseMessage | list[BaseMessage], **kwargs
) -> LLMInterface:
    """
    Generate response from messages
    Args:
        messages (str | BaseMessage | list[BaseMessage]): history of messages to
            generate response from
        **kwargs: additional arguments to pass to the OpenAI API
    Returns:
        LLMInterface: generated response
    """
    if isinstance(messages, str):
        input_ = [HumanMessage(content=messages)]
    elif isinstance(messages, BaseMessage):
        input_ = [messages]
    else:
        input_ = messages

    def decide_role(message: BaseMessage):
        if isinstance(message, SystemMessage):
            return "system"
        elif isinstance(message, AIMessage):
            return "assistant"
        else:
            return "user"

    request_json = {
        "messages": [{"content": m.text, "role": decide_role(m)} for m in input_]
    }

    response = requests.post(self.endpoint_url, json=request_json).json()

    content = ""
    candidates = []
    if response["choices"]:
        candidates = [
            each["message"]["content"]
            for each in response["choices"]
            if each["message"]["content"]
        ]
        content = candidates[0]

    return LLMInterface(
        content=content,
        candidates=candidates,
        completion_tokens=response["usage"]["completion_tokens"],
        total_tokens=response["usage"]["total_tokens"],
        prompt_tokens=response["usage"]["prompt_tokens"],
    )

invoke 

1
invoke(messages, **kwargs)

Same as run

Source code in libs\kotaemon\kotaemon\llms\chats\endpoint_based.py 
79
80
81
82
83
def invoke(
    self, messages: str | BaseMessage | list[BaseMessage], **kwargs
) -> LLMInterface:
    """Same as run"""
    return self.run(messages, **kwargs)

LlamaCppChat

Bases: ChatLLM

Wrapper around the llama-cpp-python's Llama model

Source code in libs\kotaemon\kotaemon\llms\chats\llamacpp.py 
 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
 79
 80
 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
156
class LlamaCppChat(ChatLLM):
    """Wrapper around the llama-cpp-python's Llama model"""

    model_path: Optional[str] = Param(
        help="Path to the model file. This is required to load the model.",
    )
    repo_id: Optional[str] = Param(
        help="Id of a repo on the HuggingFace Hub in the form of `user_name/repo_name`."
    )
    filename: Optional[str] = Param(
        help="A filename or glob pattern to match the model file in the repo."
    )
    chat_format: str = Param(
        help=(
            "Chat format to use. Please refer to llama_cpp.llama_chat_format for a "
            "list of supported formats. If blank, the chat format will be auto-"
            "inferred."
        ),
        required=True,
    )
    lora_base: Optional[str] = Param(None, help="Path to the base Lora model")
    n_ctx: Optional[int] = Param(512, help="Text context, 0 = from model")
    n_gpu_layers: Optional[int] = Param(
        0,
        help="Number of layers to offload to GPU. If -1, all layers are offloaded",
    )
    use_mmap: Optional[bool] = Param(
        True,
        help=(),
    )
    vocab_only: Optional[bool] = Param(
        False,
        help="If True, only the vocabulary is loaded. This is useful for debugging.",
    )

    _role_mapper: dict[str, str] = {
        "human": "user",
        "system": "system",
        "ai": "assistant",
    }

    @Param.auto()
    def client_object(self) -> "Llama":
        """Get the llama-cpp-python client object"""
        try:
            from llama_cpp import Llama
        except ImportError:
            raise ImportError(
                "llama-cpp-python is not installed. "
                "Please install it using `pip install llama-cpp-python`"
            )

        errors = []
        if not self.model_path and (not self.repo_id or not self.filename):
            errors.append(
                "- `model_path` or `repo_id` and `filename` are required to load the"
                " model"
            )

        if not self.chat_format:
            errors.append(
                "- `chat_format` is required to know how to format the chat messages. "
                "Please refer to llama_cpp.llama_chat_format for a list of supported "
                "formats."
            )
        if errors:
            raise ValueError("\n".join(errors))

        if self.model_path:
            return Llama(
                model_path=cast(str, self.model_path),
                chat_format=self.chat_format,
                lora_base=self.lora_base,
                n_ctx=self.n_ctx,
                n_gpu_layers=self.n_gpu_layers,
                use_mmap=self.use_mmap,
                vocab_only=self.vocab_only,
            )
        else:
            return Llama.from_pretrained(
                repo_id=self.repo_id,
                filename=self.filename,
                chat_format=self.chat_format,
                lora_base=self.lora_base,
                n_ctx=self.n_ctx,
                n_gpu_layers=self.n_gpu_layers,
                use_mmap=self.use_mmap,
                vocab_only=self.vocab_only,
            )

    def prepare_message(
        self, messages: str | BaseMessage | list[BaseMessage]
    ) -> list[dict]:
        input_: list[BaseMessage] = []

        if isinstance(messages, str):
            input_ = [HumanMessage(content=messages)]
        elif isinstance(messages, BaseMessage):
            input_ = [messages]
        else:
            input_ = messages

        output_ = [
            {"role": self._role_mapper[each.type], "content": each.content}
            for each in input_
        ]

        return output_

    def invoke(
        self, messages: str | BaseMessage | list[BaseMessage], **kwargs
    ) -> LLMInterface:

        pred: "CCCR" = self.client_object.create_chat_completion(
            messages=self.prepare_message(messages),
            stream=False,
        )

        return LLMInterface(
            content=pred["choices"][0]["message"]["content"] if pred["choices"] else "",
            candidates=[
                c["message"]["content"]
                for c in pred["choices"]
                if c["message"]["content"]
            ],
            completion_tokens=pred["usage"]["completion_tokens"],
            total_tokens=pred["usage"]["total_tokens"],
            prompt_tokens=pred["usage"]["prompt_tokens"],
        )

    def stream(
        self, messages: str | BaseMessage | list[BaseMessage], **kwargs
    ) -> Iterator[LLMInterface]:
        pred = self.client_object.create_chat_completion(
            messages=self.prepare_message(messages),
            stream=True,
        )
        for chunk in pred:
            if not chunk["choices"]:
                continue

            if "content" not in chunk["choices"][0]["delta"]:
                continue

            yield LLMInterface(content=chunk["choices"][0]["delta"]["content"])

client_object 

1
client_object()

Get the llama-cpp-python client object

Source code in libs\kotaemon\kotaemon\llms\chats\llamacpp.py 
 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
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@Param.auto()
def client_object(self) -> "Llama":
    """Get the llama-cpp-python client object"""
    try:
        from llama_cpp import Llama
    except ImportError:
        raise ImportError(
            "llama-cpp-python is not installed. "
            "Please install it using `pip install llama-cpp-python`"
        )

    errors = []
    if not self.model_path and (not self.repo_id or not self.filename):
        errors.append(
            "- `model_path` or `repo_id` and `filename` are required to load the"
            " model"
        )

    if not self.chat_format:
        errors.append(
            "- `chat_format` is required to know how to format the chat messages. "
            "Please refer to llama_cpp.llama_chat_format for a list of supported "
            "formats."
        )
    if errors:
        raise ValueError("\n".join(errors))

    if self.model_path:
        return Llama(
            model_path=cast(str, self.model_path),
            chat_format=self.chat_format,
            lora_base=self.lora_base,
            n_ctx=self.n_ctx,
            n_gpu_layers=self.n_gpu_layers,
            use_mmap=self.use_mmap,
            vocab_only=self.vocab_only,
        )
    else:
        return Llama.from_pretrained(
            repo_id=self.repo_id,
            filename=self.filename,
            chat_format=self.chat_format,
            lora_base=self.lora_base,
            n_ctx=self.n_ctx,
            n_gpu_layers=self.n_gpu_layers,
            use_mmap=self.use_mmap,
            vocab_only=self.vocab_only,
        )

AzureOpenAI

Bases: LCCompletionMixin, LLM

Wrapper around Langchain's AzureOpenAI class, focusing on key parameters

Source code in libs\kotaemon\kotaemon\llms\completions\langchain_based.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class AzureOpenAI(LCCompletionMixin, LLM):
    """Wrapper around Langchain's AzureOpenAI class, focusing on key parameters"""

    def __init__(
        self,
        azure_endpoint: Optional[str] = None,
        deployment_name: Optional[str] = None,
        openai_api_version: str = "",
        openai_api_key: Optional[str] = None,
        model_name: str = "text-davinci-003",
        temperature: float = 0.7,
        max_tokens: int = 256,
        top_p: float = 1,
        frequency_penalty: float = 0,
        n: int = 1,
        best_of: int = 1,
        request_timeout: Optional[float] = None,
        max_retries: int = 2,
        streaming: bool = False,
        **params,
    ):
        super().__init__(
            azure_endpoint=azure_endpoint,
            deployment_name=deployment_name,
            openai_api_version=openai_api_version,
            openai_api_key=openai_api_key,
            model_name=model_name,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=top_p,
            frequency_penalty=frequency_penalty,
            n=n,
            best_of=best_of,
            request_timeout=request_timeout,
            max_retries=max_retries,
            streaming=streaming,
            **params,
        )

    def _get_lc_class(self):
        try:
            from langchain_openai import AzureOpenAI
        except ImportError:
            from langchain.llms import AzureOpenAI

        return AzureOpenAI

LlamaCpp

Bases: LCCompletionMixin, LLM

Wrapper around Langchain's LlamaCpp class, focusing on key parameters

Source code in libs\kotaemon\kotaemon\llms\completions\langchain_based.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class LlamaCpp(LCCompletionMixin, LLM):
    """Wrapper around Langchain's LlamaCpp class, focusing on key parameters"""

    def __init__(
        self,
        model_path: str,
        lora_base: Optional[str] = None,
        n_ctx: int = 512,
        n_gpu_layers: Optional[int] = None,
        use_mmap: bool = True,
        **params,
    ):
        super().__init__(
            model_path=model_path,
            lora_base=lora_base,
            n_ctx=n_ctx,
            n_gpu_layers=n_gpu_layers,
            use_mmap=use_mmap,
            **params,
        )

    def _get_lc_class(self):
        try:
            from langchain_community.llms import LlamaCpp
        except ImportError:
            from langchain.llms import LlamaCpp

        return LlamaCpp

OpenAI

Bases: LCCompletionMixin, LLM

Wrapper around Langchain's OpenAI class, focusing on key parameters

Source code in libs\kotaemon\kotaemon\llms\completions\langchain_based.py 
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
class OpenAI(LCCompletionMixin, LLM):
    """Wrapper around Langchain's OpenAI class, focusing on key parameters"""

    def __init__(
        self,
        openai_api_key: Optional[str] = None,
        openai_api_base: Optional[str] = None,
        model_name: str = "text-davinci-003",
        temperature: float = 0.7,
        max_tokens: int = 256,
        top_p: float = 1,
        frequency_penalty: float = 0,
        n: int = 1,
        best_of: int = 1,
        request_timeout: Optional[float] = None,
        max_retries: int = 2,
        streaming: bool = False,
        **params,
    ):
        super().__init__(
            openai_api_key=openai_api_key,
            openai_api_base=openai_api_base,
            model_name=model_name,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=top_p,
            frequency_penalty=frequency_penalty,
            n=n,
            best_of=best_of,
            request_timeout=request_timeout,
            max_retries=max_retries,
            streaming=streaming,
            **params,
        )

    def _get_lc_class(self):
        try:
            from langchain_openai import OpenAI
        except ImportError:
            from langchain.llms import OpenAI

        return OpenAI

ManualSequentialChainOfThought

Bases: BaseComponent

Perform sequential chain-of-thought with manual pre-defined prompts

This method supports variable number of steps. Each step corresponds to a kotaemon.pipelines.cot.Thought. Please refer that section for Thought's detail. This section is about chaining thought together.

Usage:

Create and run a chain of thought without "+" operator:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
>>> from kotaemon.pipelines.cot import Thought, ManualSequentialChainOfThought
>>> llm = LCAzureChatOpenAI(...)
>>> thought1 = Thought(
>>>    prompt="Word {word} in {language} is ",
>>>    post_process=lambda string: {"translated": string},
>>> )
>>> thought2 = Thought(
>>>     prompt="Translate {translated} to Japanese",
>>>     post_process=lambda string: {"output": string},
>>> )
>>> thought = ManualSequentialChainOfThought(thoughts=[thought1, thought2], llm=llm)
>>> thought(word="hello", language="French")
{'word': 'hello',
 'language': 'French',
 'translated': '"Bonjour"',
 'output': 'こんにちは (Konnichiwa)'}

Create and run a chain of thought without "+" operator: Please refer the kotaemon.pipelines.cot.Thought section for examples.

This chain-of-thought optionally takes a termination check callback function. This function will be called after each thought is executed. It takes in a dictionary of all thought outputs so far, and it returns True or False. If True, the chain-of-thought will terminate. If unset, the default callback always returns False.

Source code in libs\kotaemon\kotaemon\llms\cot.py 
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class ManualSequentialChainOfThought(BaseComponent):
    """Perform sequential chain-of-thought with manual pre-defined prompts

    This method supports variable number of steps. Each step corresponds to a
    `kotaemon.pipelines.cot.Thought`. Please refer that section for
    Thought's detail. This section is about chaining thought together.

    _**Usage:**_

    **Create and run a chain of thought without "+" operator:**

    ```pycon
    >>> from kotaemon.pipelines.cot import Thought, ManualSequentialChainOfThought
    >>> llm = LCAzureChatOpenAI(...)
    >>> thought1 = Thought(
    >>>    prompt="Word {word} in {language} is ",
    >>>    post_process=lambda string: {"translated": string},
    >>> )
    >>> thought2 = Thought(
    >>>     prompt="Translate {translated} to Japanese",
    >>>     post_process=lambda string: {"output": string},
    >>> )
    >>> thought = ManualSequentialChainOfThought(thoughts=[thought1, thought2], llm=llm)
    >>> thought(word="hello", language="French")
    {'word': 'hello',
     'language': 'French',
     'translated': '"Bonjour"',
     'output': 'こんにちは (Konnichiwa)'}
    ```

    **Create and run a chain of thought without "+" operator:** Please refer the
    `kotaemon.pipelines.cot.Thought` section for examples.

    This chain-of-thought optionally takes a termination check callback function.
    This function will be called after each thought is executed. It takes in a
    dictionary of all thought outputs so far, and it returns True or False. If
    True, the chain-of-thought will terminate. If unset, the default callback always
    returns False.
    """

    thoughts: List[Thought] = Param(
        default_callback=lambda *_: [], help="List of Thought"
    )
    llm: LLM = Param(help="The LLM model to use (base of kotaemon.llms.BaseLLM)")
    terminate: Callable = Param(
        default=lambda _: False,
        help="Callback on terminate condition. Default to always return False",
    )

    def run(self, **kwargs) -> Document:
        """Run the manual chain of thought"""

        inputs = deepcopy(kwargs)
        for idx, thought in enumerate(self.thoughts):
            if self.llm:
                thought.llm = self.llm
            self._prepare_child(thought, f"thought{idx}")

            output = thought(**inputs)
            inputs.update(output.content)
            if self.terminate(inputs):
                break

        return Document(inputs)

    def __add__(self, next_thought: Thought) -> "ManualSequentialChainOfThought":
        return ManualSequentialChainOfThought(
            thoughts=self.thoughts + [next_thought], llm=self.llm
        )

run 

1
run(**kwargs)

Run the manual chain of thought

Source code in libs\kotaemon\kotaemon\llms\cot.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def run(self, **kwargs) -> Document:
    """Run the manual chain of thought"""

    inputs = deepcopy(kwargs)
    for idx, thought in enumerate(self.thoughts):
        if self.llm:
            thought.llm = self.llm
        self._prepare_child(thought, f"thought{idx}")

        output = thought(**inputs)
        inputs.update(output.content)
        if self.terminate(inputs):
            break

    return Document(inputs)

Thought

Bases: BaseComponent

A thought in the chain of thought

  • Input: **kwargs pairs, where key is the placeholder in the prompt, and value is the value.
  • Output: an output dictionary

Usage:

Create and run a thought:

1
2
3
4
5
6
7
8
9
>> from kotaemon.pipelines.cot import Thought
>> thought = Thought(
     prompt="How to {action} {object}?",
     llm=LCAzureChatOpenAI(...),
     post_process=lambda string: {"tutorial": string},
   )
>> output = thought(action="install", object="python")
>> print(output)
{'tutorial': 'As an AI language model,...'}

Basically, when a thought is run, it will:

  1. Populate the prompt template with the input **kwargs.
  2. Run the LLM model with the populated prompt.
  3. Post-process the LLM output with the post-processor.

This Thought allows chaining sequentially with the + operator. For example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
>> llm = LCAzureChatOpenAI(...)
>> thought1 = Thought(
       prompt="Word {word} in {language} is ",
       llm=llm,
       post_process=lambda string: {"translated": string},
   )
>> thought2 = Thought(
        prompt="Translate {translated} to Japanese",
        llm=llm,
        post_process=lambda string: {"output": string},
   )

>> thought = thought1 + thought2
>> thought(word="hello", language="French")
{'word': 'hello',
 'language': 'French',
 'translated': '"Bonjour"',
 'output': 'こんにちは (Konnichiwa)'}

Under the hood, when the + operator is used, a ManualSequentialChainOfThought is created.

Source code in libs\kotaemon\kotaemon\llms\cot.py 
 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
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class Thought(BaseComponent):
    """A thought in the chain of thought

    - Input: `**kwargs` pairs, where key is the placeholder in the prompt, and
    value is the value.
    - Output: an output dictionary

    _**Usage:**_

    Create and run a thought:

    ```python
    >> from kotaemon.pipelines.cot import Thought
    >> thought = Thought(
         prompt="How to {action} {object}?",
         llm=LCAzureChatOpenAI(...),
         post_process=lambda string: {"tutorial": string},
       )
    >> output = thought(action="install", object="python")
    >> print(output)
    {'tutorial': 'As an AI language model,...'}
    ```

    Basically, when a thought is run, it will:

    1. Populate the prompt template with the input `**kwargs`.
    2. Run the LLM model with the populated prompt.
    3. Post-process the LLM output with the post-processor.

    This `Thought` allows chaining sequentially with the + operator. For example:

    ```python
    >> llm = LCAzureChatOpenAI(...)
    >> thought1 = Thought(
           prompt="Word {word} in {language} is ",
           llm=llm,
           post_process=lambda string: {"translated": string},
       )
    >> thought2 = Thought(
            prompt="Translate {translated} to Japanese",
            llm=llm,
            post_process=lambda string: {"output": string},
       )

    >> thought = thought1 + thought2
    >> thought(word="hello", language="French")
    {'word': 'hello',
     'language': 'French',
     'translated': '"Bonjour"',
     'output': 'こんにちは (Konnichiwa)'}
    ```

    Under the hood, when the `+` operator is used, a `ManualSequentialChainOfThought`
    is created.
    """

    prompt: str = Param(
        help=(
            "The prompt template string. This prompt template has Python-like variable"
            " placeholders, that then will be substituted with real values when this"
            " component is executed"
        )
    )
    llm: LLM = Node(LCAzureChatOpenAI, help="The LLM model to execute the input prompt")
    post_process: Function = Node(
        help=(
            "The function post-processor that post-processes LLM output prediction ."
            "It should take a string as input (this is the LLM output text) and return "
            "a dictionary, where the key should"
        )
    )

    @Node.auto(depends_on="prompt")
    def prompt_template(self):
        """Automatically wrap around param prompt. Can ignore"""
        return BasePromptComponent(template=self.prompt)

    def run(self, **kwargs) -> Document:
        """Run the chain of thought"""
        prompt = self.prompt_template(**kwargs).text
        response = self.llm(prompt).text
        response = self.post_process(response)

        return Document(response)

    def get_variables(self) -> List[str]:
        return []

    def __add__(self, next_thought: "Thought") -> "ManualSequentialChainOfThought":
        return ManualSequentialChainOfThought(
            thoughts=[self, next_thought], llm=self.llm
        )

prompt_template 

1
prompt_template()

Automatically wrap around param prompt. Can ignore

Source code in libs\kotaemon\kotaemon\llms\cot.py 
85
86
87
88
@Node.auto(depends_on="prompt")
def prompt_template(self):
    """Automatically wrap around param prompt. Can ignore"""
    return BasePromptComponent(template=self.prompt)

run 

1
run(**kwargs)

Run the chain of thought

Source code in libs\kotaemon\kotaemon\llms\cot.py 
90
91
92
93
94
95
96
def run(self, **kwargs) -> Document:
    """Run the chain of thought"""
    prompt = self.prompt_template(**kwargs).text
    response = self.llm(prompt).text
    response = self.post_process(response)

    return Document(response)

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
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)

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
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)

BasePromptComponent

Bases: BaseComponent

Base class for prompt components.

Parameters:

Name Type Description Default
template PromptTemplate

The prompt template.

 required
**kwargs

Any additional keyword arguments that will be used to populate the given template.

 {}
Source code in libs\kotaemon\kotaemon\llms\prompts\base.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
 79
 80
 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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class BasePromptComponent(BaseComponent):
    """
    Base class for prompt components.

    Args:
        template (PromptTemplate): The prompt template.
        **kwargs: Any additional keyword arguments that will be used to populate the
            given template.
    """

    class Config:
        middleware_switches = {"theflow.middleware.CachingMiddleware": False}
        allow_extra = True

    template: str | PromptTemplate

    @Param.auto(depends_on="template")
    def template__(self):
        return (
            self.template
            if isinstance(self.template, PromptTemplate)
            else PromptTemplate(self.template)
        )

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.__set(**kwargs)

    def __check_redundant_kwargs(self, **kwargs):
        """
        Check for redundant keyword arguments.

        Parameters:
            **kwargs (dict): A dictionary of keyword arguments.

        Raises:
            ValueError: If any keys provided are not in the template.

        Returns:
            None
        """
        self.template__.check_redundant_kwargs(**kwargs)

    def __check_unset_placeholders(self):
        """
        Check if all the placeholders in the template are set.

        This function checks if all the expected placeholders in the template are set as
            attributes of the object. If any placeholders are missing, a `ValueError`
            is raised with the names of the missing keys.

        Parameters:
            None

        Returns:
            None
        """
        self.template__.check_missing_kwargs(**self.__dict__)

    def __validate_value_type(self, **kwargs):
        """
        Validates the value types of the given keyword arguments.

        Parameters:
            **kwargs (dict): A dictionary of keyword arguments to be validated.

        Raises:
            ValueError: If any of the values in the kwargs dictionary have an
                unsupported type.

        Returns:
            None
        """
        type_error = []
        for k, v in kwargs.items():
            if k.startswith("template"):
                continue
            if not isinstance(v, (str, int, Document, Callable)):  # type: ignore
                type_error.append((k, type(v)))

        if type_error:
            raise ValueError(
                "Type of values must be either int, str, Document, Callable, "
                f"found unsupported type for (key, type): {type_error}"
            )

    def __set(self, **kwargs):
        """
        Set the values of the attributes in the object based on the provided keyword
            arguments.

        Args:
            kwargs (dict): A dictionary with the attribute names as keys and the new
                values as values.

        Returns:
            None
        """
        self.__check_redundant_kwargs(**kwargs)
        self.__validate_value_type(**kwargs)

        self.__dict__.update(kwargs)

    def __prepare_value(self):
        """
        Generate a dictionary of keyword arguments based on the template's placeholders
            and the current instance's attributes.

        Returns:
            dict: A dictionary of keyword arguments.
        """

        def __prepare(key, value):
            if isinstance(value, str):
                return value
            if isinstance(value, (int, Document)):
                return str(value)

            raise ValueError(
                f"Unsupported type {type(value)} for template value of key {key}"
            )

        kwargs = {}
        for k in self.template__.placeholders:
            v = getattr(self, k)

            # if get a callable, execute to get its output
            if isinstance(v, Callable):  # type: ignore[arg-type]
                v = v()

            if isinstance(v, list):
                v = str([__prepare(k, each) for each in v])
            elif isinstance(v, (str, int, Document)):
                v = __prepare(k, v)
            else:
                raise ValueError(
                    f"Unsupported type {type(v)} for template value of key `{k}`"
                )
            kwargs[k] = v

        return kwargs

    def set_value(self, **kwargs):
        """
        Similar to `__set` but for external use.

        Set the values of the attributes in the object based on the provided keyword
            arguments.

        Args:
            kwargs (dict): A dictionary with the attribute names as keys and the new
                values as values.

        Returns:
            None
        """
        self.__set(**kwargs)

    def run(self, **kwargs):
        """
        Run the function with the given keyword arguments.

        Args:
            **kwargs: The keyword arguments to pass to the function.

        Returns:
            The result of calling the `populate` method of the `template` object
            with the given keyword arguments.
        """
        self.__set(**kwargs)
        self.__check_unset_placeholders()
        prepared_kwargs = self.__prepare_value()

        text = self.template__.populate(**prepared_kwargs)
        return Document(text=text, metadata={"origin": "PromptComponent"})

    def flow(self):
        return self.__call__()

set_value 

1
set_value(**kwargs)

Similar to __set but for external use.

Set the values of the attributes in the object based on the provided keyword arguments.

Parameters:

Name Type Description Default
kwargs dict

A dictionary with the attribute names as keys and the new values as values.

 {}

Returns:

Type Description

None
Source code in libs\kotaemon\kotaemon\llms\prompts\base.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def set_value(self, **kwargs):
    """
    Similar to `__set` but for external use.

    Set the values of the attributes in the object based on the provided keyword
        arguments.

    Args:
        kwargs (dict): A dictionary with the attribute names as keys and the new
            values as values.

    Returns:
        None
    """
    self.__set(**kwargs)

run 

1
run(**kwargs)

Run the function with the given keyword arguments.

Parameters:

Name Type Description Default
**kwargs

The keyword arguments to pass to the function.

 {}

Returns:

Type Description

The result of calling the populate method of the template object

with the given keyword arguments.
Source code in libs\kotaemon\kotaemon\llms\prompts\base.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def run(self, **kwargs):
    """
    Run the function with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to pass to the function.

    Returns:
        The result of calling the `populate` method of the `template` object
        with the given keyword arguments.
    """
    self.__set(**kwargs)
    self.__check_unset_placeholders()
    prepared_kwargs = self.__prepare_value()

    text = self.template__.populate(**prepared_kwargs)
    return Document(text=text, metadata={"origin": "PromptComponent"})

PromptTemplate

Base class for prompt templates.

Source code in libs\kotaemon\kotaemon\llms\prompts\template.py 
  5
  6
  7
  8
  9
 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
 79
 80
 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
class PromptTemplate:
    """
    Base class for prompt templates.
    """

    def __init__(self, template: str, ignore_invalid=True):
        template = template
        formatter = Formatter()
        parsed_template = list(formatter.parse(template))

        placeholders = set()
        for _, key, _, _ in parsed_template:
            if key is None:
                continue
            if not key.isidentifier():
                if ignore_invalid:
                    warnings.warn(f"Ignore invalid placeholder: {key}.", UserWarning)
                else:
                    raise ValueError(
                        "Placeholder name must be a valid Python identifier, found:"
                        f" {key}."
                    )
            placeholders.add(key)

        self.template = template
        self.placeholders = placeholders
        self.__formatter = formatter
        self.__parsed_template = parsed_template

    def check_missing_kwargs(self, **kwargs):
        """
        Check if all the placeholders in the template are set.

        This function checks if all the expected placeholders in the template are set as
            attributes of the object. If any placeholders are missing, a `ValueError`
            is raised with the names of the missing keys.

        Parameters:
            None

        Returns:
            None
        """
        missing_keys = self.placeholders.difference(kwargs.keys())
        if missing_keys:
            raise ValueError(f"Missing keys in template: {','.join(missing_keys)}")

    def check_redundant_kwargs(self, **kwargs):
        """
        Check if all the placeholders in the template are set.

        This function checks if all the expected placeholders in the template are set as
            attributes of the object. If any placeholders are missing, a `ValueError`
            is raised with the names of the missing keys.

        Parameters:
            None

        Returns:
            None
        """
        provided_keys = set(kwargs.keys())
        redundant_keys = provided_keys - self.placeholders

        if redundant_keys:
            warnings.warn(
                f"Keys provided but not in template: {','.join(redundant_keys)}",
                UserWarning,
            )

    def populate(self, **kwargs) -> str:
        """
        Strictly populate the template with the given keyword arguments.

        Args:
            **kwargs: The keyword arguments to populate the template.
                      Each keyword corresponds to a placeholder in the template.

        Returns:
            The populated template.

        Raises:
            ValueError: If an unknown placeholder is provided.
        """
        self.check_missing_kwargs(**kwargs)

        return self.partial_populate(**kwargs)

    def partial_populate(self, **kwargs):
        """
        Partially populate the template with the given keyword arguments.

        Args:
            **kwargs: The keyword arguments to populate the template.
                      Each keyword corresponds to a placeholder in the template.

        Returns:
            str: The populated template.
        """
        self.check_redundant_kwargs(**kwargs)

        prompt = []
        for literal_text, field_name, format_spec, conversion in self.__parsed_template:
            prompt.append(literal_text)

            if field_name is None:
                continue

            if field_name not in kwargs:
                if conversion:
                    value = f"{{{field_name}}}!{conversion}:{format_spec}"
                else:
                    value = f"{{{field_name}:{format_spec}}}"
            else:
                value = kwargs[field_name]
                if conversion is not None:
                    value = self.__formatter.convert_field(value, conversion)
                if format_spec is not None:
                    value = self.__formatter.format_field(value, format_spec)

            prompt.append(value)

        return "".join(prompt)

    def __add__(self, other):
        """
        Create a new PromptTemplate object by concatenating the template of the current
            object with the template of another PromptTemplate object.

        Parameters:
            other (PromptTemplate): Another PromptTemplate object.

        Returns:
            PromptTemplate: A new PromptTemplate object with the concatenated templates.
        """
        return PromptTemplate(self.template + "\n" + other.template)

check_missing_kwargs 

1
check_missing_kwargs(**kwargs)

Check if all the placeholders in the template are set.

This function checks if all the expected placeholders in the template are set as attributes of the object. If any placeholders are missing, a ValueError is raised with the names of the missing keys.

Returns:

Type Description

None
Source code in libs\kotaemon\kotaemon\llms\prompts\template.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def check_missing_kwargs(self, **kwargs):
    """
    Check if all the placeholders in the template are set.

    This function checks if all the expected placeholders in the template are set as
        attributes of the object. If any placeholders are missing, a `ValueError`
        is raised with the names of the missing keys.

    Parameters:
        None

    Returns:
        None
    """
    missing_keys = self.placeholders.difference(kwargs.keys())
    if missing_keys:
        raise ValueError(f"Missing keys in template: {','.join(missing_keys)}")

check_redundant_kwargs 

1
check_redundant_kwargs(**kwargs)

Check if all the placeholders in the template are set.

This function checks if all the expected placeholders in the template are set as attributes of the object. If any placeholders are missing, a ValueError is raised with the names of the missing keys.

Returns:

Type Description

None
Source code in libs\kotaemon\kotaemon\llms\prompts\template.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def check_redundant_kwargs(self, **kwargs):
    """
    Check if all the placeholders in the template are set.

    This function checks if all the expected placeholders in the template are set as
        attributes of the object. If any placeholders are missing, a `ValueError`
        is raised with the names of the missing keys.

    Parameters:
        None

    Returns:
        None
    """
    provided_keys = set(kwargs.keys())
    redundant_keys = provided_keys - self.placeholders

    if redundant_keys:
        warnings.warn(
            f"Keys provided but not in template: {','.join(redundant_keys)}",
            UserWarning,
        )

populate 

1
populate(**kwargs)

Strictly populate the template with the given keyword arguments.

Parameters:

Name Type Description Default
**kwargs

The keyword arguments to populate the template. Each keyword corresponds to a placeholder in the template.

 {}

Returns:

Type Description
str

The populated template.

Raises:

Type Description
ValueError

If an unknown placeholder is provided.
Source code in libs\kotaemon\kotaemon\llms\prompts\template.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def populate(self, **kwargs) -> str:
    """
    Strictly populate the template with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to populate the template.
                  Each keyword corresponds to a placeholder in the template.

    Returns:
        The populated template.

    Raises:
        ValueError: If an unknown placeholder is provided.
    """
    self.check_missing_kwargs(**kwargs)

    return self.partial_populate(**kwargs)

partial_populate 

1
partial_populate(**kwargs)

Partially populate the template with the given keyword arguments.

Parameters:

Name Type Description Default
**kwargs

The keyword arguments to populate the template. Each keyword corresponds to a placeholder in the template.

 {}

Returns:

Name Type Description
str

The populated template.
Source code in libs\kotaemon\kotaemon\llms\prompts\template.py 
 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
def partial_populate(self, **kwargs):
    """
    Partially populate the template with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to populate the template.
                  Each keyword corresponds to a placeholder in the template.

    Returns:
        str: The populated template.
    """
    self.check_redundant_kwargs(**kwargs)

    prompt = []
    for literal_text, field_name, format_spec, conversion in self.__parsed_template:
        prompt.append(literal_text)

        if field_name is None:
            continue

        if field_name not in kwargs:
            if conversion:
                value = f"{{{field_name}}}!{conversion}:{format_spec}"
            else:
                value = f"{{{field_name}:{format_spec}}}"
        else:
            value = kwargs[field_name]
            if conversion is not None:
                value = self.__formatter.convert_field(value, conversion)
            if format_spec is not None:
                value = self.__formatter.format_field(value, format_spec)

        prompt.append(value)

    return "".join(prompt)