注：以下所提的“文档”，是指LangGraph官方指南（https://langchain-ai.github.io/langgraph/guides/），参考版本是0.6.8。

以下代码的开发环境：

[project]
name = "my-langgraph"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "dotenv>=0.9.9",
    "langchain>=0.3.27",
    "langchain-openai>=0.3.33",
    "langfuse>=3.5.0",
    "langgraph>=0.6.7",
    "langgraph-checkpoint-sqlite>=2.0.11",
    "numpy>=2.3.3",
]

一、工具（Tools）在 LangGraph 中的角色与价值

在许多 Agent / LLM 驱动的系统中，“工具”（如检索、计算、API 调用）是自然语言代理从 prompt → 动作执行的桥梁。LangGraph 也在其体系里对工具做了系统支持。以下是工具在 LangGraph 中的定位与能力：

1. 作为图 / Agent 中可调用的组件
   工具被视为可插入节点（或与 Agent 集成的调用点），在执行图 / Agent 时被触发调用。工具可以产生副作用、调用外部服务、做计算、返回结果等。这样模型或节点就可以“说出”要调用工具、再接收工具结果、再继续流程。
2. 工具与图 / 节点的协作
   在 Graph / StateGraph / Agent 执行中，工具调用可以被节点发起，也可以由 Agent 的 prompt 决定何时调用哪一个工具。工具执行结果写回 state，后续节点可读可用。
3. 统一的工具调用机制 & 流式支持
   文档强调工具调用整合在 LangGraph 的流 (streaming) / checkpoint / context / memory 等体系里：你可以在工具中调用 get_stream_writer() 输出进度 /日志、自定义流；工具可以在异步 /同步环境中被管理。
4. 工具元数据 / 可扩展性
   工具通常带有名称、输入 / 输出签名、元数据标签，便于 Agent 在 prompt /规划时做选择。LangGraph 支持把工具注册到 Agent 或图里供动态选择。

二、工具概念篇（concepts/tools）：主要内容与要点

指南文档中描述的是工具机制的设计概览，讲工具在 LangGraph 中的语义、规范、调用路径、与其他模块（context, memory, persistence 等）的关系。下面是其主要脉络与内容点。

1. 工具（Tool）的定义与接口

• 工具通常被定义为函数 / 可调用对象，接受一些输入（可能是 state / context /参数），返回输出（可能写入 state / 返回结果）。
• 工具可以是同步或异步的（支持 async 函数）。
• 工具可在内部调用 get_stream_writer() 以发出中间流（日志 /进度 /自定义消息）。

2. 工具在 Agent / Graph 执行中的触发

• 在 Agent 里，模型可以通过输出中含有某种“指令 /标识”，告诉 LangGraph “我要调用这个工具（Tool X）”。LangGraph 会根据工具名称 /输入，安排在图 /节点层面执行工具节点。
• Graph 内部也可显式调用工具节点 /子图，把工具作为图的一部分。

3. 流式工具调用 & 中间输出

• 工具内部可以用 get_stream_writer() 向外发送 custom 流（进度 /日志 /内中状态）。如果图 / Agent 启用了 stream_mode="custom"，这些流就会被消费者捕获。
• 在异步环境或写入上下文传播受限时，工具可能需要把 writer 显式传入。文档对此做了说明。

4. 错误处理 / 重试 / 副作用控制

• 工具可能调用外部服务失败或出错，LangGraph 允许捕获工具异常、重试机制 / fallback 机制 /继续或失败的策略。
• 由于工具常伴有副作用（例如写数据库、发送请求），工具设计应考虑幂等性 /重复调用的安全性，以配合可恢复执行（durable execution）机制。

5. 工具与上下文 / memory / persistence 的整合

• 工具可以访问静态 context（通过 get_runtime(...)）以获取用户配置、API key、环境参数等
• 工具可以读写 state（短期 memory）与长期 memory（通过 Store）
• 工具调用时可能产生新的状态 /记忆 /中间 artifact，这些都融入 checkpoint / persistence 体系
• 工具在中断 /恢复 / replay 场景下，应避免不可恢复的副作用重复执行，或应设计幂等操作。

三、Tool Calling 教程篇（how-tos/tool-calling）：操作指南与典型示例

