Problem integrating jina with async system

[LawlAoux]

Describe the bug
I have an async system for a chat bot, and we would like to integrate jina for matching faqs. When trying to run it (the jina code for faq works perfectly well alone) I'm getting RuntimeError: you have an eventloop running but not using Jupyter/ipython, this may mean you are using Jina with other integration? if so, then you may want to use Client/Flow(asyncio=True). If not, then please report this issue here: https://github.com/jina-ai/jina. We tried to use Flow(asyncio=True) or AsyncFlow but still got the same error... Would appreciate it if you could help me with it.

Describe how you solve it
I think in general it would be easier to use jina with function compositions rather than the Flow object (maybe it is possible but we are not aware of how to do it). For example, instead of using TransformerTorchEncoder and SimpleIndexer in the flow, it would be convenient to use these as functions and not classes (e.g. indexer = SimpleIndexer(TransformerTorchEncoder) and then we give the input to this function). We think this API is much more convenient to developers, who can then build the indexer and query much more easily

[hanxiao]

Hi, thanks for reporting this issue. The exception message is on the point, yet it does not point out how to solve it. @jina-ai/team-core need better docs to explain the following:

If you are building a Jina app, which involves calling another Jina server via Client object. You will encounter this runtime error.

The solution is simple, use AsyncClient instead of Client in your @requests decorated function. Here is an example how it should be solved:

https://github.com/jina-ai/dalle-flow/blob/2c6ac56f7fc75d453714ef396b65e39db40f6064/executors/rerank/executor.py#L11-L13

In this example, DALLE Flow's executor calls CLIP-as-service for ranking. But as CLIP-as-service is a Jina-based service, directly using the sync-version of the Jina client will give that error you described.

The solution is simple. Just use AsyncClient, as described above. In the last example, CLIP-as-service wraps Jina Client and provides a single Client that can do both .rank() and .arank(). Hence you see I use .arank() there.

In the GLID3-XL there is another example, where I need CLIP encoding, the problem shows up again. The solution is again the same, use the async version of .encode():

https://github.com/hanxiao/glid-3-xl/blob/master/dalle_flow_glid3/sample.py#L116-L120

Note, that CLIP-as-service is just an example. So if we put it back to the context:

If you are building a Jina app, which involves calling another Jina server via Client object. You will encounter this runtime error.

Then, the two examples above are mapped the following:

1. If you are building a Jina app (i.e. DALLE Flow), which involves calling another Jina server (CLIP-as-service) via Client object. You will encounter this runtime error.
2. If you are building a Jina app (i.e. GLID3 XL), which involves calling another Jina server (CLIP-as-service) via Client object. You will encounter this runtime error.

In the case you do not have well-wrapped .ablahblah() method, simply make one via AsyncClient.

https://github.com/jina-ai/clip-as-service/blob/main/client/clip_client/client.py#L401

Finally, the difference between AsyncClient and Client is not as big as you imagined, they are both efficient:

While the standard Client is also asynchronous under the hood, its async version exposes this fact to the outside world, by allowing coroutines as input, and returning an asynchronous iterator. This means you can iterate over Responses one by one, as they come in.

https://docs.jina.ai/fundamentals/flow/client/#async-python-client

[LawlAoux]

Thanks for your reply! I'm not sure where I should replace this client with AsyncClient, (I tried to replace Flow with AsyncFlow) so I'm attaching the code that works (when running locally, but not when it is integrated with the chat bot).
I would like to emphasize that this Jina app works, it doesn't work with our chat bot that we developed (not related to Jina) which is an async system

import os
import functools
from typing import Iterator, Tuple

from jina import AsyncFlow, Document, DocumentArray, Flow
import gamla


_DIR_NAME = os.path.dirname(__file__)


def _input_generator(answers: Tuple[str, ...]) -> Iterator[Document]:
        for answer in answers:
            yield Document(text=answer)

@functools.cache
def index(answers: Tuple[str, ...]) -> None:
    flow = Flow(asyncio=True).load_config(os.path.join(_DIR_NAME, 'flows/flow-index.yaml'))
    data_path = os.path.join(_DIR_NAME, "sexual_health_faq.csv")
    with flow:
        flow.post(on="/index", inputs=_input_generator(answers), show_progress=True)

