以编程方式构建代理时，无法使用webview将消息发送到faust代理/主题

Question

我正在尝试以编程方式构建代理，并且一直在遵循this example。当我查看脚本开头生成的faust -A <> agents时，我可以看到主题和代理。根据示例，代理名称类似于{topic}_agent。所有这些代理分别连接到一些rest api。下面是代码。

def create_agent(next_topic, model_metadata: Dict):
    """ 
         creation of a single agent.
         `start_topic`:  str
             Just a string that you can use in other functions
             to figure out how messages in that topic can be
             transformed
         `next_topic`:  faust.topics.Topic
             A faust `app.topic` instance
    """
    async def agent(stream):
        """Call domino model and return response
        """
        async for message in stream:
            domino_req = {"data": message.asdict()}
            app.logger.info(domino_req)
            response = requests.post(
                model_metadata['url'],
                auth=(
                    model_metadata['auth'],
                    model_metadata['auth'],
                ),
                json=domino_req,
            )
            if response.status_code == 200:
                answer = response.json()
                await next_topic.send(answer['result'])
            else:
                app.logger.error(response.reason)
    # app.logger.info(f"NEW Agent Created: ## Agent - {consumer} ##")
    return agent

def agents(registry_response):
    """
        configuration of multiple agents
    """
    agents = []
    for key, value in registry_response.items():
        """ `topic`:  app.topic instance """
        agents.append(
            # `topic.start`: str
            # `topic.next`: faust.topics.Topic
            (create_agent(next_topic= all_responses, model_metadata = value),key)
        )
    return agents

def attach_agent(agent, topic):
    """ Attach the agent to the Faust app """
    # `topic.faust`: faust.topics.Topic
    # it is equivalent to `app.topic(topic.start)`
    print("hello")
    app.agent(channel=app.topic(topic), name=f"{topic}_agent")(agent)
    # new_agent.start()
    # app.logger.info("hello")
    # app.logger.info(new_agent)
    # app.logger.info(new_agent.info())
    # app.logger.info(app.agents)

# @app.task
# async def get_model_registry():
#     """
#     Create topics and agents for the initial set of models present in the registry
#     """
app.logger.info('APP STARTED')
app.logger.info('Fetching Models from Model Registry')
#TODO: Call the Model Registry and process it
#Just mocking the registry 
registry_response = initial_model_registry_metadata
app.logger.info(f'Number of Models Found {len(registry_response)}')
for agent,topic in agents(registry_response):
    attach_agent(agent, topic)

@app.page('/model/{key_ai_model}')
class frontdoor(View):
    async def get(self, request: Request) -> Response:
        return self.json({'key_ai_model': key_ai_model})
    async def post(self, request: Request, key_ai_model: str) -> Response:
        request_id = str(uuid.uuid4())
        src = await request.json()
        msg = GdeltRequest(**src)
        app.logger.info(msg)
        await gdelt_agent.cast(msg, key=request_id)
        return self.json({'request_id': request_id, 'key_ai_model': key_ai_model})

@app.agent(all_responses)
async def print_responses(stream):
    async for message in stream:
        print(message)

然而，当我发布一些东西时，我得到了一个错误- NameError: name 'gdelt_agent' is not defined。这里的任何帮助都将不胜感激。

Answer 1

我已经能够解决这个问题，方法是将代理作为一个值赋值在二进制中，然后使用它将消息传递给代理。

agents_dict = {}
topics_dict ={}
def attach_agent(agent, topic):
    """ Attach the agent to the Faust app """
    # `topic.faust`: faust.topics.Topic
    # it is equivalent to `app.topic(topic.start)`
    print("hello")
    topics_dict[topic] = app.topic(topic)
    agents_dict[topic] = app.agent(channel=topics_dict[topic], name=f"{topic}_agent")(agent)
    agents_dict[topic].start()

谢谢大家。