在指南文档中的Tool Calling教程更偏重实践操作，教你在 LangGraph / Agent / Graph 中如何定义工具、如何在 prompt /节点里调用、怎样消费工具结果等。下面是它的主要内容与典型用法示意。

1. 定义工具 / 装饰工具

• 使用 @tool 装饰器将一个函数标记为可被调用工具。例如：

@tool
def lookup_wikipedia(query: str) -> str:
    ...

• 这个工具函数应定义明确的输入 / 输出类型，方便 Agent / Graph 在 prompt 或节点里调用、验证。

2. 在 Agent / Graph 中使用工具

• 在 Agent 构建时传入工具列表（tools=[lookup_wikipedia, another_tool]）
• Agent 在与模型 / prompt 交互时，模型可以输出 “call tool_name(...)” 的意图
• LangGraph 捕获这个意图，将其映射为工具节点执行，并把执行结果注入 state / prompt 下一步使用

3. 在节点 / 子图里显式调用工具

• 除了 Agent 层面，Graph / node 函数内部也可以直接调用工具函数（就像普通函数一样），例如：

def node_search(state: State):
    res = lookup_wikipedia(state["topic"])
    return {"wiki": res}

• 这样工具就成为图的一部分，流程更加透明化、可组合。

4. 流式工具结果 / 中间日志

• 如果工具内部需要在执行过程中输出日志 /进度，可以使用 get_stream_writer()，如：

@tool
def long_api(q):
    writer = get_stream_writer()
    writer({"step": 1})
    ... do part 1 ...
    writer({"step": 2})
    ...
    return result

用户若开启 stream_mode="custom"，这些 writer 输出会被捕获展示。

5. 异常 / fallback / 重试机制

• 文档建议给工具设计失败重试 / fallback 机制，以在外部服务异常时保持健壮性
• 在工具中捕获异常并返回错误结构（而不是抛出）是一个常见模式，使执行流不会中断整个图

6. 工具调用的元数据 / 标签 / 策略控制

• Agent / 模型可以根据工具的标签 /级别 /成本 /可靠性做选择（prompt 给模型知道哪些工具可用 /代价）
• 工具可以有元数据（如 cost_estimate, timeout, required_permissions 等）以供调度 /安全控制

四、对比 / 综合理解 + 使用建议

下面是基于Tools文档 + 实践经验的理解、建议与注意事项。

综合理解

• Tools 在 LangGraph 是关键接口，把语言推理与外部操作 /计算 /API 整合起来，是驱动智能系统能力的“外部动作”层。
• LangGraph 的设计让工具既能在 Agent 层被 prompt 驱动调用，也能在 Graph / node 内部以函数方式调用，两种方式互补。
• 工具调用与流式输出、状态保存 (checkpoint)、上下文 (context)、memory 等体系深度整合，使得工具不仅是“黑盒调用”，而是可观测、可恢复、可组合的节点。

使用建议 / 注意事项

1. 设计工具为幂等 / 可重入安全
   由于可能在 recovery / replay / retry 场景被重复触发，工具调用最好设计成幂等或有检测机制避免重复执行破坏。
2. 避免把不可控副作用集中在工具里
   如果工具有外部影响（写数据库、修改状态等），尽量把这些影响放在可以回滚 /恢复 /幂等控制的结构中。
3. 合理使用 get_stream_writer() 输出中间日志 / 进度
   对于耗时、分阶段的工具，建议用 writer 输出进度，这样前端 /控制台可以实时看到进展。
4. 工具异常处理 / fallback
   工具内部最好捕获异常、包装为错误输出，而不是直接抛光，以免打断整个图 / Agent 执行。或外层包 fallback 工具。
5. 工具元数据 / 可选调用控制
   给工具加上标签、成本估计、时间复杂度信息，以便在 prompt / Agent 策略层做选择（比如“优先调用 cheap 工具”）。
6. Tool 与 context / memory / state 协作
   • 工具内部可访问 runtime.context 获取静态配置 /参数
   • 工具可读写 state（短期 memory）或长期 memory（Store）
   • 要关心工具调用结果如何影响后续节点 / prompt 构造
7. 性能 / 并发 /限流考虑
   多个工具、多个并发调用时要考虑外部 API 限额 /响应时间 /并发控制。工具内部或模型层可插入 RateLimiter、timeout、并发队列等。

五、例子代码

