| layout | default |
|---|---|
| title | Chapter 2: Architecture and Routing System |
| nav_order | 2 |
| parent | AgenticSeek Tutorial |
Welcome to Chapter 2: Architecture and Routing System. In this part of AgenticSeek Tutorial: Local-First Autonomous Agent Operations, you will build an intuitive mental model first, then move into concrete implementation details and practical production tradeoffs.
This chapter explains how AgenticSeek decomposes tasks across planner and specialist agents.
- map the high-level component boundaries
- understand how routing chooses specialist agents
- identify where prompts, tools, and execution loops live
- reason about failure domains across routing and execution
Core directories to understand first:
llm_router/: routing and model dispatch utilitiessources/agents/: specialist agents (planner,browser,code,file,mcp,casual)sources/tools/: executable capabilities invoked by agent tool blocksprompts/: prompt templates (baseandjarvisvariants)frontend/+ backend entrypoints: user interaction and orchestration surface
flowchart TD
A[User request] --> B[Routing decision]
B --> C[Planner agent]
C --> D[Specialist agent selection]
D --> E[Tool block generation]
E --> F[Tool execution]
F --> G[Answer + reasoning]
- planner quality directly affects downstream execution quality
- explicit user intents reduce wrong-agent assignment risk
- tool-block parsing is a critical control point for deterministic execution
- verify which agent class owns each new behavior
- keep tool scope narrow and single-purpose
- update related prompt templates when adding tools
- test routing behavior on at least three intent categories
You now understand where routing, agent logic, and tool execution boundaries sit.
Next: Chapter 3: Installation, Runtime, and Provider Setup
The get_latest_answer function in api.py handles a key part of this chapter's functionality:
@api.get("/latest_answer")
async def get_latest_answer():
global query_resp_history
if interaction.current_agent is None:
return JSONResponse(status_code=404, content={"error": "No agent available"})
uid = str(uuid.uuid4())
if not any(q["answer"] == interaction.current_agent.last_answer for q in query_resp_history):
query_resp = {
"done": "false",
"answer": interaction.current_agent.last_answer,
"reasoning": interaction.current_agent.last_reasoning,
"agent_name": interaction.current_agent.agent_name if interaction.current_agent else "None",
"success": interaction.current_agent.success,
"blocks": {f'{i}': block.jsonify() for i, block in enumerate(interaction.get_last_blocks_result())} if interaction.current_agent else {},
"status": interaction.current_agent.get_status_message if interaction.current_agent else "No status available",
"uid": uid
}
interaction.current_agent.last_answer = ""
interaction.current_agent.last_reasoning = ""
query_resp_history.append(query_resp)
return JSONResponse(status_code=200, content=query_resp)
if query_resp_history:
return JSONResponse(status_code=200, content=query_resp_history[-1])
return JSONResponse(status_code=404, content={"error": "No answer available"})
async def think_wrapper(interaction, query):
try:
interaction.last_query = query
logger.info("Agents request is being processed")
success = await interaction.think()
if not success:This function is important because it defines how AgenticSeek Tutorial: Local-First Autonomous Agent Operations implements the patterns covered in this chapter.
The think_wrapper function in api.py handles a key part of this chapter's functionality:
return JSONResponse(status_code=404, content={"error": "No answer available"})
async def think_wrapper(interaction, query):
try:
interaction.last_query = query
logger.info("Agents request is being processed")
success = await interaction.think()
if not success:
interaction.last_answer = "Error: No answer from agent"
interaction.last_reasoning = "Error: No reasoning from agent"
interaction.last_success = False
else:
interaction.last_success = True
pretty_print(interaction.last_answer)
interaction.speak_answer()
return success
except Exception as e:
logger.error(f"Error in think_wrapper: {str(e)}")
interaction.last_answer = f""
interaction.last_reasoning = f"Error: {str(e)}"
interaction.last_success = False
raise e
@api.post("/query", response_model=QueryResponse)
async def process_query(request: QueryRequest):
global is_generating, query_resp_history
logger.info(f"Processing query: {request.query}")
query_resp = QueryResponse(
done="false",
answer="",
reasoning="",
agent_name="Unknown",This function is important because it defines how AgenticSeek Tutorial: Local-First Autonomous Agent Operations implements the patterns covered in this chapter.
The process_query function in api.py handles a key part of this chapter's functionality:
@api.post("/query", response_model=QueryResponse)
async def process_query(request: QueryRequest):
global is_generating, query_resp_history
logger.info(f"Processing query: {request.query}")
query_resp = QueryResponse(
done="false",
answer="",
reasoning="",
agent_name="Unknown",
success="false",
blocks={},
status="Ready",
uid=str(uuid.uuid4())
)
if is_generating:
logger.warning("Another query is being processed, please wait.")
return JSONResponse(status_code=429, content=query_resp.jsonify())
try:
is_generating = True
success = await think_wrapper(interaction, request.query)
is_generating = False
if not success:
query_resp.answer = interaction.last_answer
query_resp.reasoning = interaction.last_reasoning
return JSONResponse(status_code=400, content=query_resp.jsonify())
if interaction.current_agent:
blocks_json = {f'{i}': block.jsonify() for i, block in enumerate(interaction.current_agent.get_blocks_result())}
else:This function is important because it defines how AgenticSeek Tutorial: Local-First Autonomous Agent Operations implements the patterns covered in this chapter.
The AgentRouter class in sources/router.py handles a key part of this chapter's functionality:
from sources.logger import Logger
class AgentRouter:
"""
AgentRouter is a class that selects the appropriate agent based on the user query.
"""
def __init__(self, agents: list, supported_language: List[str] = ["en", "fr", "zh"]):
self.agents = agents
self.logger = Logger("router.log")
self.lang_analysis = LanguageUtility(supported_language=supported_language)
self.pipelines = self.load_pipelines()
self.talk_classifier = self.load_llm_router()
self.complexity_classifier = self.load_llm_router()
self.learn_few_shots_tasks()
self.learn_few_shots_complexity()
self.asked_clarify = False
def load_pipelines(self) -> Dict[str, Type[pipeline]]:
"""
Load the pipelines for the text classification used for routing.
returns:
Dict[str, Type[pipeline]]: The loaded pipelines
"""
animate_thinking("Loading zero-shot pipeline...", color="status")
return {
"bart": pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
}
def load_llm_router(self) -> AdaptiveClassifier:
"""
Load the LLM router model.
returns:This class is important because it defines how AgenticSeek Tutorial: Local-First Autonomous Agent Operations implements the patterns covered in this chapter.
flowchart TD
A[get_latest_answer]
B[think_wrapper]
C[process_query]
D[AgentRouter]
E[that]
A --> B
B --> C
C --> D
D --> E