agentconnect.prompts.agent_prompts module¶

Agent workflow definitions for the AgentConnect framework.

This module provides the core workflow definitions for different agent types, using LangGraph to create stateful, multi-step workflows. These workflows enable agents to make decisions, collaborate with other agents, and process complex tasks.

The module implements the ReAct (Reasoning + Acting) pattern using LangGraph’s StateGraph to create workflows with multiple nodes for preprocessing, execution, and postprocessing.

class AgentMode(*values)¶

Bases: Enum

Enum representing the operational modes of an agent.

CUSTOM_RUNNABLE¶: Agent uses a custom runnable for processing

SYSTEM_PROMPT¶: Agent uses a system prompt for processing

CUSTOM_RUNNABLE = 'custom_runnable'¶

SYSTEM_PROMPT = 'system_prompt'¶

class WorkflowState(*values)¶

Bases: Enum

Enum for workflow states.

THINKING¶: Agent is thinking about the problem

RESPONDING¶: Agent is generating a response

TOOL_CALLING¶: Agent is calling a tool

COLLABORATION¶: Agent is collaborating with another agent

TASK_DECOMPOSITION¶: Agent is breaking down a task

CAPABILITY_MATCHING¶: Agent is matching capabilities to tasks

ERROR¶: An error occurred in the workflow

COMPLETE¶: Workflow is complete

THINKING = 'thinking'¶

RESPONDING = 'responding'¶

TOOL_CALLING = 'tool_calling'¶

COLLABORATION = 'collaboration'¶

TASK_DECOMPOSITION = 'task_decomposition'¶

CAPABILITY_MATCHING = 'capability_matching'¶

ERROR = 'error'¶

COMPLETE = 'complete'¶

class AgentState¶

Bases: TypedDict

Base state for agent workflows.

This TypedDict defines the structure of the state object used in agent workflows. It includes fields for messages, sender/receiver information, capabilities, and various tracking fields.

messages¶: Sequence of messages in the conversation

sender¶: ID of the message sender

receiver¶: ID of the message receiver

mode¶: Optional mode of operation

capabilities¶: Optional list of agent capabilities

runnable_result¶: Optional result from a runnable

collaboration_results¶: Optional results from collaboration

agents_found¶: Optional list of agents found

retry_count¶: Optional count of retries

error¶: Optional error message

context_reset¶: Optional flag for context reset

topic_changed¶: Optional flag for topic change

last_interaction_time¶: Optional timestamp of last interaction

messages: Annotated[Sequence[BaseMessage]]¶

sender: str¶

receiver: str¶

mode: Optional[str]¶

capabilities: Optional[List[str]]¶

runnable_result: Optional[Dict[str, Any]]¶

collaboration_results: Optional[Dict[str, Any]]¶

agents_found: Optional[List[Dict[str, Any]]]¶

retry_count: Optional[Dict[str, int]]¶

error: Optional[str]¶

context_reset: Optional[bool]¶

topic_changed: Optional[bool]¶

last_interaction_time: Optional[float]¶

class CollaborationState¶

Bases: AgentState

State for collaboration workflows.

This TypedDict extends AgentState with additional fields specific to collaboration workflows.

found_agents¶: List of agent IDs found for collaboration

capabilities¶: List of agent capabilities

collaboration_result¶: Optional result from collaboration

subtasks¶: List of subtasks

current_subtask¶: Optional current subtask

completed_subtasks¶: List of completed subtasks

task_description¶: Optional task description

action¶: Optional action to take

error¶: Optional error message

found_agents: List[str]¶

capabilities: List[Dict[str, Any]]¶

collaboration_result: Optional[Dict[str, Any]]¶

subtasks: List[Dict[str, Any]]¶

current_subtask: Optional[Dict[str, Any]]¶

completed_subtasks: List[Dict[str, Any]]¶

task_description: Optional[str]¶

action: Optional[str]¶

error: Optional[str]¶

messages: Annotated[Sequence[BaseMessage]]¶

sender: str¶

receiver: str¶

mode: Optional[str]¶

runnable_result: Optional[Dict[str, Any]]¶

collaboration_results: Optional[Dict[str, Any]]¶

agents_found: Optional[List[Dict[str, Any]]]¶

retry_count: Optional[Dict[str, int]]¶

context_reset: Optional[bool]¶

topic_changed: Optional[bool]¶

last_interaction_time: Optional[float]¶

class DecisionOutput(**data)¶

Bases: BaseModel

Output schema for decision nodes.

Parameters:

action (str)
reason (str)

action¶: The action to take next

reason¶: The reason for the decision

action: str¶

reason: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class TaskDecompositionOutput(**data)¶

Bases: BaseModel

Output schema for task decomposition.

Parameters:: subtasks (List[Dict[str, Any]])

subtasks¶: List of subtasks

subtasks: List[Dict[str, Any]]¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class AgentWorkflow(agent_id, llm, tools, prompt_templates, custom_tools=None)¶

Bases: object

Base class for agent workflows.

This class provides the foundation for creating agent workflows using LangGraph. It handles the creation of the workflow graph, tools, and prompt templates.

Parameters:

agent_id (str)
llm (BaseChatModel)
tools (PromptTools)
prompt_templates (PromptTemplates)
custom_tools (List[BaseTool] | None)

agent_id¶: Unique identifier for the agent

llm¶: Language model to use

tools¶: Tools available to the agent

prompt_templates¶: Prompt templates for the agent

custom_tools¶: Optional list of custom LangChain tools

workflow¶: The workflow graph

mode¶: The agent’s operational mode

build_workflow()¶

Build the workflow graph for the agent.

Return type:: StateGraph
Returns:: A StateGraph instance representing the agent’s workflow.

compile()¶

Compile the workflow with memory persistence.

Returns:: The compiled workflow with memory persistence

class AIAgentWorkflow(agent_id, system_prompt_config, llm, tools, prompt_templates, custom_tools=None)¶

Bases: AgentWorkflow

Workflow for AI agents with enhanced capabilities.

This workflow extends the base AgentWorkflow with AI-specific capabilities and system prompt configuration.

Parameters:

agent_id (str)
system_prompt_config (SystemPromptConfig)
llm (BaseChatModel)
tools (PromptTools)
prompt_templates (PromptTemplates)
custom_tools (List[BaseTool] | None)

system_prompt_config¶: Configuration for the system prompt

class TaskDecompositionWorkflow(agent_id, system_prompt_config, llm, tools, prompt_templates, custom_tools=None)¶

Bases: AgentWorkflow

Workflow for decomposing tasks into subtasks.

This workflow extends the base AgentWorkflow with task decomposition capabilities and system prompt configuration.

Parameters:

agent_id (str)
system_prompt_config (SystemPromptConfig)
llm (BaseChatModel)
tools (PromptTools)
prompt_templates (PromptTemplates)
custom_tools (List[BaseTool] | None)

system_prompt_config¶: Configuration for the system prompt

class CollaborationRequestWorkflow(agent_id, system_prompt_config, llm, tools, prompt_templates, custom_tools=None)¶

Bases: AgentWorkflow

Workflow for handling collaboration requests.

This workflow extends the base AgentWorkflow with collaboration capabilities and system prompt configuration.

Parameters:

agent_id (str)
system_prompt_config (SystemPromptConfig)
llm (BaseChatModel)
tools (PromptTools)
prompt_templates (PromptTemplates)
custom_tools (List[BaseTool] | None)

system_prompt_config¶: Configuration for the system prompt

create_workflow_for_agent(agent_type, system_config, llm, tools, prompt_templates, agent_id=None, custom_tools=None)¶

Factory function to create workflows based on agent type.

Parameters:

agent_type (str) – Type of agent workflow to create
system_config (SystemPromptConfig) – System prompt configuration
llm (BaseChatModel) – Language model to use
tools (PromptTools) – Tools available to the agent
prompt_templates (PromptTemplates) – Prompt templates for the agent
agent_id (Optional[str]) – Optional agent ID for tool context
custom_tools (Optional[List[BaseTool]]) – Optional list of custom LangChain tools

Return type:

AgentWorkflow

Returns:

An AgentWorkflow instance

Raises:

ValueError – If the agent type is unknown