import os, json, re, traceback
from typing import TypedDict, Dict, Any, List, Callable, Literal, Optional
from dataclasses import dataclass
from openai import OpenAI

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# ============== 0) LLM Client ==============
client = OpenAI(api_key="abc", base_url="http://192.168.0.8:1294/v1")
MODEL = "qwen"

# ============== 1) Graph State（强 schema） ==============
class GraphState(TypedDict, total=False):
    question: str
    route: Literal["tool_hub", "direct"]
    # LLM tool hub 只维护 messages/assistant_text/need_more
    messages: List[Dict[str, Any]]
    assistant_text: str
    need_more: bool
    # 最终答案由 Assembler 统一写入
    final_answer: str

# ============== 2) 工具实现（与 Schema 解耦） ==============
def tool_calculator(expression: str) -> Dict[str, Any]:
    nums = list(map(float, re.findall(r"[-+]?\d*\.?\d+", expression)))
    if not nums:
        return {"ok": True, "result": 0.0}
    if "*" in expression:
        res = 1.0
        for n in nums:
            res *= n
        return {"ok": True, "result": res}
    return {"ok": True, "result": sum(nums)}

def tool_retrieval(query: str, top_k: int = 3) -> Dict[str, Any]:
    hits = [
        {"id": "D1", "title": "LangGraph 入门", "snippet": "StateGraph 与节点..."},
        {"id": "D2", "title": "RAG 检索重排", "snippet": "BM25 + 向量融合..."},
        {"id": "D3", "title": "MCP 工具化", "snippet": "注册工具/调用..."},
    ]
    return {"ok": True, "hits": hits[:top_k]}

def tool_web_search(query: str) -> Dict[str, Any]:
    return {
        "ok": True,
        "top": {
            "title": "Official LangGraph Docs",
            "url": "https://langchain-ai.github.io/langgraph/",
            "summary": "Docs about StateGraph, persistence, commands..."
        }
    }

TOOL_IMPLS: Dict[str, Callable[..., Dict[str, Any]]] = {
    "tool_calculator": tool_calculator,
    "tool_retrieval": tool_retrieval,
    "tool_web_search": tool_web_search,
}

# ============== 3) 工具 Schema（给 LLM） ==============
TOOLS: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "tool_calculator",
            "description": "进行基础算术计算，支持加法与乘法。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "算式，如 '12 + 3 * 4'"},
                },
                "required": ["expression"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "tool_retrieval",
            "description": "从知识库检索相关文档片段。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "top_k": {"type": "integer", "minimum": 1, "maximum": 20, "default": 3},
                },
                "required": ["query"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "tool_web_search",
            "description": "执行网页搜索（示例）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                },
                "required": ["query"],
                "additionalProperties": False,
            },
        },
    },
]

SYSTEM_PROMPT = """你是一个能选择并组合工具完成任务的助手。
- 当需要事实/数据时，用检索或网页搜索。
- 当问题包含运算，先用计算器工具得到结果再说明过程。
- 工具返回 JSON，请把关键结果整理进最终回答里。
- 如果一次需要多个工具，你可以在同一条回复里下发多个 tool_calls。
- 若工具返回失败(ok=false)，请换策略或解释原因。"""

# ============== 4) Router（只决定走不走工具） ==============
@dataclass
class SimpleRouterLLM:
    def decide(self, q: str) -> Literal["tool_hub", "direct"]:
        ql = q.lower()
        if any(k in ql for k in ["查", "资料", "检索", "web", "网页", "官网", "sum", "+", "*", "计算"]):
            return "tool_hub"
        return "direct"

router_llm = SimpleRouterLLM()

def node_router(state: GraphState) -> GraphState:
    route = router_llm.decide(state["question"])
    return {"route": route}

