简记负责的项目上线后复盘与总结-------一些开源工具使用后过于封装，简单探索底部逻辑。

背景

先理解 Agent 的核心概念：

• RAG：是固定流程 → 用户问题 → 检索 → LLM 回答。逻辑写死的。

• Agent：让 LLM 来决定下一步动作，比如：

  • 需要调用哪个工具（数据库、API、计算器…）

  • 是否要多步推理（先查天气，再查日历，最后组合回答）

  • 什么时候停止（给出最终答案）

常见的 agent 模式有：

• ReAct (Reason + Act)：最常见，用 LLM 决策 → 调用工具 → 再判断 → 直到输出。

• Plan-and-Execute：模型先生成完整计划，再一步步执行。

• LangGraph Agent：更灵活，用状态机来定义 agent 工作流。

RAG 是“流水线”，Agent 是“决策者”。

在大语言模型（LLM）的应用开发中，工具调用（Function Calling） 是实现智能体（Agent）能力扩展的核心机制。无论是构建检索增强生成（RAG）系统，还是让模型具备联网、数据库查询、计算能力，工具调用都是绕不过去的关键点。

1. 传统 Tool 工具类写法

之前在 LangChain、Autogen 、Xinference等框架中，最常见的方式是通过 Tool 定义一个工具：

# LangChain 示例：定义一个简单计算器 Tool
from langchain.tools import tool

@tool
def add_numbers(x: int, y: int) -> int:
    """计算两个数字的和"""
    return x + y

然后在 Agent 中把它注册进去：

from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")

agent = initialize_agent(
    tools=[add_numbers],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION
)

agent.run("请帮我计算 12 + 7")

这样模型就能在推理过程中调用 add_numbers 工具。

问题是：

• 每个框架都有自己的一套 Tool 定义方式（LangChain、Autogen、LangGraph 各不相同）

• 工具写死在框架里，难以跨框架复用

• 无法独立运行，一个 Tool 函数必须依附在 Python 进程里

结果就是：工具开发者和框架高度耦合，复用性差。

2. MCP 协议的出现

MCP（Model Context Protocol）由 Anthropic 提出，目标是让工具调用像 HTTP 一样标准化。

它的核心思想：

• 工具作为服务端（MCP Server，如@playwright/mcp）

• 大模型框架作为客户端（MCP Client，如 Autogen、LangGraph agent）

• 双方通过统一协议（JSON-RPC over stdio/socket）通信

好处：

• 工具可以独立运行，不再绑定某个框架

• 工具开发者只需实现 MCP server，用户可以在任何支持 MCP 的框架中使用

• 工具可以开源、共享，比如 npm 包 @playwright/mcp 就能开箱即用浏览器能力

3. MCP 工具简单示例

以一个简单的“数学计算工具”为例，MCP 的写法如下：

# all_tools.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MathTools")

@mcp.tool()
async def add_numbers(x: int, y: int) -> str:
    """计算两个数字的和"""
    return f"{x} + {y} = {x + y}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

这个 all_tools.py 就是一个 MCP Server，它提供了 add_numbers 工具。

然后在客户端（比如用 LangGraph 写一个 Agent）：

from all_tools import add_numbers
from langgraph.graph import StateGraph, START, END
from langchain.schema import HumanMessage, AIMessage

class State(dict): pass

graph_builder = StateGraph(State)

async def math_node(state: State):
    last_message = state["messages"][-1]
    question = last_message.content
    answer = await add_numbers(12, 7)  # 直接调用 MCP 工具
    return {"messages": [AIMessage(content=answer)]}

graph_builder.add_node("math_node", math_node)
graph_builder.add_edge(START, "math_node")
graph_builder.add_edge("math_node", END)
graph = graph_builder.compile()

这里，工具不再依赖 LangChain/Autogen 自己的 Tool 类，而是以 MCP server 提供的接口被调用。

4. MCP 与传统 Tool 对比

特性	传统 Tool 工具类	MCP 协议工具
定义方式	Python 函数 + Tool 装饰器	独立 MCP server（跨语言实现）
依赖框架	强依赖（LangChain/Autogen 各不同）	标准协议，框架无关
复用性	较差，需要改写	高，可被任何 MCP client 调用
运行方式	进程内函数调用	Client ↔ Server 进程间通信
开发者体验	写一次只能在一个框架用	写一次，到处可用

5. LangGraph + MCP + KG-RAG+Xinference 的组合

在实际应用里，可以这样组合：

• Xinference

  • 提供开源大模型推理能力（例如本地部署的 Qwen 模型）。

  • 处理自然语言理解和生成，完成推理任务。

• MCP (Multi-Tool Connector Platform)

  • 提供工具扩展能力，把各种工具封装成统一接口，例如：

    • 数据库查询工具（RAG 检索）

    • 浏览器或 Web API 查询工具

    • 自定义 Python 工具

  • 在 LangGraph 的节点中被调用，实现“模型 + 工具”的组合推理。

• KG-RAG (Knowledge Graph Retrieval-Augmented Generation)

  • 提供基于知识图谱的知识增强能力，通过图谱查询获取实体、关系和上下文信息，为模型提供结构化知识支持。

  • 例如接入 NebulaGraph，实现多实体、多关系的知识增强，帮助模型完成逻辑推理和复杂问答。

