I developed a coding assistant that leverages multiple, local large language models.
Each model is assigned a role (i.e., Supervisor, Researcher, Coder, or Critic). The Supervisor is the brains of the operation in that it decides which agent is called next. It typically begins by calling the Researcher. The Researcher is responsible for researching how to build the project referenced in the user query. This system is agentic in the sense that the researcher is capable of scrapping the web for useful information. Next, the Critic analyzes the Researcher's findings and breaks it down into implementable steps. Then the Coder writes the actual code. It's up to the Supervisor to decide whether to continue cycling through these agents or conclude the process.
I chose to employ models that were trained on unique datasets to fulfill specific purposes (i.e., coding, long-term reasoning, etc.). For instance, ornith:9b is marketed as a agentic coding assistant on it's own therefore I use it as the coding agent in this system.
# Models
researcher_model = ChatOllama(model="llama3.1:8b")
coder_model = ChatOllama(model="ornith:9b")
critic_model = ChatOllama(model="mistral:7b")
supervisor_model = ChatOllama(model="dolphin3:8b", temperature=0)
Additionally, I designed the system to route tasks automatically. In other words, there is no human in this loop! The program terminates whenever the Supervisor agent is satisfied with the output.
def supervisor_node(state: AgentState):
orig = state["messages"][-1].content.split()
res = " ".join(orig)
response = supervisor_model.invoke(
[SystemMessage(content=SUPERVISOR_PROMPT), str(res)]
)
valid_choices = {"Researcher", "Coder", "Critic"}
content = getattr(response, "content", "") if response else ""
matches = re.findall(r"\b(?:Researcher|Coder|Critic)\b", content)
choice = matches[0] if len(matches) == 1 and matches[0] in valid_choices else "FINISH"
return {
"messages": [
AIMessage(content=f"## Routing to {choice}.", name="Supervisor")
],
"next": choice,
}
# Routing after the supervisor chooses a destination.
def route_from_supervisor(state: AgentState) -> str:
return state["next"]
# Routing after a worker: execute requested tools, or return to supervisor.
def route_after_worker(state: AgentState) -> str:
last_message = state["messages"][-1]
if isinstance(last_message, AIMessage) and last_message.tool_calls:
return "tools"
return "Supervisor"
# Build graph
graph = StateGraph(AgentState)
graph.add_node("Supervisor", supervisor_node)
graph.add_node("Researcher", researcher)
graph.add_node("Coder", coder)
graph.add_node("Critic", critic)
graph.add_node("tools", tool_node)
graph.set_entry_point("Supervisor")
graph.add_conditional_edges(
"Supervisor",
route_from_supervisor,
{
"Researcher": "Researcher",
"Coder": "Coder",
"Critic": "Critic",
"FINISH": END,
},
)
This was my introduction into agentic systems and I'm beyond amazed. I can only imagine what larger models are capable of; it's almost frightening. I'm content experimenting with these little models for now.
Copyright © Lucas Thormann