name	langgraph-custom-tool-calling
description	在LangGraph中使用ToolNode时保留自定义工具调用构造方式。当模型不支持标准bind_tools()、需要手动解析流式响应中的tool_calls、或要从手写工具执行逻辑迁移到ToolNode时使用。

LangGraph 自定义工具调用集成

核心约束

ToolNode 的 invoke() 内部用 isinstance(m, AIMessage) 查找最新 assistant 消息。plain dict 不会被识别，必须返回 AIMessage 实例。

完整模式：自定义 LLM + ToolNode

1. llm_node 返回 AIMessage（不是 dict！）

import json
from langchain_core.messages import AIMessage

def _llm_node(state):
    messages = state["messages"]

    # history 可能含 AIMessage/ToolMessage，转为 dict 给 OpenAI SDK
    api_messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for m in messages:
        if isinstance(m, dict):
            api_messages.append(m)
        else:
            d = {"role": getattr(m, "role", "assistant"),
                 "content": getattr(m, "content", "") or ""}
            tc = getattr(m, "tool_calls", None)
            if tc:
                # ⚠️ AIMessage 存的是 LangChain 格式，必须转回 OpenAI 格式
                d["tool_calls"] = [
                    {"id": t["id"], "type": "function",
                     "function": {"name": t["name"],
                                  "arguments": json.dumps(t.get("args", {}), ensure_ascii=False)}}
                    for t in tc
                ]
            tcid = getattr(m, "tool_call_id", None)
            if tcid:
                d["tool_call_id"] = tcid
            rkc = getattr(m, "additional_kwargs", {}).get("reasoning_content")
            if rkc:
                d["reasoning_content"] = rkc
            api_messages.append(d)

    # --- 自定义流式调用（保持你的 reasoning_content、stream 逻辑） ---
    stream = your_llm_stream(api_messages, tools_schema)
    content, tool_calls_raw, reasoning = parse_stream(stream)  # 你的解析逻辑

    # --- OpenAI 格式 → LangChain 格式 ---
    lc_tool_calls = []
    for tc in tool_calls_raw:
        args_str = tc.get("function", {}).get("arguments", "{}")
        try:
            args = json.loads(args_str)
        except Exception:
            args = {"raw": args_str}
        lc_tool_calls.append({
            "name": tc["function"]["name"],
            "args": args,
            "id": tc.get("id", ""),
            "type": "tool_call",
        })

    ai_msg = AIMessage(content=content, tool_calls=lc_tool_calls)
    if reasoning:
        ai_msg.additional_kwargs["reasoning_content"] = reasoning

    return {"messages": [*messages, ai_msg]}

2. 条件边兼容两种消息类型

from langchain_core.messages import AIMessage

def _should_continue(state):
    last = state["messages"][-1]
    if isinstance(last, AIMessage):
        return "tools" if last.tool_calls else END
    return "tools" if last.get("tool_calls") else END

3. 兼容辅助函数（dict ↔ LangChain 混合 history）

history 会同时包含 plain dict（user/tool 消息）和 AIMessage/ToolMessage，需要用兼容函数访问：

def _msg_attr(m, key, default=None):
    if isinstance(m, dict):
        return m.get(key, default)
    # 先试直接属性，再试 additional_kwargs（reasoning_content 存在这里）
    val = getattr(m, key, None)
    if val is not None:
        return val
    return getattr(m, "additional_kwargs", {}).get(key, default)

格式对照

字段	OpenAI API 格式	LangChain AIMessage 格式
tool_calls	`[{"id":"x", "type":"function", "function":{"name":"...", "arguments":"..."}}]`	`[{"name":"...", "args":{...}, "id":"x", "type":"tool_call"}]`
tool_call_id	`"tool_call_id": "x"`	`ToolMessage.tool_call_id` 属性
reasoning	`"reasoning_content": "..."` (顶层)	`additional_kwargs["reasoning_content"]`

⚠️ 双向转换是关键

history 转 dict 发给 API 时，必须把 LangChain 格式的 tool_calls 转回 OpenAI 格式：

t["name"] → function.name
json.dumps(t["args"]) → function.arguments

直接 d["tool_calls"] = tc 会把 LangChain 格式发给 API，模型无法识别。

常见错误

