G1-03 · 手写 StateGraph 与条件边：ReAct 循环自己搭

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)                       # 1. 拿一张空图
builder.add_node("agent", agent_node)             # 2. 摆节点
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")                  # 3. 定入口
builder.add_conditional_edges("agent", router, ["tools", END])   # 4. 条件边
builder.add_edge("tools", "agent")                #    工具完回来接着想
graph = builder.compile()                         # 5. 编译成可执行图
graph.invoke({"messages": [...]})                 #    开跑

import json
def router(state: State) -> str:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):         # 模型要求调工具？
        return "tools"
    return "__end__"                              # 说人话了 → 结束

# g1_handmade_react.py —— 不用 create_agent 的 ReAct 循环
import os
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import AnyMessage, ToolMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

@tool
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """两数相乘"""
    return a * b
TOOLS = {t.name: t for t in [add, multiply]}

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

llm = ChatOpenAI(model=os.environ.get("LG_MODEL", "Qwen/Qwen3.5-4B"),
                 api_key=os.environ["SILICONFLOW_API_KEY"],
                 base_url="https://api.siliconflow.cn/v1", temperature=0).bind_tools(list(TOOLS.values()))

def agent_node(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

def tool_node(state: State) -> dict:
    last = state["messages"][-1]
    out = []
    for call in last.tool_calls:                          # 可能点多个菜
        result = TOOLS[call["name"]].invoke(call["args"])
        out.append(ToolMessage(content=str(result), tool_call_id=call["id"]))
    return {"messages": out}

def router(state: State) -> str:
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(State)
builder.add_node("agent", agent_node); builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", router)
builder.add_edge("tools", "agent")

graph = builder.compile()
result = graph.invoke({"messages": [HumanMessage(content="请计算 (3+5)×12")]})
for m in result["messages"]:
    print(f"[{m.__class__.__name__}] {m.content or '(工具调用)'}")