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=False, 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
¶
is_followup = param.Boolean(default=False)
class-attribute
instance-attribute
¶
execute(context=None, **kwargs)
async
¶
render_task_history(i=None, 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=[MetadataLookup], 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=False, 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
¶
is_followup = param.Boolean(default=False)
class-attribute
instance-attribute
¶
execute(context=None, **kwargs)
async
¶
render_task_history(i=None, 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=[MetadataLookup], 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 The sequence of steps to resolve the user's query.\n Ensure no two consecutive steps use the same actor.\n\n Adhere to separation of concerns:\n - Each step's instruction should be limited to that actor's task.\n - Do not include downstream objectives in upstream instructions.\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 - ❌ Too low: implementation details (SQL syntax, chart specs)\n - ❌ Too high: vague ("handle this", "process data")\n - ❌ Leaking: mentioning downstream purpose ("for plotting", "for the chart", "for analysis")\n - ✅ Just right: what THIS actor should do, nothing about why\n ', examples=['Query top 5 countries by sales', 'Get wind, temperature, dewpoint, and precipitation data for Seattle on Jan 1, 2020', 'Plot a line chart of sales over time.'])
class-attribute
instance-attribute
¶
title
instance-attribute
¶
Reasoning
¶
Bases: