AI Coordinator¶
lumen.ai.coordinator
¶
__all__ = ['Coordinator', 'Plan', 'DependencyResolver', 'Planner']
module-attribute
¶
Coordinator
¶
Bases:
A Coordinator is responsible for coordinating the actions of a number of agents towards the user defined query by computing an execution graph and then executing each step along the graph.
agents = param.List(default=[ChatAgent], doc='\n List of agents to coordinate.')
class-attribute
instance-attribute
¶
context = param.Dict(default={})
class-attribute
instance-attribute
¶
history = param.Integer(default=3, doc='\n Number of previous user-assistant interactions to include in the chat history.')
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'Coordinator' / 'main.jinja2'}, 'tool_relevance': {'template': PROMPTS_DIR / 'Coordinator' / 'tool_relevance.jinja2', 'response_model': ThinkingYesNo}})
class-attribute
instance-attribute
¶
validation_enabled = param.Boolean(default=True, allow_refs=True, doc='\n Whether to enable the ValidationAgent in the planning process.')
class-attribute
instance-attribute
¶
verbose = param.Boolean(default=False, allow_refs=True, doc='\n Whether to show verbose output.')
class-attribute
instance-attribute
¶
within_ui = param.Boolean(default=False, constant=True, doc='\n Whether this coordinator is being used within the UI.')
class-attribute
instance-attribute
¶
respond(messages, context, **kwargs)
async
¶
sync(context)
async
¶
DependencyResolver
¶
Bases:
DependencyResolver is a type of Coordinator that chooses the agent to answer the query and then recursively resolves all the information required for that agent until the answer is available.
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'DependencyResolver' / 'main.jinja2', 'response_model': make_agent_model}})
class-attribute
instance-attribute
¶
Plan
¶
Bases:
A Plan is a Task that is a collection of other Tasks.
abort_on_error = param.Boolean(default=True, doc='\n If True, the report will abort if an error occurs.')
class-attribute
instance-attribute
¶
agents = param.List(item_type=Actor, default=[])
class-attribute
instance-attribute
¶
coordinator = param.ClassSelector(class_=(param.Parameterized))
class-attribute
instance-attribute
¶
interface = param.ClassSelector(class_=ChatFeed)
class-attribute
instance-attribute
¶
execute(context=None, **kwargs)
async
¶
render_task_history(i, failed=False)
¶
Planner
¶
Bases:
The Planner develops a plan to solve the user query step-by-step and then executes it.
planner_tools = param.List(default=[TableLookup], doc='\n List of tools to use to provide context for the planner prior\n to making a plan.')
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'Planner' / 'main.jinja2', 'response_model': make_plan_model}, 'follow_up': {'template': PROMPTS_DIR / 'Planner' / 'follow_up.jinja2', 'response_model': ThinkingYesNo}})
class-attribute
instance-attribute
¶
sync(context)
async
¶
Sync both main tools and planner tools.
base
¶
Coordinator
¶
Bases:
A Coordinator is responsible for coordinating the actions of a number of agents towards the user defined query by computing an execution graph and then executing each step along the graph.
agents = param.List(default=[ChatAgent], doc='\n List of agents to coordinate.')
class-attribute
instance-attribute
¶
context = param.Dict(default={})
class-attribute
instance-attribute
¶
history = param.Integer(default=3, doc='\n Number of previous user-assistant interactions to include in the chat history.')
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'Coordinator' / 'main.jinja2'}, 'tool_relevance': {'template': PROMPTS_DIR / 'Coordinator' / 'tool_relevance.jinja2', 'response_model': ThinkingYesNo}})
class-attribute
instance-attribute
¶
validation_enabled = param.Boolean(default=True, allow_refs=True, doc='\n Whether to enable the ValidationAgent in the planning process.')
class-attribute
instance-attribute
¶
verbose = param.Boolean(default=False, allow_refs=True, doc='\n Whether to show verbose output.')
class-attribute
instance-attribute
¶
within_ui = param.Boolean(default=False, constant=True, doc='\n Whether this coordinator is being used within the UI.')
class-attribute
instance-attribute
¶
respond(messages, context, **kwargs)
async
¶
sync(context)
async
¶
Plan
¶
Bases:
A Plan is a Task that is a collection of other Tasks.
abort_on_error = param.Boolean(default=True, doc='\n If True, the report will abort if an error occurs.')
class-attribute
instance-attribute
¶
agents = param.List(item_type=Actor, default=[])
class-attribute
instance-attribute
¶
coordinator = param.ClassSelector(class_=(param.Parameterized))
class-attribute
instance-attribute
¶
interface = param.ClassSelector(class_=ChatFeed)
class-attribute
instance-attribute
¶
execute(context=None, **kwargs)
async
¶
render_task_history(i, failed=False)
¶
dependency
¶
DependencyResolver
¶
Bases:
DependencyResolver is a type of Coordinator that chooses the agent to answer the query and then recursively resolves all the information required for that agent until the answer is available.
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'DependencyResolver' / 'main.jinja2', 'response_model': make_agent_model}})
class-attribute
instance-attribute
¶
make_agent_model(agent_names, primary=False)
¶
planner
¶
Planner
¶
Bases:
The Planner develops a plan to solve the user query step-by-step and then executes it.
planner_tools = param.List(default=[TableLookup], doc='\n List of tools to use to provide context for the planner prior\n to making a plan.')
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'Planner' / 'main.jinja2', 'response_model': make_plan_model}, 'follow_up': {'template': PROMPTS_DIR / 'Planner' / 'follow_up.jinja2', 'response_model': ThinkingYesNo}})
class-attribute
instance-attribute
¶
sync(context)
async
¶
Sync both main tools and planner tools.
RawPlan
¶
Bases:
steps = Field(description='\n A list of steps to perform that will solve user query. Each step MUST use a DIFFERENT actor than the previous step.\n Review your plan to ensure this constraint is met.\n ')
class-attribute
instance-attribute
¶
title = Field(description='A title that describes this plan, up to three words.')
class-attribute
instance-attribute
¶
RawStep
¶
Bases:
actor
instance-attribute
¶
instruction = Field(description='\n Concise instruction capturing user intent at the right altitude.\n\n Right altitude:\n - ❌ Too low: implementation details (SQL syntax, chart specs, row limits)\n - ❌ Too high: vague ("handle this", "process data")\n - ✅ Just right: clear intent + relationship to context\n\n Encode intent: add/modify/filter/compare/create — reference existing context when applicable.\n\n Never include implementation details unless user explicitly specified them.\n ')
class-attribute
instance-attribute
¶
title
instance-attribute
¶
Reasoning
¶
Bases: