Skip to content

FastAgentProtocol

The FastAgentProtocol is designed to provide a simplified, faster alternative to the previous AgentProtocol. It enables efficient communication between cognitive agents and the main system without the overhead of remote connections.

Why Use FastAgentProtocol?

  • Speed & Simplicity: It removes the need for remote server connections, significantly reducing latency.
  • Standardization: All agents in the cognition layer must implement this protocol for consistent communication.
  • Flexibility: The main system no longer needs to manage execution, as the agent handles it independently.

How It Works

The FastAgentProtocol requires defining how an agent processes tasks in steps, where each step returns essential feedback to the main system.

Key Components

  • Iterator: A function that processes input iteratively, yielding steps.
  • Content Getter: Retrieves the content of each step.
  • Is Last Getter: Determines if the current step is the final one.
  • Step Name Getter (optional): Assigns a name to each step.

Implementing a Custom Agent with FastAgentProtocol

  1. Define an Iterative Process: The agent's main function should yield steps, each containing a name, content, and finalization status.
  2. Create the Protocol Server:
    • Use the FastAgentProtocol class.
    • Define getters for step information.
  3. Example Custom Agent
from cognition_layer.api import FastAgentProtocol
from ecm.mediator.Interpreter import Interpreter
from operator import attrgetter

class MyAgent:
    def __init__(self, interpreter: Interpreter):
        self.interpreter = interpreter

    def complete_task(self, input: str):

        my_agent.send_user_input(input)
        for step in steps:
            next_action = my_agent.next_action()
            result = interpreter.run(next_action)
            step = parse_result_contents(result)
            yield step

def get_fast_ap_server(interpreter: Interpreter) -> FastAgentProtocol:
    agent = MyAgent(interpreter)

    return FastAgentProtocol(
        name="My Custom Agent",
        iterator=agent.complete_task,
        step_name_getter=attrgetter("name"),    # Equivalent to step.name
        content_getter=attrgetter("content"),   # Equivalent to step.content
        is_last_getter=attrgetter("is_last"),   # Equivalent to step.is_last
    )

Using an implemented serverwith Fast AP

To use an agent with FastAgentProtocol in the main program:

from cognition_layer.fast_react.api.server import get_fast_ap_server
from cognition_layer.tools.registry import ItemRegistry

server = get_fast_ap_server(interpreter=interpreter)

while True:
    query = input("Task: ")
    for step in server.send_task(query):
        print(f"Step: {step.name} -> {step.content}")

Conclusion

FastAgentProtocol provides a unified, efficient approach to agent communication. By following the pattern of defining iterators and feedback structures, agents can seamlessly integrate with the main system while optimizing performance and reducing complexity.