13 - Case Study: Autonomous Research Assistant

The problem: putting the pieces together without telling a story that does not exist

After twelve articles on the individual ingredients (loops, tools, memory, orchestration, tests, security, costs) the natural question is how they fit together in a system that does a job. The most convenient answer would be a "real case" with times, costs and success percentages. This article is not that: it is a constructed example, and I say so right away. The sources are invented (the town of Valdoro does not exist), no model is called, and there are no production figures because there is no production. What the example can show, and does show, is the structure: where the state lives, where a loop stops, where a check in code keeps the system from making things up.

The task is a research assistant with three roles. A researcher retrieves sources relevant to the question, an analyst extracts the requested value from them and weighs how much the sources agree, an editor writes the report; a fourth node, the verifier, rejects any citation that is not in the state. If the sources are not enough, the system widens the search once and then admits it does not know. The listings were run with LangGraph 1.2.21 (from PyPI, in an environment outside the repository), twice with identical output.

First of all: do you really need more than one agent?

Anthropic distinguishes between workflows, systems in which models and tools are orchestrated through predefined code paths, and agents, in which the model dynamically directs its own process. And it recommends looking for the simplest solution and increasing complexity only when needed, which may mean not building an agentic system at all. The system in this article is, strictly speaking, a workflow: the order of the nodes is fixed and the only loop has a cap. The autonomy would sit inside the nodes, when a model decided which sub-questions to search for (we saw this in the articles on ReAct and on tool calling).

Why three roles, then, and not one function? Because in a single function you can tell apart neither the permissions nor the failures. Splitting makes sense when:

• the tools are different: whoever reads the web does not need to write reports, and whoever writes the report does not need the network;
• you want to evaluate the pieces separately: if the report is wrong, you must be able to say whether the one who searched, the one who extracted or the one who wrote got it wrong;
• the boundary is a checkpoint: between analyst and editor passes structured data that the code can verify.

None of these reasons is "every agent has a name and a personality". And there are costs to account for: in its data Anthropic reports that multi-agent systems use about 15 times the tokens of a chat (against about 4 times for a single agent), and warns that tasks with many dependencies among the agents are not today a good ground for these systems. For a simple task, three roles are an addition of cost and of points of failure; the article on orchestration discusses this.

State as a contract: sources, retrieval and shared data

The first fragment defines the data we work with and the way of retrieving it. The five sources are invented sentences, with a type (agency or blog) that later becomes a weight. Retrieval is deliberately crude: it counts the words in common between question and text, and considers a source relevant only if it has at least two. It is not a search engine: it is the stand-in for a tool that in the real system would be a search API or a vector index (see the article on memory), and it is deterministic, so the run can be repeated.

import operator
import re
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

# ---- part 1: the sources (INVENTED: the town of Valdoro does not exist) and retrieval
SOURCES = {
    "F1": ("agency", "Water balance of Valdoro: water network losses are 31 percent of the water supplied"),
    "F2": ("agency", "Technical report on the water network of Valdoro: losses at 31 percent in 2023"),
    "F3": ("blog", "In Valdoro the water network loses about 28 percent, according to our calculations"),
    "F4": ("agency", "Green plan of Valdoro: new pedestrian areas"),
    "F5": ("agency", "District heating tariffs in Valdoro"),
}
WEIGHT = {"agency": 2, "blog": 1}


STOPWORDS = {"of", "the", "in", "a", "and", "are", "is", "on", "for", "to"}


def words(text):
    return set(re.findall(r"[a-z]+", text.lower())) - STOPWORDS


def retrieve(question, k):
    """The k sources sharing the most words with the question; below 2 words it is not relevant."""
    q = words(question)
    scores = sorted(((-len(q & words(t)), id_) for id_, (_, t) in SOURCES.items()))
    return [id_ for p, id_ in scores if -p >= 2][:k]

The STOPWORDS set is not a detail: without it the question about the condominium's tariffs retrieves the sources on water and on greenery, because they share with the question little more than "of", "the" and the name of the town (measured by running the listing without the filter). It is a small and typical mistake: retrieval by overlap without a stop-word filter answers with irrelevant sources, and an analyst downstream treats them as relevant. Real retrieval has the same risk in other forms, and it is a good reason to test the researcher on its own.

The shared state and the nodes that work

The state is a TypedDict: each node receives the whole state and returns only the updates. LangGraph's documentation explains that updates go through a reducer for each field: by default the last write wins, while with a reducer the values are combined, for example by being appended to a list. In the listing only warnings has the reducer (operator.add), because it is the only field to which several nodes can append without erasing the others' notices.

