Function build_nodes

workflow/nodes.py:197–238

(
    *,
    llm: "BaseChatModel",
    prompts: PromptBundle,
    tool_specs: list[ToolSpec],
    context: WorkflowContext,
)

def build_nodes(
    *,
    llm: "BaseChatModel",
    prompts: PromptBundle,
    tool_specs: list[ToolSpec],
    context: WorkflowContext,
) -> NodeSet:
    try:
        from langchain.agents import create_react_agent
        from langchain_core.prompts import PromptTemplate
    except ImportError as exc:
        raise RuntimeError(
            "LangChain is required to build workflow nodes."
        ) from exc

    scan_tools = build_langchain_tools([tool for tool in tool_specs if tool.name == "EXECMD"])
    inquire_tools = build_langchain_tools([tool for tool in tool_specs if tool.name == "ReadHTML"])
    exploit_tools = build_langchain_tools([tool for tool in tool_specs if tool.name == "EXECMD"])

    scan_agent = create_react_agent(
        llm=llm,
        tools=scan_tools,
        prompt=PromptTemplate.from_template(prompts.scan),
    )
    inquire_agent = create_react_agent(
        llm=llm,
        tools=inquire_tools,
        prompt=PromptTemplate.from_template(prompts.inquire),
    )
    exploit_agent = create_react_agent(
        llm=llm,
        tools=exploit_tools,
        prompt=PromptTemplate.from_template(prompts.exploit),
    )

    return NodeSet(
        scan=partial(agent_node, context=context, agent=scan_agent, tools=scan_tools, sender_name="Scan"),
        inquire=partial(agent_node, context=context, agent=inquire_agent, tools=inquire_tools, sender_name="Inquire"),
        exploit=partial(agent_node, context=context, agent=exploit_agent, tools=exploit_tools, sender_name="Exploit"),
        vuln_select=partial(vuln_select_node, context=context, sender_name="Vuln_select"),
        check=partial(check_node, context=context, sender_name="Check"),
    )