1
2
from core import Flow
from manas_ai.base import Edge

1
2
3
4
5
6
def __init__(
    self,
    name: Optional[str] = None,
    parallel_execution: bool = False,
    max_concurrency: Optional[int] = None
)

1
def add_node(self, node: Node) -> str

1
2
3
4
5
6
def add_edge(
    self,
    source_node: Union[str, Node],
    target_node: Union[str, Node],
    name: Optional[str] = None
) -> Edge

1
2
3
4
5
async def process(
    self,
    inputs: Dict[str, Any],
    node_inputs: Optional[Dict[str, Dict[str, Any]]] = None
) -> Dict[str, Dict[str, Any]]

1
def get_node(self, node_id: str) -> Node

1
def validate(self) -> bool

1
2
@property
def nodes(self) -> Dict[str, Node]

1
2
@property
def edges(self) -> List[Edge]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# Create nodes
qa_node = QANode(name="qa", config=qa_config)
summary_node = QANode(name="summary", config=summary_config)

# Create flow
flow = Flow(name="qa_flow")

# Add nodes
qa_id = flow.add_node(qa_node)
summary_id = flow.add_node(summary_node)

# Connect nodes
flow.add_edge(qa_id, summary_id)

# Process
result = await flow.process({
    "question": "Explain quantum computing",
    "summary_prompt": lambda qa_result: f"Summarize: {qa_result['answer']}"
})

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
# Create parallel flow
flow = Flow(
    name="parallel_research",
    parallel_execution=True,
    max_concurrency=3
)

# Add independent research nodes
physics = QANode(name="physics", config=physics_config)
chemistry = QANode(name="chemistry", config=chemistry_config)
biology = QANode(name="biology", config=biology_config)

# Add analyzer node
analyzer = QANode(name="analyzer", config=analyzer_config)

# Add nodes
p_id = flow.add_node(physics)
c_id = flow.add_node(chemistry)
b_id = flow.add_node(biology)
a_id = flow.add_node(analyzer)

# Connect nodes
for node_id in [p_id, c_id, b_id]:
    flow.add_edge(node_id, a_id)

# Process - physics, chemistry, and biology will run in parallel
result = await flow.process({
    "topic": "energy transformation in nature"
})

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
33
34
35
36
from typing import List

def create_research_flow(topics: List[str]) -> Flow:
    """Create a dynamic research flow based on topics."""
    flow = Flow(parallel_execution=True)
    
    # Create researcher nodes for each topic
    researcher_ids = []
    for topic in topics:
        researcher = QANode(
            name=f"researcher_{topic}",
            config=QAConfig(
                system_prompt=f"Research {topic} thoroughly"
            )
        )
        researcher_ids.append(flow.add_node(researcher))
    
    # Create analyzer node
    analyzer = QANode(
        name="analyzer",
        config=QAConfig(
            system_prompt="Synthesize research findings"
        )
    )
    analyzer_id = flow.add_node(analyzer)
    
    # Connect researchers to analyzer
    for r_id in researcher_ids:
        flow.add_edge(r_id, analyzer_id)
    
    return flow

# Usage
topics = ["quantum_computing", "machine_learning", "robotics"]
flow = create_research_flow(topics)
result = await flow.process({"depth": "comprehensive"})