Build Your First Agent Application#

Install EvoFabric#

You can use the pip command to install EvoFabric:

pip install evofabric

Define StateSchema#

The construction of the graph requires first declaring the states being passed and the update mechanism within the graph. The purpose is to clearly define what messages are passed between nodes and how the messages should be updated.

Among them,

StateSchema supports two types: BaseModel and TypedDict.
All parameters must use Annotated to declare variable types and update mechanism.
The update mechanism needs to be registered in advance using @StateUpdater.register("update_name"), and the framework has already provided two update mechanisms: append_messages and overwrite.
When introducing AgentNode into the graph, it is necessary to declare messages: Annotated[list[StateMessage], "append_messages"], which maintains the large model context required by AgentNode. (append_messages appends each node’s output messages to the message list.)

Define Node#

We currently provide two predefined Node types: AgentNode and UserNode, and support users in customizing nodes through various methods.

AgentNode obtains large model outputs via Chat mode and has tool calling, memory management, input formatting, and output formatting features.
UserNode allows users to input information via the console as a UserMessage added to the messages list.

Add nodes and edges#

Graph construction requires first adding nodes and edges, and then setting the graph’s starting node. Subsequently, call the build() method to obtain a runnable graph engine.

When constructing a node, need to specify node name, node instance, node’s action mode (any and all), and state merging strategy required for multi-input nodes (optional).
When adding an ordinary edge, you need to specify the source node and target node. (See add_edge())
When adding a conditional edge, specify the source node, a routing function, and the full set of target nodes returned by the routing function. (Reference add_condition_edge())

Note

When conditional edges are executed, if the routing function returns a node name not in the complete set of target nodes, an exception is thrown.

Example:

def fc_router(state: State):
    last_message = state.messages[-1]
    if isinstance(last_message, AssistantMessage):
        reply = last_message.content
        if "::TO::user:" in reply:
            return "user"
        elif "::TO::end:" in reply:
            return "end"
    elif isinstance(last_message, ToolMessage):
        return "agent"
    return "end"

graph_builder = GraphBuilder(state_schema=StateSchema)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("user", user_node)
graph_builder.set_entry_point("agent")
graph_builder.add_condition_edge(
    "agent",
    router=fc_router,
    possible_targets={"user", "end", "agent"}
)
graph_builder.add_edge("user", "agent")
graph = graph_builder.build()

running diagram#

Graph execution can use the run() method, with the input being a dictionary strictly corresponding to the declaration of StateSchema.

Note

Fields not declared in StateSchema will be discarded.
It is not necessary to fully declare all fields in StateSchema; all unassigned fields will be given default values based on their types.

response = await graph.run({
    "messages": [UserMessage(content="What's the weather of my city?")],
    "user": "xxx"
})

Complete Demo#

import asyncio
from typing import Annotated

from pydantic import BaseModel

from evofabric.core.agent import AgentNode, UserNode
from evofabric.core.clients import OpenAIChatClient
from evofabric.core.graph import GraphBuilder
from evofabric.core.tool import ToolManager
from evofabric.core.typing import AssistantMessage, State, StateMessage, ToolMessage, UserMessage


class StateSchema(BaseModel):
    messages: Annotated[list[StateMessage], "append_messages"]


def check_weather(city: str):
    """Check city weather"""
    return f"Weather of {city} if good"


async def main():
    llm_chat_client = OpenAIChatClient(
        model="your-model-name",
        client_kwargs={"api_key": "<your-api-key>"}
    )
    agent_node = AgentNode(
        client=llm_chat_client,
        system_prompt="You are a helpful assistant. You can make tool calls to solve user's query."
                      "If you need more information from user, output ::TO::user:"
                      "If you wish to end the conversation, output ::TO::end:",
        tool_manager=ToolManager(tools=[check_weather]),
    )

    user_node = UserNode()

    def fc_router(state: State):
        last_message = state.messages[-1]
        if isinstance(last_message, AssistantMessage):
            reply = last_message.content
            if "::TO::user:" in reply:
                return "user"
            elif "::TO::end:" in reply:
                return "end"
        elif isinstance(last_message, ToolMessage):
            return "agent"
        return "end"

    graph_builder = GraphBuilder(state_schema=StateSchema)
    graph_builder.add_node("agent", agent_node)
    graph_builder.add_node("user", user_node)
    graph_builder.set_entry_point("agent")
    graph_builder.add_condition_edge(
        "agent",
        router=fc_router,
        possible_targets={"user", "end", "agent"}
    )
    graph_builder.add_edge("user", "agent")
    graph = graph_builder.build()

    response = await graph.run({
        "messages": [UserMessage(content="What's the weather of my city?")]
    })
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

Build Your First Agent Application

Contents

Build Your First Agent Application#

Install EvoFabric#

Define StateSchema#

Define Node#

Add nodes and edges#

running diagram#

Complete Demo#