Dựng ReAct agent với LangGraph cho tool mạng

uv add langgraph langchain-anthropic
export ANTHROPIC_API_KEY='***'

import ipaddress
import subprocess
 
from langchain_core.tools import tool
 
INVENTORY = {
    "pe1-hni": {"mgmt": "198.51.100.11", "os": "iosxr"},
    "pe2-hcm": {"mgmt": "198.51.100.12", "os": "junos"},
}
SHOW_BGP = {"iosxr": "show bgp summary", "junos": "show bgp summary"}
 
 
@tool
def ping(target: str, count: int = 3) -> str:
    """Ping một địa chỉ IP (không nhận hostname). Dùng để kiểm tra reachability."""
    ip = ipaddress.ip_address(target)
    out = subprocess.run(["ping", "-c", str(min(count, 10)), "-W", "1", str(ip)],
                         capture_output=True, text=True, timeout=30)
    return out.stdout[-1500:] or out.stderr
 
 
@tool
def show_bgp_summary(device: str) -> str:
    """Chạy 'show bgp summary' (chỉ đọc) trên thiết bị theo tên inventory, vd pe1-hni."""
    d = INVENTORY.get(device)
    if d is None:
        return f"Không có thiết bị {device}. Thiết bị hợp lệ: {', '.join(INVENTORY)}"
    return run_show(d["mgmt"], d["os"], SHOW_BGP[d["os"]])
 
 
def run_show(host: str, os_name: str, command: str) -> str:
    # Gọi netmiko/scrapli/PyEZ ở đây; chỉ cho phép lệnh show
    if not command.startswith("show "):
        raise ValueError("chỉ cho phép lệnh show")
    raise NotImplementedError

from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
 
from net_tools import ping, show_bgp_summary
 
model = ChatAnthropic(model="claude-sonnet-5-5", temperature=0)
agent = create_react_agent(
    model,
    tools=[ping, show_bgp_summary],
    prompt="Bạn là trợ lý NOC. Chỉ dùng tool đọc, trả lời ngắn, nêu rõ lệnh đã chạy.",
    checkpointer=InMemorySaver(),
)
 
config = {"configurable": {"thread_id": "noc-1"}, "recursion_limit": 20}
result = agent.invoke({"messages": [("user", "pe1-hni có ping được không?")]}, config)
print(result["messages"][-1].content)

from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
 
from net_tools import ping, show_bgp_summary
 
tools = [ping, show_bgp_summary]
model = ChatAnthropic(model="claude-sonnet-5-5", temperature=0).bind_tools(tools)
 
 
def agent(state: MessagesState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}
 
 
builder = StateGraph(MessagesState)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)  # có tool_calls -> "tools", không -> END
builder.add_edge("tools", "agent")
graph = builder.compile(checkpointer=InMemorySaver())
 
config = {"configurable": {"thread_id": "noc-1"}, "recursion_limit": 20}
for chunk in graph.stream({"messages": [("user", "BGP trên pe1-hni có peer nào down?")]},
                          config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

uv run agent_graph.py