错误	原因	修复
`ValueError: No AIMessage found in input`	_llm_node 返回 plain dict	返回 AIMessage 实例
`tool_call() got an unexpected keyword argument 'function'`	传了 OpenAI 格式 tool_calls 给 AIMessage	转为 LangChain 格式
`AttributeError: 'AIMessage' object has no attribute 'get'`	用 `.get()` 访问 AIMessage	用 `getattr` 或 `_msg_attr`
API 返回格式错误或工具调用失败	history 转 dict 时没把 LangChain tool_calls 转回 OpenAI 格式	双向转换

适用场景

模型不支持标准 bind_tools() 接口
需要处理流式响应中的 tool_calls
需要保留自定义的 reasoning_content 处理
从手写工具执行逻辑迁移到 LangGraph 标准组件
需要注入外部依赖（数据库连接等）到工具函数

Más de este repositorio

mismo repositorio

navi-design

abxxvrv/navi_agent

Navi Agent 设计参考库。当用户询问 navi agent 的架构设计、功能方案、技术选型、或某个具体实现应该如何做时，自动参考 Claude Code (cc_src/)、KIMI-CLI (kimi-cli/)、Hermes (hermes-agent/) 三个成熟项目的源码来提供建议。触发场景包括：讨论 agent loop 设计、工具系统、审批机制、会话管理、UI/CLI 交互、技能系统、插件机制、subagent、MCP 集成等任何 navi 的功能设计问题。也适用于：添加新工具、修改工具行为、集成外部 API。也适用于：多智能体系统设计、Worker 模式、sub_agent 委派模式、Interactive SubAgent 模式、Human-in-the-Loop 模式。也适用于：LangGraph 功能评估、工具调用兼容性分析、模型适配方案选型。

2026-07-151

cli-tool-output-arch

abxxvrv/navi_agent

CLI agent 工具输出架构设计与调试。当排查工具输出重复、输出丢失、输出乱序、实时流与结果渲染冲突、输出截断策略等问题时使用。也适用于为新的 CLI agent 工具设计输出通道。触发词：工具输出重复、output twice、双份输出、流式输出、tool result rendering、输出截断、truncat。

2026-06-271

navi-tool-dev

abxxvrv/navi_agent

Navi Agent 工具开发指南。当需要为 Navi 添加新工具（tool）、修改已有工具、或理解工具注册机制时使用。触发场景：用户说"写一个 XX 工具"、"添加工具"、"tool 怎么写"、"注册工具"、涉及 tool.py 或 runtime.py 中工具注册部分的修改。也适用于：集成外部工具协议（MCP 等）、添加工具子系统、批量注册动态发现的工具。也适用于：添加或修改斜杠命令（/xxx）、斜杠命令子命令。

2026-06-271

pptx

abxxvrv/navi_agent

演示文稿创建、编辑和分析。当用户需要处理 PowerPoint 演示文稿（.pptx 文件）时使用，包括创建新演示文稿、修改内容、处理布局、添加批注或演讲者备注等。也适用于将 guizang-ppt HTML 转换为 PPTX。也适用于基于 Word/文本文档生成学术汇报 PPT（docx→PPT），包含中文学术 PPT 配色方案、卡片布局、条形图、表格等常用模式。触发词：做PPT、做个PPT、演示文稿、slides、presentation、汇报PPT。

2026-06-271

cli-hub-meta-skill

abxxvrv/navi_agent

Discover agent-native CLIs for professional software. Access the live catalog to find tools for creative workflows, productivity, AI, and more.

2026-06-261

code-architecture-analysis

abxxvrv/navi_agent

深入分析复杂项目的架构设计、核心机制和实现原理。当用户询问"为什么 X 能做到 Y"、"系统是怎么工作的"、"分析一下这个项目的架构"时使用。也适用于主动分析开源项目的技术栈、模块划分、数据流、配置体系。覆盖 LangChain/LangGraph Agent、前后端分离 Web 应用、CLI 工具等常见项目类型。也适用于：代码缺陷审查（安全漏洞、架构问题、设计反模式）、两个或多个项目的全面对比分析。触发词："缺陷"、"有什么问题"、"对比"、"区别"、"哪个更好"、"review"、"审计"。

2026-06-261

name	langgraph-custom-tool-calling
description	在LangGraph中使用ToolNode时保留自定义工具调用构造方式。当模型不支持标准bind_tools()、需要手动解析流式响应中的tool_calls、或要从手写工具执行逻辑迁移到ToolNode时使用。

LangGraph 自定义工具调用集成

核心约束

ToolNode 的 invoke() 内部用 isinstance(m, AIMessage) 查找最新 assistant 消息。plain dict 不会被识别，必须返回 AIMessage 实例。

完整模式：自定义 LLM + ToolNode