def query(user_utterance: str) -> Tuple[str, float]:
    flow = Flow().load_config('flows/flow-query.yaml')
    with flow:
        doc = Document(content=user_utterance)
        result = flow.post(on='/search', inputs=DocumentArray([doc]),
                           parameters={'top_k': 1},
                           line_format='text',
                           return_results=True,
                           )

        return gamla.pipe(result, gamla.head, gamla.attrgetter("_data"), gamla.attrgetter("matches"), gamla.attrgetter("_data"), gamla.head, gamla.juxt(gamla.attrgetter("text"), gamla.attrgetter("scores")))

if __name__ == "__main__":
    index(("we are open everyday from 8am to 4pm.",))
    query("what is your opening time?")
    

[JohannesMessner]

Hi @LawlAoux , let me see if I can help!

I see that you create an async Flow using or Python API (Flow(asyncio=True)), and then load a YAML config into that Flow. The resulting Flow object won't be async unless you specify the same in your flow-index.yaml file as well. The reason for this is that we consider the YAML file to be the source of truth when it comes to your Flow definition.
This is how to do that:

jtype: Flow
with:
  asyncio: True

Then instantiate your Flow:

flow = Flow().load_config(os.path.join(_DIR_NAME, 'flows/flow-index.yaml'))  # asyncio=True now unnecessary

But I agree that this is unexpected behaviour from the user's perspective, we will look into how to improve this experience.

Secondly, you only seem to make one of your Flows async, but you will have to apply the same treatment to both of them.
Also, as a general tip, for querying you usually do not need a Flow at all, and can instead use our Client, which is the recommended way of doing it:

from jina import Flow, Client
from docarray import Document, DocumentArray

flow = Flow().load_config(os.path.join(_DIR_NAME, 'flows/flow-index.yaml')
port = flow.port
... # do your other stuff
c = Client(port=port, asyncio=True)
async for results in c.post(on='/search', inputs=DocumentArray(), ...):
    ...  # process your results

Lastly, I like your idea for a functional API, but I don't think that such an API can handle the complexity that a Jina Flow offers, unfortunately. Consider, as just two examples, Flows that have complex, branching topologies, and Executors which take __init__() arguments. Both of these are not naturally doable using a functional API (imo), and since, as a matter of design principle, we don't like to have two ways of doing the same thing, I don't think that we will be able to support such an API even for simple Flows.

I hope that helps!

[LawlAoux]

Thank you very much guys! I have managed to overcome these errors with your help. I would like to ask you about optimization though, because the index and query flow run quite slow in my case (index is torch encoder + simple indexer from jina hub, and the query is the index + simple ranker from jina hub, and they run around 3 and 7 seconds, respectively). I understand that the cause of it may be the fact that I'm running it on cpu (I have a macbook pro), but do you have any tips on how to optimize the running time?

[JohannesMessner]

Glad we were able to help!
How many documents are you trying to index at a time? In general though, when it comes to performance, there are usually two bottlenecks: Encoding and matching.

1. For encoding, there is really not much you can do other than getting more compute. We have recently done some rudimentary benchmarks on, and the time spent during encoding is vastly dominated by the forward pass of your NN, whereas overhead introduced by Jina is really quite small (@samsja knows more on this).
   If you don't have access to a GPU locally you can always host your Flow on your favourite cloud provider's machines, or you may even want to give our early-access jcloud a try (although I am not sure about current GPU availability on jcloud).
2. On matching/querying, there might be easier room for improvement. SimpleIndexer searches through all possible answer candidates (embeddings) in a brute force fashion, which can become quite slow quite quickly. What you usually want is to use an approximate nearest neighbours (ANN) algorithm to perform this search way more quickly. One possibility is to use ANNLiteIndexer for this purpose. You may also want to look into our storage backends for fast vector retrieval.

If you want to go into more details regarding performance optimization and/or your specific performance bottlenecks, feel free to open a separate ticket on the issue, it really helps the rest of the community to find this information!

[LawlAoux]

Thanks for the reply! I tried a single doc (just tested it for a single faq pair). I will definitely try the indexer that you suggested and will let you know if it perform better or not (also I will now try it on a set of 10 faqs)

[JoanFM]

Hey @LawlAoux, thanks for reporting the issue, as the original issue is solved, I will proceed to close it. Feel free to open new issues with any concerns or questions.