# ---- part 2: the shared state and the two nodes that work on the data
class State(TypedDict):
    question: str
    rounds: int
    sources: list
    values: dict
    enough: bool
    report: str
    warnings: Annotated[list, operator.add]   # nodes append, they do not overwrite


MAX_ROUNDS = 2
MIN_SOURCES = 3


def researcher(state):
    rounds = state["rounds"] + 1
    sources = retrieve(state["question"], k=2 * rounds)   # the net widens at every round
    return {"rounds": rounds, "sources": sources}


def analyst(state):
    values = {}
    for id_ in state["sources"]:
        kind, text = SOURCES[id_]
        found = re.search(r"(\d+) percent", text)
        if found:
            values.setdefault(int(found.group(1)), []).append(id_)
    return {"values": values, "enough": len(state["sources"]) >= MIN_SOURCES}


def after_analysis(state):
    return "editor" if state["enough"] or state["rounds"] >= MAX_ROUNDS else "researcher"

Three choices to read. The researcher widens the net at every round (k = 2 * rounds): at the first round it takes two sources, at the second four. The analyst extracts the value with a regular expression and sets the sufficiency rule: at least three relevant sources are needed, otherwise enough is false. And after_analysis decides whether to go forward or back: the MAX_ROUNDS cap sits in my code, because LangGraph's recursion limit exists but counts graph steps (from version 1.1.4 on the documentation gives 10,007 super-steps as the default) and is not meant as a search cap; the article on LangChain and LangGraph shows when it fires.

The editor writes, the verifier checks

The report is written by the editor, but the rule that matters is not written in its prompt: it is a code node that compares the citations in the text with the sources present in the state. This is the principle of the article on security applied to a small case: a check that does not depend on the model's goodwill. Here the editor is a function, so it does not make things up; to see the verifier at work there is also a lying_editor that cites a nonexistent source.

# ---- part 3: the editor writes, the verifier checks the citations
def editor(state):
    if not state["values"] or not state["enough"]:
        return {"report": f"Insufficient evidence (search rounds: {state['rounds']}): no conclusion.",
                "warnings": ["report without conclusion"]}
    weights = {v: sum(WEIGHT[SOURCES[i][0]] for i in ids) for v, ids in state["values"].items()}
    winner = max(weights, key=weights.get)
    confidence = weights[winner] / sum(weights.values())
    lines = [f"Network losses: {winner}% [{', '.join(state['values'][winner])}], "
             f"confidence {confidence:.2f}."]
    for v, ids in state["values"].items():
        if v != winner:
            lines.append(f"Dissenting value: {v}% [{', '.join(ids)}].")
    return {"report": " ".join(lines)}


def lying_editor(state):
    return {"report": "Network losses: 12% [F9]."}


def verifier(state):
    cited = set(re.findall(r"F\d+", state["report"]))
    invented = cited - set(state["sources"])
    if invented:
        raise ValueError(f"citations not present in the state: {sorted(invented)}")
    return {}

In the honest editor the confidence is the weight of the agreeing sources over the total weight, with weights 2 for the agency and 1 for the blog: with two agencies at 31% and a blog at 28% the confidence is 4 out of 5. It is a convention of mine, not a measure of truth; its usefulness is that the report declares the dissenting value instead of hiding it. If the evidence is not enough the editor does not write a conclusion: it writes that evidence is missing, and the warnings field flags it to whoever reads the outcome.

The graph and the three trials

The last fragment assembles the graph and launches it on two questions and on an editor that lies. The conditional edge receives the decision function and the list of possible nodes; the fixed edges give the order. The two questions are chosen because they exercise different paths: the first requires a second round to reach three sources, the second does not have enough sources and must stop.

# ---- part 4: the graph and three trials
def build(editor_node):
    g = StateGraph(State)
    g.add_node("researcher", researcher)
    g.add_node("analyst", analyst)
    g.add_node("editor", editor_node)
    g.add_node("verifier", verifier)
    g.add_edge(START, "researcher")
    g.add_edge("researcher", "analyst")
    g.add_conditional_edges("analyst", after_analysis, ["editor", "researcher"])
    g.add_edge("editor", "verifier")
    g.add_edge("verifier", END)
    return g.compile()


def start(question):
    return {"question": question, "rounds": 0, "sources": [], "values": {}, "enough": False,
            "report": "", "warnings": []}