# ============== 5) LLM Tool Hub（只做 tool_calls 回合） ==============
def node_llm_tool_hub(state: GraphState) -> GraphState:
    messages = state.get("messages") or [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": state["question"]},
    ]

    # 回合 1：规划 + 可能的多工具调用
    resp = client.chat.completions.create(
        model=MODEL, messages=messages, tools=TOOLS, tool_choice="auto", temperature=0.0
    )
    msg1 = resp.choices[0].message

    if msg1.tool_calls:
        messages.append({
            "role": "assistant",
            "content": msg1.content or None,
            "tool_calls": [tc.model_dump() if hasattr(tc, "model_dump") else tc for tc in msg1.tool_calls],
        })
        # 执行工具
        for tc in msg1.tool_calls:
            name = tc.function.name
            call_id = tc.id
            try:
                args = json.loads(tc.function.arguments or "{}")
            except Exception:
                args = {}
            impl = TOOL_IMPLS.get(name)
            if not impl:
                out = {"ok": False, "error": f"Unknown tool: {name}"}
            else:
                try:
                    out = impl(**args)
                except Exception as e:
                    out = {"ok": False, "error": str(e), "trace": traceback.format_exc()}

            messages.append({
                "role": "tool",
                "tool_call_id": call_id,
                "name": name,
                "content": json.dumps(out, ensure_ascii=False),
            })

        # 回合 2：带工具结果让模型给出自然语言
        resp2 = client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS, tool_choice="auto", temperature=0.0
        )
        msg2 = resp2.choices[0].message

        if msg2.tool_calls:
            # 若模型还要继续规划，这里标记 need_more，让上层选择：直接 END 等待人工/下次 resume
            messages.append({
                "role": "assistant",
                "content": msg2.content or None,
                "tool_calls": [tc.model_dump() if hasattr(tc, "model_dump") else tc for tc in msg2.tool_calls],
            })
            return {"messages": messages, "assistant_text": msg2.content or "", "need_more": True}

        assistant_text = msg2.content or ""
        messages.append({"role": "assistant", "content": assistant_text})
        return {"messages": messages, "assistant_text": assistant_text, "need_more": False}

    # 无 tool_calls：直接回答
    assistant_text = msg1.content or ""
    messages.append({"role": "assistant", "content": assistant_text})
    return {"messages": messages, "assistant_text": assistant_text, "need_more": False}

# ============== 6) Assembler（只读状态组装最终答案） ==============
def node_assemble(state: GraphState) -> GraphState:
    text = state.get("assistant_text") or ""
    # 这里可以统一附加格式、加来源、裁剪等；保持职责：不调用工具
    final_answer = text.strip() or "（没有可用答案）"
    return {"final_answer": final_answer}

# ============== 7) 组图 ==============
def build_graph():
    g = StateGraph(GraphState)

    g.add_node("router", node_router)
    g.add_node("tool_hub", node_llm_tool_hub)
    g.add_node("assemble", node_assemble)

    g.set_entry_point("router")

    # Router → 分支
    g.add_conditional_edges(
        "router",
        lambda st: st["route"],
        {
            "tool_hub": "tool_hub",
            "direct": "assemble",
        },
    )

    # Tool Hub →（需要更多工具？）→ 直接结束或进入 Assembler
    g.add_conditional_edges(
        "tool_hub",
        lambda st: "need_more" if st.get("need_more") else "done",
        {
            "need_more": END,     # 简化处理：交给下一次 resume（input=None）继续推进
            "done": "assemble",
        },
    )

    g.add_edge("assemble", END)

    return g.compile(checkpointer=MemorySaver())

# ============== 8) 演示运行 ==============
def main():
    graph = build_graph()
    cfg = {"configurable": {"thread_id": "multi-node-thread-1"}}

    print("\nQ1) 12 + 30 * 2 等于多少？（将走 tool_hub）")
    last = None
    for ev in graph.stream({"question": "12 + 30 * 2 等于多少？"}, config=cfg):
        for node, payload in ev.items():
            last = payload
            print(f"  [{node}] -> {list(payload.keys())}")
    print("A1:", last.get("final_answer"))

    print("\nQ2) 给我 LangGraph 的入门资料并找官网地址。（多工具）")
    last = None
    for ev in graph.stream({"question": "给我 LangGraph 的入门资料并找官网地址。"}, config=cfg):
        for node, payload in ev.items():
            last = payload
            print(f"  [{node}] -> {list(payload.keys())}")
    print("A2:", last.get("final_answer"))

    # 恢复演示：若上一次 tool_hub 标记了 need_more=True，则可用 input=None 继续
    print("\n[恢复] 同一 thread_id + input=None 继续推进上一轮（如仍有未执行的 tool_calls）")
    last = None
    for ev in graph.stream(input=None, config=cfg):
        for node, payload in ev.items():
            last = payload
            print(f"  [{node}] -> {list(payload.keys())}")
    print("A[resume]:", last.get("final_answer"))

def tool_calls_multi_node_example():
    main()