目录

(1).创建虚拟环境，python=>3.11, 安装库

(2)下载langgraph新项目 ，下载时，按enter

（3）项目下载好后，文件如下， 把.env.sample -------修改---->.env 

(4) 把百炼阿里云平台的API-KEY、智普等平台的API-KEY，复制到.env中配置。

（5）修改anggraph.json中代码路径和 构建的Agent名称，要与实际的对应。

（5） 在src/agent/langgraph.py中构建代码，如下。

（6）魔搭社区免费部署mcp

---

(1).创建虚拟环境，python=>3.11, 安装库

pip install langgraph  #安装核心库
pip install --upgrade "langgraph-cli[inmem]" -i https://pypi.tuna.tsinghua.edu.cn/simple  
# 安装服务器、监控 管理用于部署

pip install langchain-openai

(2)下载langgraph新项目 ，下载时，按enter

langgraph new +项目名称

（3）项目下载好后，文件如下， 把.env.sample -------修改---->.env 

(4) 把百炼阿里云平台的API-KEY、智普等平台的API-KEY，复制到.env中配置。

（5）修改anggraph.json中代码路径和 构建的Agent名称，要与实际的对应。

（5） 在src/agent/langgraph.py中构建代码，如下。

把src设置为根目录

import json
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.messages import ToolMessage,HumanMessage,AIMessage,SystemMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState, StateGraph
from langgraph.constants import  START,END
import asyncio
from langgraph.prebuilt import tools_condition
from  typing import Any,Dict,List

import sys
import os

# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from demo1.env_utils import ZHIPU_API_KEY 


#mcp工具
from demo1.llms import llm

mcp_zhipu_websearch_config = {
    "url": "https://open.bigmodel.cn/api/mcp/web_search/sse?Authorization=" + ZHIPU_API_KEY,
    'transport': 'sse'
}
mcp_12306_server_config = {
    "url": "https://mcp.api-inference.modelscope.net/74b96a1ea0e547/sse",
    'transport': 'sse'
}
mcp_chart_server_config = {
    "url": "https://mcp.api-inference.modelscope.net/8f29f2b5aced47/sse",
    'transport': 'sse'
}

mcp_client = MultiServerMCPClient(
    {
    # 'chart_mcp':mcp_chart_server_config,
    # '12306_mcp':mcp_12306_server_config,
     'zhipu_mcp':mcp_zhipu_websearch_config,
     })




# 改进原始的 BasicToolNode，添加错误处理
class BasicToolNode:
    def __init__(self, tools: list):
        self.tools_by_name = {tool.name: tool for tool in tools}

    async def __call__(self, state: Dict[str, Any]) -> Dict[str, List[ToolMessage]]:
        if not (messages := state.get('messages')):
            raise ValueError('输入的数据中未找到消息内容')
        message: AIMessage = messages[-1]
        outputs = await self._execute_tool_calls(message.tool_calls)
        return {'messages': outputs}

    async def _execute_tool_calls(self, tool_calls: List[Dict]) -> List[ToolMessage]:
        async def _invoke_tool(tool_call: Dict) -> ToolMessage:
            try:
                tool = self.tools_by_name.get(tool_call['name'])
                if not tool:
                    raise KeyError(f"未注册的工具：{tool_call['name']}")

                if hasattr(tool, 'ainvoke'):
                    tool_result = await tool.ainvoke(tool_call['args'])
                else:
                    loop = asyncio.get_running_loop()
                    tool_result = await loop.run_in_executor(
                        None, tool.invoke, tool_call['args']
                    )

                return ToolMessage(
                    content=json.dumps(tool_result, ensure_ascii=False),
                    name=tool_call['name'],
                    tool_call_id=tool_call['id'],
                )
            except Exception as e:
                # 添加错误处理
                return ToolMessage(
                    content=json.dumps({"error": str(e)}, ensure_ascii=False),
                    name=tool_call['name'],
                    tool_call_id=tool_call['id'],
                )

        return await asyncio.gather(*[_invoke_tool(tool_call) for tool_call in tool_calls])


class State(MessagesState):
    pass



#自定义路由函数
def rout_func(state:State):

    if hasattr(state,'messages'):
        messages = state.messages
    elif isinstance(state,dict) and 'messages' in state:
        messages = state.get('messages', [])
    else:
        raise ValueError(f"无法从状态中获取到messages:{state}")

    last_message = messages[-1]

    if hasattr(last_message,'tool_calls') and len(last_message.tool_calls)>0:
        return 'tools'




async  def create_graph():
    builder = StateGraph(State)

    tools = await  mcp_client.get_tools()

    llm_with_tools = llm.bind_tools(tools)


    async  def chabot(state:State):
        response = await  llm_with_tools.ainvoke(state['messages'])
        return {'messages':[response]}



    builder.add_node('chatbot',chabot)

    tool_node = BasicToolNode(tools)
    builder.add_node('tools',tool_node)

    builder.add_edge(START,'chatbot')
    builder.add_edge('tools','chatbot')
    #条件边
    #builder.add_conditional_edges('chatbot',tools_condition)
    #
    builder.add_conditional_edges(
        'chatbot',
        rout_func,
        path_map={
            'tools':'tools',
            END:END
        }
    )

    memory = MemorySaver()
    # #interrupt_before=[节点1，节点2....]   如果用langgraph dev 启动，不需要写chekpoint,自带有，不然会报错。
    graph = builder.compile(checkpointer=memory,interrupt_before=['tools'])
    #记得返回graph
    return graph


#agent = create_graph()
agent = asyncio.run(create_graph())  #异步初始化智能体

构建env_utils.py---加载环境变量

import  os
from dotenv import load_dotenv
load_dotenv(override=True)

ALIBABA_API_KEY = os.getenv('ALIBABA_API_KEY')
ZHIPU_API_KEY = os.getenv('ZHIPU_API_KEY')

构建大模型调用函数my_llm.py

from  langchain_openai import ChatOpenAI
from agent.env_utils import  ALIBABA_API_KEY


# 百炼阿里云
llm = ChatOpenAI(
    model = 'qwen-plus',  #更换模型
    temperature=0.5, 
    api_key = ALIBABA_API_KEY,
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

（6）魔搭社区免费部署mcp

 代码中用到的mcp-URL 从魔搭社区中复制。

（7）启动项目

langgraph dev  #自动弹出langgraph smith界面