if __name__ == "__main__":
    honest = build(editor)
    for question in ("water network losses of Valdoro", "tariffs of the condominium of Valdoro"):
        end = honest.invoke(start(question))
        print(f"{question!r}: rounds={end['rounds']} sources={end['sources']}")
        print("  ", end["report"], end["warnings"])

    try:
        build(lying_editor).invoke(start("water network losses of Valdoro"))
    except ValueError as error:
        print("lying editor ->", error)

# Output:
# 'water network losses of Valdoro': rounds=2 sources=['F1', 'F2', 'F3']
#    Network losses: 31% [F1, F2], confidence 0.80. Dissenting value: 28% [F3]. []
# 'tariffs of the condominium of Valdoro': rounds=2 sources=['F5']
#    Insufficient evidence (search rounds: 2): no conclusion. ['report without conclusion']
# lying editor -> citations not present in the state: ['F9']

In the output, the first question takes two rounds, finds three sources, chooses 31% and reports 28% as dissenting. The second reaches the cap with a single source and the report says there is no conclusion: it is the honest behavior, and in systems with a model it is the first to be lost, because a model tends to produce an answer anyway. The last line shows the verifier stopping the invented citation. The exception interrupts the run: in a real system you decide whether to relaunch the editor, hand over to a person or return an error.

Testing the system: outcome, trajectory, regression

A multi-node system fails at different points, so the checks are written on the path and not only on the outcome, as in the article on evaluation. The listing has three trials, one for the behavior that must be there, one for what must be missing and one for the check that must fire. Then it simulates a regression: the round cap brought down to one, a plausible change made "to save money" (see the article on costs).

import research
from research import build, editor, lying_editor, start

WITH_EVIDENCE = "water network losses of Valdoro"
WITHOUT_EVIDENCE = "tariffs of the condominium of Valdoro"


def answers_with_sources():
    end = build(editor).invoke(start(WITH_EVIDENCE))
    assert "31%" in end["report"], end["report"]
    assert len(end["values"][31]) >= 2


def admits_not_knowing():
    end = build(editor).invoke(start(WITHOUT_EVIDENCE))
    assert "no conclusion" in end["report"]
    assert end["warnings"] == ["report without conclusion"]


def no_invented_citations():
    try:
        build(lying_editor).invoke(start(WITH_EVIDENCE))
    except ValueError:
        return
    raise AssertionError("the invented citation was not stopped")


TRIALS = [answers_with_sources, admits_not_knowing, no_invented_citations]


def run_all(title):
    print(title)
    for trial in TRIALS:
        try:
            trial()
            outcome = "ok"
        except AssertionError as error:
            outcome = f"FAILS ({str(error)[:48]})"
        print(f"  {trial.__name__:28} {outcome}")


run_all("code as it is")
research.MAX_ROUNDS = 1      # simulated regression: a single search round
run_all("with MAX_ROUNDS = 1")

# Output:
# code as it is
#   answers_with_sources         ok
#   admits_not_knowing           ok
#   no_invented_citations        ok
# with MAX_ROUNDS = 1
#   answers_with_sources         FAILS (Insufficient evidence (search rounds: 1): no con)
#   admits_not_knowing           ok
#   no_invented_citations        ok

The answers_with_sources trial fails with the cap at one round: the system, with only two sources, admits it does not know, and a saving on the number of rounds changed the quality of the result without any visible error. This is the reason the trials sit next to the code and run on every change. The other two stay green, and this is as informative as the red: it says the regression touches the researcher and not the checks. The trials are simple functions, not a test framework.

What would change with real models

The listing deliberately removes everything that is non-deterministic. With a model in place of the functions, each node acquires a risk that the other articles of the series have already described: the table says which one, and where to go and reread.

The editor's row is the one not to forget: the verifier is the only part of the system that stays valid whatever model writes. When you replace the functions with calls to the model, keep the tests on outcomes and on the verifier as they are, and add the repeated evaluation of the ninth article: the same task run several times, because the result will no longer be the same at every trial. To see how a single agent is built on LangGraph I refer you to the article on the framework.

Frequently asked questions

What you can do now

You can structure a system of roles as a graph with a declared state, a loop with a cap of your own and a check node that does not depend on the model. You can make the system admit it has no evidence, write trials on the outcome, on the missing behavior and on the check, and recognize a regression that no exception reports. You can also answer the question upstream: a workflow with a loop is enough, and separate roles are justified by different permissions, evaluation and checkpoints, not by the number of agents. The concluding chapter, The Future of AI Agents: Emergent Capabilities and Limitations, takes up the criteria for deciding whether a task deserves an agent.