• LangGraph

  • 提供工作流/状态图管理功能。

  • 将每个操作（模型推理、工具调用、RAG 检索）抽象为节点，并通过有向图或条件分支进行组合。

  • 管理对话状态、上下文、节点间的条件路由，实现复杂的多轮推理流程。

一个典型流程是：

• 用户提出问题。

• LangGraph 状态图接收问题，判断是否需要工具调用：

  • 需要知识增强 → 调用 RAG 工具（MCP server 实现），将检索到的信息补充到上下文中。

  • 需要外部查询/联网 → 调用浏览器 MCP 工具或 API 接口。

  • 无需工具 → 直接由 Xinference 本地模型生成答案。

• 将结果更新到状态图（State），支持多轮交互或进一步推理。

• 返回最终答案给用户。

这样，你的系统就既能用本地模型， 又能像 ChatGPT 一样调用各种插件。

代码简单实现：

# complete_agent.py
import asyncio
import httpx
import json
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.schema import HumanMessage, AIMessage
from mcp.server.fastmcp import FastMCP

# -------------------------- MCP工具部分 --------------------------
mcp = FastMCP("AllTools_中文工具集")

XINFERENCE_URL = "http://xxx:port/v1/chat/completions"
MODEL_ID = "qwen3"
MODEL_PARAMS = {
    "max_tokens": 512,
    "temperature": 0.1,
    "top_p": 0.9,
    "timeout": 60.0
}

async def call_qwen3_chinese(prompt: str) -> str:
    try:
        payload = {
            "model": MODEL_ID,
            "messages": [
                {
                    "role": "system",
                    "content": (
                        "你是专业的中文学术助手，仅用中文回答问题："
                        "请根据用户输入判断问题类型并回答，可以是数学、物理、历史、新闻、笑话等。"
                        "语言简洁易懂，普通人能看懂。"
                    )
                },
                {"role": "user", "content": prompt}
            ],
            **MODEL_PARAMS
        }

        async with httpx.AsyncClient() as client:
            response = await client.post(
                url=XINFERENCE_URL,
                json=payload,
                timeout=MODEL_PARAMS["timeout"]
            )
            response.raise_for_status()

        response_data = response.json()
        if (
            "choices" in response_data
            and len(response_data["choices"]) > 0
            and "message" in response_data["choices"][0]
            and "content" in response_data["choices"][0]["message"]
        ):
            return response_data["choices"][0]["message"]["content"].strip()
        else:
            return f"Qwen3响应格式错误，原始数据：{json.dumps(response_data)[:500]}"
    except httpx.HTTPError as e:
        response_text = getattr(e.response, "text", "无响应内容")[:500]
        return f" HTTP错误：{str(e)}，响应内容：{response_text}"
    except json.JSONDecodeError:
        return f"JSON解析错误：{response.text[:500]}"
    except Exception as e:
        return f"调用Qwen3时发生意外错误：{str(e)}"

# 通用工具：让模型自己判断类型
@mcp.tool()
async def universal_tool(question: str) -> str:
    return await call_qwen3_chinese(question)

# RAG工具占位：将来可接向量数据库/NebulaGraph
@mcp.tool()
async def rag_tool(query: str) -> str:
    # 占位逻辑，返回示例文本
    return f"[RAG检索结果占位] 查询内容：{query}"

# -------------------------- Agent部分 --------------------------
class State(TypedDict):
    messages: Annotated[list[HumanMessage | AIMessage], add_messages]

graph_builder = StateGraph(State)

# 节点：通用问题（交给模型判断）
async def universal_node(state: State):
    last_message = state["messages"][-1]
    question = last_message.content if last_message.content else ""
    answer = await universal_tool(question)
    return {"messages": [AIMessage(content=answer)]}

# 节点：RAG知识增强
async def rag_node(state: State):
    last_message = state["messages"][-1]
    query = last_message.content if last_message.content else ""
    rag_result = await rag_tool(query)
    # 可以将RAG结果再交给模型处理
    answer = await call_qwen3_chinese(f"{query}\n根据以下知识回答问题：{rag_result}")
    return {"messages": [AIMessage(content=answer)]}

# 添加节点
graph_builder.add_node("universal_node", universal_node)
graph_builder.add_node("rag_node", rag_node)

# 简单流程示例：START → RAG → UNIVERSAL → END
graph_builder.add_edge(START, "rag_node")
graph_builder.add_edge("rag_node", "universal_node")
graph_builder.add_edge("universal_node", END)

graph = graph_builder.compile()

# -------------------------- 测试函数 --------------------------
async def main():
    test_state: State = {
        "messages": [HumanMessage(content="请讲一下鸦片战争的经过")],
    }

    result = await graph.ainvoke(test_state)
    print("=== 交互结果 ===")
    for msg in result["messages"]:
        if isinstance(msg, HumanMessage):
            print(f"用户：{msg.content}")
        elif isinstance(msg, AIMessage):
            print(f"AI：{msg.content}")

# -------------------------- 启动 --------------------------
if __name__ == "__main__":
    # 启动 MCP 服务（异步）
    import threading
    threading.Thread(target=lambda: mcp.run(transport="stdio"), daemon=True).start()
    # 启动 Agent 测试
    asyncio.run(main())

6. 结语

过去的 Tool 工具类能在小范围内扩展，但它们彼此割裂，生态分散。
MCP 的出现，就像 给 Agent 装上了 USB 插口：任何 MCP 工具都能即插即用。