Skip to content

Branching

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

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)