1. llm_node 返回 AIMessage（不是 dict！）

import json
from langchain_core.messages import AIMessage

def _llm_node(state):
    messages = state["messages"]

    # history 可能含 AIMessage/ToolMessage，转为 dict 给 OpenAI SDK
    api_messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for m in messages:
        if isinstance(m, dict):
            api_messages.append(m)
        else:
            d = {"role": getattr(m, "role", "assistant"),
                 "content": getattr(m, "content", "") or ""}
            tc = getattr(m, "tool_calls", None)
            if tc:
                # ⚠️ AIMessage 存的是 LangChain 格式，必须转回 OpenAI 格式
                d["tool_calls"] = [
                    {"id": t["id"], "type": "function",
                     "function": {"name": t["name"],
                                  "arguments": json.dumps(t.get("args", {}), ensure_ascii=False)}}
                    for t in tc
                ]
            tcid = getattr(m, "tool_call_id", None)
            if tcid:
                d["tool_call_id"] = tcid
            rkc = getattr(m, "additional_kwargs", {}).get("reasoning_content")
            if rkc:
                d["reasoning_content"] = rkc
            api_messages.append(d)

    # --- 自定义流式调用（保持你的 reasoning_content、stream 逻辑） ---
    stream = your_llm_stream(api_messages, tools_schema)
    content, tool_calls_raw, reasoning = parse_stream(stream)  # 你的解析逻辑

    # --- OpenAI 格式 → LangChain 格式 ---
    lc_tool_calls = []
    for tc in tool_calls_raw:
        args_str = tc.get("function", {}).get("arguments", "{}")
        try:
            args = json.loads(args_str)
        except Exception:
            args = {"raw": args_str}
        lc_tool_calls.append({
            "name": tc["function"]["name"],
            "args": args,
            "id": tc.get("id", ""),
            "type": "tool_call",
        })

    ai_msg = AIMessage(content=content, tool_calls=lc_tool_calls)
    if reasoning:
        ai_msg.additional_kwargs["reasoning_content"] = reasoning

    return {"messages": [*messages, ai_msg]}

2. 条件边兼容两种消息类型

from langchain_core.messages import AIMessage

def _should_continue(state):
    last = state["messages"][-1]
    if isinstance(last, AIMessage):
        return "tools" if last.tool_calls else END
    return "tools" if last.get("tool_calls") else END

3. 兼容辅助函数（dict ↔ LangChain 混合 history）

history 会同时包含 plain dict（user/tool 消息）和 AIMessage/ToolMessage，需要用兼容函数访问：

def _msg_attr(m, key, default=None):
    if isinstance(m, dict):
        return m.get(key, default)
    # 先试直接属性，再试 additional_kwargs（reasoning_content 存在这里）
    val = getattr(m, key, None)
    if val is not None:
        return val
    return getattr(m, "additional_kwargs", {}).get(key, default)

格式对照

字段	OpenAI API 格式	LangChain AIMessage 格式
tool_calls	`[{"id":"x", "type":"function", "function":{"name":"...", "arguments":"..."}}]`	`[{"name":"...", "args":{...}, "id":"x", "type":"tool_call"}]`
tool_call_id	`"tool_call_id": "x"`	`ToolMessage.tool_call_id` 属性
reasoning	`"reasoning_content": "..."` (顶层)	`additional_kwargs["reasoning_content"]`

⚠️ 双向转换是关键

history 转 dict 发给 API 时，必须把 LangChain 格式的 tool_calls 转回 OpenAI 格式：

t["name"] → function.name
json.dumps(t["args"]) → function.arguments

直接 d["tool_calls"] = tc 会把 LangChain 格式发给 API，模型无法识别。

常见错误

错误	原因	修复
`ValueError: No AIMessage found in input`	_llm_node 返回 plain dict	返回 AIMessage 实例
`tool_call() got an unexpected keyword argument 'function'`	传了 OpenAI 格式 tool_calls 给 AIMessage	转为 LangChain 格式
`AttributeError: 'AIMessage' object has no attribute 'get'`	用 `.get()` 访问 AIMessage	用 `getattr` 或 `_msg_attr`
API 返回格式错误或工具调用失败	history 转 dict 时没把 LangChain tool_calls 转回 OpenAI 格式	双向转换

适用场景

模型不支持标准 bind_tools() 接口
需要处理流式响应中的 tool_calls
需要保留自定义的 reasoning_content 处理
从手写工具执行逻辑迁移到 LangGraph 标准组件
需要注入外部依赖（数据库连接等）到工具函数