Zhang XiJun

uv tool install pytest

pip install langchain-mcp-adapters

import asyncio # 需要导入 asyncio 来运行异步函数
# 从langchain_mcp_adapters.client模块导入MultiServerMCPClient类
# 从langgraph.prebuilt模块导入create_react_agent函数
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

# 导入 LLM 相关库
from langchain_openai import ChatOpenAI

# 将主要逻辑封装在一个异步函数中
async def main():
    # 创建MultiServerMCPClient实例，配置两个不同的服务
    client = MultiServerMCPClient(
        {
            "math": {  # 数学计算服务
                "command": "python",  # 使用python命令启动
                # 替换为你的math_server.py文件的绝对路径
                "args": ["/workspace/langgraph-mcp/math_server.py"],
                "transport": "stdio",  # 使用标准输入输出传输
            },
            "weather": {  # 天气服务
                # 确保你的天气服务器在8000端口运行
                # *** 确保这个 URL 是正确的，并且服务器正在运行 ***
                "url": "http://localhost:8000/mcp",
                "transport": "streamable_http",  # 使用可流式HTTP传输
            }
        }
    )

    tools = []
    try:
        # 在异步函数内部正确使用 await
        tools = await client.get_tools()
        print(f"成功获取到 {len(tools)} 个MCP工具。")
        for tool_item in tools:
            print(f"  - {tool_item.name}: {tool_item.description}")
    except Exception as e:
        print(f"获取MCP工具失败: {e}")
        print("请确保MCP服务URL有效且可访问，或者您已正确配置了认证信息。")
        # 在函数内部，如果出错可以选择返回或继续处理
        # return # 这里可以 return，但会结束 main 函数

    if not tools:
        print("没有获取到工具，无法创建代理。")
        return

    # 创建ReAct代理
    llm = ChatOpenAI(
        model="qwen3-235b-a22b-thinking-2507",
        api_key="sk-a8ef27c47ea84224ac6eed6d4bba1bab",
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 修正了末尾多余的空格
    )
    agent = create_react_agent(llm, tools)
    

    # 异步调用代理来解决数学问题
    # 确保在异步函数内部使用 await
    math_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
    )
    print("\n--- 数学问题回答 ---")
    print(math_response["messages"][-1].content) # 打印最后一条消息（LLM的回答）

    # 异步调用代理来查询天气
    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
    )
    print("\n--- 天气问题回答 ---")
    print(weather_response["messages"][-1].content) # 打印最后一条消息（LLM的回答）


# --- 这是脚本的入口点 ---
# 使用 asyncio.run() 来运行你的主异步函数
if __name__ == "__main__":
    asyncio.run(main())

# Create a new directory for our project
uv init weather
cd weather

# Create virtual environment and activate it
uv venv
source .venv/bin/activate

# Install dependencies
uv add "mcp[cli]" httpx

# Create our server file
touch weather.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

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

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """Get weather for location."""
    return "It's always sunny in New York"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

"mcpServers": {
  "fetch": {
    "command": "uvx",
    "args": ["mcp-server-fetch"]
  }
}

pip install mcp-server-fetch

"mcpServers": {
  "fetch": {
    "command": "python",
    "args": ["-m", "mcp_server_fetch"]
  }
}

{
  "mcpServers": {
    "fetch": {
      "type": "sse",
      "url": "https://mcp.api-inference.modelscope.net/991cf46/sse"
    }
  }
}

# 创建项目目录
uv init mcp-client
cd mcp-client

# 创建虚拟环境
uv venv

# 激活虚拟环境
# 在 Windows 上:
.venv\Scripts\activate
# 在 Unix 或 MacOS 上:
source .venv/bin/activate

# 安装所需包
uv add mcp anthropic python-dotenv
#使用镜像源安装
uv add mcp anthropic python-dotenv --index-url https://pypi.tuna.tsinghua.edu.cn/simple/

# 删除样板文件
# 在 Windows 上:
del main.py
# 在 Unix 或 MacOS 上:
rm main.py

# 创建我们的主文件
touch client.py

# Create .env file
touch .env

ANTHROPIC_API_KEY=<your key here>

echo ".env" >> .gitignore

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # 从 .env 文件加载环境变量

class MCPClient:
    def __init__(self):
        # 初始化会话和客户端对象
        self.session: Optional[ClientSession] = None  # MCP客户端会话
        self.exit_stack = AsyncExitStack()  # 异步上下文管理器堆栈，用于资源清理
        self.anthropic = Anthropic()  # Anthropic AI 客户端
        
    # 后续方法将在这里定义

async def connect_to_server(self, server_script_path: str):
    """连接到MCP服务器

    Args:
        server_script_path: 服务器脚本路径 (.py 或 .js 文件)
    """
    # 检查是否为Python文件
    is_python = server_script_path.endswith('.py')
    # 检查是否为JavaScript文件
    is_js = server_script_path.endswith('.js')
    
    # 如果不是Python或JavaScript文件，则抛出错误
    if not (is_python or is_js):
        raise ValueError("服务器脚本必须是 .py 或 .js 文件")

    # 根据文件类型确定执行命令
    command = "python" if is_python else "node"
    
    # 创建服务器参数对象
    server_params = StdioServerParameters(
        command=command,           # 执行命令
        args=[server_script_path], # 脚本路径作为参数
        env=None                   # 环境变量（使用默认环境）
    )

    # 建立stdio客户端连接并将其添加到异步上下文管理器中
    stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
    self.stdio, self.write = stdio_transport
    
    # 创建客户端会话并将其添加到异步上下文管理器中
    self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

    # 初始化会话
    await self.session.initialize()

    # 列出可用的工具
    response = await self.session.list_tools()
    tools = response.tools
    
    # 打印连接的服务器提供的工具列表
    print("\n已连接到服务器，可用工具:", [tool.name for tool in tools])

async def process_query(self, query: str) -> str:
    """使用Claude和可用工具处理查询"""
    # 构建消息列表
    messages = [
        {
            "role": "user",      # 用户角色
            "content": query     # 用户查询内容
        }
    ]

    # 获取可用工具列表
    response = await self.session.list_tools()
    available_tools = [{
        "name": tool.name,           # 工具名称
        "description": tool.description,  # 工具描述
        "input_schema": tool.inputSchema  # 工具输入模式
    } for tool in response.tools]

    # 初始Claude API调用
    response = self.anthropic.messages.create(
        model="qwen3-235b-a22b",  # 使用的模型
        max_tokens=1000,                     # 最大返回令牌数
        messages=messages,                   # 消息历史
        tools=available_tools               # 可用工具
    )

    # 处理响应并处理工具调用
    final_text = []  # 存储最终文本结果

    assistant_message_content = []  # 存储助手消息内容
    for content in response.content:  # 遍历响应内容
        if content.type == 'text':  # 如果是文本内容
            final_text.append(content.text)  # 添加到最终结果
            assistant_message_content.append(content)  # 添加到助手消息
        elif content.type == 'tool_use':  # 如果是工具调用
            tool_name = content.name    # 工具名称
            tool_args = content.input   # 工具参数

            # 执行工具调用
            result = await self.session.call_tool(tool_name, tool_args)
            final_text.append(f"[调用工具 {tool_name}，参数 {tool_args}]")

            assistant_message_content.append(content)
            # 添加助手消息到历史
            messages.append({
                "role": "assistant",
                "content": assistant_message_content
            })
            # 添加工具执行结果到历史
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",      # 工具结果类型
                        "tool_use_id": content.id,  # 工具使用ID
                        "content": result.content   # 工具执行结果
                    }
                ]
            })

            # 获取Claude的下一个响应
            response = self.anthropic.messages.create(
                model="qwen3-235b-a22b",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )

            # 添加响应文本到最终结果
            final_text.append(response.content[0].text)

    # 返回连接后的最终文本结果
    return "\n".join(final_text)

async def chat_loop(self):
    """运行交互式聊天循环"""
    print("\nMCP客户端已启动!")
    print("输入您的问题或输入'quit'退出。")

    while True:  # 无限循环，持续接收用户输入
        try:
            query = input("\n问题: ").strip()  # 获取用户输入并去除首尾空格

            if query.lower() == 'quit':  # 如果用户输入'quit'（不区分大小写）
                break  # 退出循环

            # 处理用户查询并获取响应
            response = await self.process_query(query)
            print("\n" + response)  # 打印AI响应结果

        except Exception as e:  # 捕获所有异常
            print(f"\n错误: {str(e)}")  # 打印错误信息

async def cleanup(self):
    """清理资源"""
    await self.exit_stack.aclose()  # 异步关闭所有在exit_stack中管理的资源

async def main():
    # 检查命令行参数数量，如果少于2个则显示使用说明
    if len(sys.argv) < 2:
        print("用法: python client.py <服务器脚本路径>")
        sys.exit(1)  # 退出程序，返回错误码1

    # 创建MCP客户端实例
    client = MCPClient()
    try:
        # 连接到服务器，sys.argv[1]是第一个命令行参数（服务器脚本路径）
        await client.connect_to_server(sys.argv[1])
        # 启动交互式聊天循环
        await client.chat_loop()
    finally:
        # 确保程序结束时清理资源
        await client.cleanup()

# 程序入口点
if __name__ == "__main__":
    import sys  # 导入sys模块用于处理命令行参数
    # 运行异步主函数
    asyncio.run(main())

uv run client.py path/to/server.py # python server
uv run client.py path/to/build/index.js # node server

import asyncio
import json
import sys
from typing import Optional
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()  # 从 .env 文件加载环境变量

class MCPClient:
    def __init__(self):
        # 初始化会话和客户端对象
        self.session: Optional[ClientSession] = None  # MCP客户端会话
        self.exit_stack = AsyncExitStack()  # 异步上下文管理器堆栈，用于资源清理
        self.anthropic = OpenAI(
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
        )  # 使用OpenAI兼容模式连接通义千问
        
    async def connect_to_server(self, server_script_path: str):
        """连接到MCP服务器
        
        Args:
            server_script_path (str): 服务器脚本路径，支持.py或.js文件
        
        Raises:
            ValueError: 当脚本文件不是.py或.js格式时抛出
        """
        # 检查是否为Python文件
        is_python = server_script_path.endswith('.py')
        # 检查是否为JavaScript文件
        is_js = server_script_path.endswith('.js')
        
        # 如果不是Python或JavaScript文件，则抛出错误
        if not (is_python or is_js):
            raise ValueError("服务器脚本必须是 .py 或 .js 文件")

        # 根据文件类型确定执行命令
        command = "python" if is_python else "node"
        
        # 创建服务器参数对象
        server_params = StdioServerParameters(
            command=command,           # 执行命令
            args=[server_script_path], # 脚本路径作为参数
            env=None                   # 环境变量（使用默认环境）
        )

        # 建立stdio客户端连接并将其添加到异步上下文管理器中
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        
        # 创建客户端会话并将其添加到异步上下文管理器中
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

        # 初始化会话
        await self.session.initialize()

        # 列出可用的工具
        response = await self.session.list_tools()
        tools = response.tools
        
        # 打印连接的服务器提供的工具列表
        print("\n已连接到服务器，可用工具:", [tool.name for tool in tools])

    async def process_query(self, query: str) -> str:
        """使用Qwen和可用工具处理查询"""
        # 构建消息列表
        messages = [
            {
                "role": "user",
                "content": query
            }
        ]

        # 获取可用工具列表并转换为OpenAI格式
        response = await self.session.list_tools()
        available_tools = []
        for tool in response.tools:
            schema = tool.inputSchema
            if isinstance(schema, str):
                schema = json.loads(schema)
            if isinstance(schema, dict) and "properties" in schema:
                schema = {"type": "object", **schema}

            available_tools.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": schema
                }
            })

        # 第一次调用模型
        response = self.anthropic.chat.completions.create(
            model="qwen3-235b-a22b",
            max_tokens=1000,
            messages=messages,
            tools=available_tools,
            extra_body={"enable_thinking": False}
        )

        final_text = []
        message = response.choices[0].message

        # 处理文本内容
        if message.content:
            final_text.append(message.content)

        # 处理工具调用
        if message.tool_calls:
            for tool_call in message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)
                
                # 执行工具调用
                result = await self.session.call_tool(tool_name, tool_args)
                final_text.append(f"[调用工具 {tool_name}，参数 {tool_args}]")
                
                # 处理工具结果
                tool_result_content = ""
                if result.content:
                    for item in result.content:
                        if hasattr(item, 'type') and item.type == 'text':
                            tool_result_content += item.text
                        else:
                            tool_result_content += str(item)

                # 添加助手消息到历史
                messages.append({
                    "role": "assistant",
                    "content": None,
                    "tool_calls": [tool_call]
                })
                
                # 添加工具执行结果到历史
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result_content
                })

                # 再次调用模型
                response = self.anthropic.chat.completions.create(
                    model="qwen3-235b-a22b",
                    max_tokens=1000,
                    messages=messages,
                    tools=available_tools,
                    extra_body={"enable_thinking": False}
                )
                
                # 处理最终响应
                if response.choices and response.choices[0].message.content:
                    final_text.append(response.choices[0].message.content)

        return "\n".join(final_text)

    async def chat_loop(self):
        """运行交互式聊天循环"""
        print("\nMCP客户端已启动!")
        print("输入您的问题或输入'quit'退出。")

        while True:  # 无限循环，持续接收用户输入
            try:
                query = input("\n问题: ").strip()  # 获取用户输入并去除首尾空格

                if query.lower() == 'quit':  # 如果用户输入'quit'（不区分大小写）
                    break  # 退出循环

                # 处理用户查询并获取响应
                response = await self.process_query(query)
                print("\n" + response)  # 打印AI响应结果

            except Exception as e:  # 捕获所有异常
                print(f"\n错误: {str(e)}")  # 打印错误信息
                import traceback
                traceback.print_exc()  # 打印详细错误信息

    async def cleanup(self):
        """清理资源"""
        await self.exit_stack.aclose()  # 异步关闭所有在exit_stack中管理的资源

async def main():
    # 检查命令行参数数量，如果少于2个则显示使用说明
    if len(sys.argv) < 2:
        print("用法: python client.py <服务器脚本路径>")
        sys.exit(1)  # 退出程序，返回错误码1

    # 创建MCP客户端实例
    client = MCPClient()
    try:
        # 连接到服务器，sys.argv[1]是第一个命令行参数（服务器脚本路径）
        await client.connect_to_server(sys.argv[1])
        # 启动交互式聊天循环
        await client.chat_loop()
    finally:
        # 确保程序结束时清理资源
        await client.cleanup()

# 程序入口点
if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

loader = PyPDFLoader("input/健康档案.pdf")
docs = loader.load()
#递归分块
text_splitter = RecursiveCharacterTextSplitter(chunk_siz

from langchain.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
embedding = OpenAIEmbeddings(
api_key="sk-", 
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model="text-embedding-v4",
check_embedding_ctx_length = False,
dimensions=1536
)
# 使用 Chroma 向量数据库存储文档 chunks
vectorstore = Chroma.from_documents(
    documents=chunks,           # 要存储的文档chunks列表（已处理好的文本片段）
    embedding=embedding,        
    persist_directory="chromaDB",  # 向量数据库的持久化存储目录路径
    collection_name="demo001"   # 集合名称，用于区分不同的文档集合
)
vectorstore.persist()

results = vectorstore.similarity_search(
    "张三九的基本信息是什么",
    k=2,
    collection_name="demo001"  # 指定检索的集合
)

# 重新加载已存在的 Chroma 数据库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embedding
)

retriever = vectorstore.as_retriever()

pip install --quiet -U langgraph langchain-community langchain-openai tavily-python

#定义工具
from langchain_tavily import TavilySearch
tools = [TavilySearch(max_results=3)]

from langchain_openai import ChatOpenAI

from langgraph.prebuilt import create_react_agent


llm = ChatOpenAI(
    model="qwen3-235b-a22b-thinking-2507",
    api_key="sk-",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
prompt = "You are a helpful assistant."
agent_executor = create_react_agent(llm, tools, prompt=prompt)

agent_executor.invoke({"messages": [("user", "今天是几月几日")]})

from typing import Literal
from langgraph.graph import END


async def execute_step(state: PlanExecute):
    """执行计划中的步骤"""
    plan = state["plan"]
    # 将计划格式化为带编号的字符串
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
    task = plan[0]  # 获取第一个待执行的任务
    task_formatted = f"""对于以下计划:
{plan_str}\n\n你被分配执行第 {1} 步, {task}。"""
    # 调用代理执行器来执行任务
    agent_response = await agent_executor.ainvoke(
        {"messages": [("user", task_formatted)]}
    )
    # 返回执行结果，添加到历史步骤中
    return {
        "past_steps": [(task, agent_response["messages"][-1].content)],
    }

import operator
from typing import Annotated, List, Tuple
from typing_extensions import TypedDict


class PlanExecute(TypedDict):
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple], operator.add]
    response: str

from pydantic import BaseModel, Field


class Plan(BaseModel):
    """未来要遵循的计划"""

    steps: List[str] = Field(
        description="需要遵循的不同步骤，应该按排序顺序排列"
    )

from langchain_community.chat_models import ChatTongyi
model=ChatTongyi(
 model="qwen-plus-2025-07-14",
 api_key="sk-"
)

from langchain_core.prompts import ChatPromptTemplate

planner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """对于给定的目标，制定一个简单的逐步计划。\
这个计划应该包含独立的任务，如果正确执行这些任务将得到正确的答案。不要添加任何多余的步骤。\
最后一步的结果应该是最终答案。确保每个步骤都包含所需的所有信息——不要跳过任何步骤。""",
        ),
        ("placeholder", "{messages}"),
    ]
)
planner = planner_prompt | model.with_structured_output(Plan)

async def plan_step(state: PlanExecute):
    """制定初始计划步骤"""
    # 使用规划器为用户输入制定计划
    plan = await planner.ainvoke({"messages": [("user", state["input"])]})
    
    return {"plan": plan.steps}

from typing import Union


class Response(BaseModel):
    """对用户的响应"""

    response: str


class Act(BaseModel):
    """要执行的动作"""

    action: Union[Response, Plan] = Field(
        description="要执行的动作。如果你想响应用户，使用 Response。"
        "如果你需要进一步使用工具来获取答案，使用 Plan。"
    )


replanner_prompt = ChatPromptTemplate.from_template(
    """对于给定的目标，制定一个简单的逐步计划。\
这个    计划应该包含独立的任务，如果正确执行这些任务将得到正确的答案。不要添加任何多余的步骤。\
最后一步的结果应该是最终答案。确保每个步骤都包含所需的所有信息——不要跳过任何步骤。

你的目标是：
{input}

你的原始计划是：
{plan}

你目前已经完成了以下步骤：
{past_steps}

相应地更新你的计划。如果不需要更多步骤并且可以返回给用户，则直接响应。否则，填写计划。只添加仍需要完成的步骤到计划中。不要将已经完成的步骤作为计划的一部分返回。"""
)


replanner = replanner_prompt | model.with_structured_output(Act)

async def replan_step(state: PlanExecute):
    """重新规划步骤"""
    # 使用重新规划器根据当前状态更新计划
    output = await replanner.ainvoke(state)
    if isinstance(output.action, Response):
        # 如果动作是响应，返回最终响应
        return {"response": output.action.response}
    else:
        # 如果动作是计划，返回新的计划步骤
        return {"plan": output.action.steps}

def should_end(state: PlanExecute):
    """判断是否结束执行流程"""
    if "response" in state and state["response"]:
        # 如果存在响应内容，结束流程
        return END
    else:
        # 否则继续执行代理步骤
        return "agent"

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

# 创建内存检查点保存器
memory = MemorySaver()

# 创建工作流图，使用 PlanExecute 状态类型
workflow = StateGraph(PlanExecute)

# 添加计划节点
workflow.add_node("planner", plan_step)

# 添加执行步骤节点
workflow.add_node("agent", execute_step)

# 添加重新规划节点
workflow.add_node("replan", replan_step)

# 从开始节点连接到计划节点
workflow.add_edge(START, "planner")

# 从计划节点连接到代理执行节点
workflow.add_edge("planner", "agent")

# 从代理执行节点连接到重新规划节点
workflow.add_edge("agent", "replan")

# 添加条件边 - 从重新规划节点根据条件决定下一步
workflow.add_conditional_edges(
    "replan",
    # 传入决定下一个调用节点的函数
    should_end,
    ["agent", END],  # 可能的下一个节点：代理节点或结束
)

# 最后，编译工作流并添加检查点功能！
# 这将其编译为 LangChain Runnable，
# 意味着你可以像使用其他任何 runnable 一样使用它
app = workflow.compile(checkpointer=memory)

# 配置递归限制，防止无限循环
config = {"configurable": {
        "thread_id": "1",  # 必需：线程ID
    },"recursion_limit": 50}

# 输入问题：2024年澳大利亚网球公开赛男单冠军的家乡是哪里？
inputs = {"input": "2024年澳大利亚网球公开赛男单冠军的家乡是哪里？"}

# 异步流式执行应用
async for event in app.astream(inputs, config=config):
    # 遍历每个事件
    for k, v in event.items():
        # 排除结束标记，打印其他所有事件内容
        if k != "__end__":
            print(v)

version: '3.8'

services:
  postgres:
    image: postgres:15        # 指定具体版本
    container_name: postgres_db
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: postgres
      TZ: Asia/Shanghai       # 设置时区
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    restart: unless-stopped
    healthcheck:             # 健康检查
      test: ["CMD", "pg_isready", "-U", "nange"]
      interval: 10s
      timeout: 5s
      retries: 5
    command: ["postgres", "-c", "max_connections=200"]  # 自定义配置

volumes:
  pgdata:

#拉取并运行
docker-compose up -d

安装依赖           
apt update    #刷新本地软件包索引           
apt install -y git build-essential postgresql-server-dev-15                          

编译并安装 pgvector      
#把 pgvector 的 v0.7.0 稳定版 源码克隆到本地目录 ./pgvector。
git clone --branch v0.7.0 https://github.com/pgvector/pgvector.git                
cd pgvector  
#调用 Makefile 根据当前操作系统 + PostgreSQL 版本编译出二进制文件（.so 共享库）。
make          
#把刚刚编好的 .so 文件和 .sql/.control 文件复制到 PostgreSQL 的扩展目录
make install                
验证安装，检查扩展文件是否安装成功                    
ls -l /usr/share/postgresql/15/extension/vector* 

sudo apt update
sudo apt install libpq-dev postgresql-server-dev-all

pip install langgraph-checkpoint-postgres                    
pip install psycopg psycopg-pool 

from langgraph.store.postgres import PostgresStore
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
# 1) 连接字符串（URI 语法）
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
#   协议://      用户   : 密码   @ 主机:端口 / 数据库名  ? 额外参数

# 2) 连接级参数
connection_kwargs = {
    "autocommit": True,     # 每条 SQL 执行完立即提交，无需手动 commit
    "prepare_threshold": 0, # 禁用服务器端 prepared statement，可减少一次往返
}

# 3) 创建池
connection_pool = ConnectionPool(
    conninfo=DB_URI,
    max_size=20,            # 最多 20 条物理连接
    kwargs=connection_kwargs,
)

# 4) 显式打开池（psycopg 3 的 ConnectionPool 默认懒启动，调 open() 会立即建 min_size 条连接）
connection_pool.open()
print("数据库连接池初始化成功")

# 初始化PostgresStore
in_postgres_store = PostgresStore(
    pool,
    index={
        "dims": 1536,
        "embed": embedding
    }
)
in_postgres_store.setup()

#初始化checkpoint
# 使用传入的连接池创建 PostgresSaver
checkpointer = PostgresSaver(pool)
checkpointer.setup()

#最后编译时添加
graph_builder.compile(checkpointer=checkpointer, store=in_postgres_store)

# 定义了一个异步函数lifespan，它接收一个FastAPI应用实例app作为参数。这个函数将管理应用的生命周期，包括启动和关闭时的操作
# 函数在应用启动时执行一些初始化操作，如加载上下文数据、以及初始化问题生成器
# 函数在应用关闭时执行一些清理操作
# @asynccontextmanager 装饰器用于创建一个异步上下文管理器，它允许你在 yield 之前和之后执行特定的代码块，分别表示启动和关闭时的操作
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    # 申明引用全局变量，在函数中被初始化，并在整个应用中使用
    global graph, connection_pool
    # 启动时执行
    try:
        logger.info("正在初始化模型、定义 Graph...")
        # 初始化 LLM
        llm, embedding = get_llm(llm_type)
        # 创建数据库连接池
        DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
        connection_kwargs = {
            "autocommit": True,
            "prepare_threshold": 0,
        }
        connection_pool = ConnectionPool(
            conninfo=DB_URI,
            max_size=20,
            kwargs=connection_kwargs,
        )
        connection_pool.open()  # 显式打开连接池
        logger.info("数据库连接池初始化成功")
        # 短期记忆 初始化checkpointer
        checkpointer = PostgresSaver(connection_pool)
        checkpointer.setup()
        # 长期记忆 初始化PostgresStore
        in_postgres_store = PostgresStore(
            connection_pool,
            index={
                "dims": 1536,
                "embed": embedding
            }
        )
        in_postgres_store.setup()
        # 定义 Graph
        graph = create_graph(llm, checkpointer, in_postgres_store )
        # 保存 Graph 可视化图
        save_graph_visualization(graph)
        logger.info("初始化完成")
    except Exception as e:
        logger.error(f"初始化过程中出错: {str(e)}")
        raise

    yield  # 应用运行期间

    # 关闭时执行
    logger.info("正在关闭...")
    if connection_pool:
        connection_pool.close()  # 关闭连接池
        logger.info("数据库连接池已关闭")


# lifespan参数用于在应用程序生命周期的开始和结束时执行一些初始化或清理工作
app = FastAPI(lifespan=lifespan)

┌──────────────┐
│ LangGraph    │ 运行图实例
└────┬─────────┘
     │1. 写 checkpoints
     │2. 写 checkpoint_writes
     │3. 拆大字段到 checkpoint_blobs
     ▼
┌──────────────┐
│ 业务代码     │ 读写 KV/向量
└────┬─────────┘
     │1. 写 store
     │2. 写 store_vectors
     ▼
 PostgreSQL (pgsql)

# 查看发行版名称和版本
cat /etc/os-release

PRETTY_NAME="Ubuntu 24.04.2 LTS"
NAME="Ubuntu"
VERSION_ID="24.04"
VERSION="24.04.2 LTS (Noble Numbat)"
VERSION_CODENAME=noble
ID=ubuntu
ID_LIKE=debian
HOME_URL="https://www.ubuntu.com/"
SUPPORT_URL="https://help.ubuntu.com/"
BUG_REPORT_URL="https://bugs.launchpad.net/ubuntu/"
PRIVACY_POLICY_URL="https://www.ubuntu.com/legal/terms-and-policies/privacy-policy"
UBUNTU_CODENAME=noble
LOGO=ubuntu-logo

# 更新软件包索引
sudo apt update

# 安装 PostgreSQL 和常用扩展
sudo apt install -y postgresql postgresql-contrib

# 检查服务状态
#Linux 容器/子系统（ Docker、WSL 或 LXC）没有使用 systemd ，因此 systemctl 无法工作
sudo systemctl status postgresql

#确认 PostgreSQL 已安装
which psql            # 应该输出 /usr/bin/psql
pg_ctl --version      # 显示版本号即已安装

sudo service postgresql start

sudo service postgresql stop

sudo service postgresql status

sudo -u postgres psql

postgresql://<用户名>:<密码>@127.0.0.1:5432/<数据库名>[?参数=值&...]

# 基于数据库持久化存储的short-term
db_uri = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# short-term短期记忆 实例化PostgresSaver对象 并初始化checkpointer
# long-term长期记忆 实例化PostgresStore对象 并初始化store
async with (
    AsyncPostgresSaver.from_conn_string(db_uri) as checkpointer,
    AsyncPostgresStore.from_conn_string(db_uri) as store

):
    await store.setup()
    await checkpointer.setup()

cat > /etc/apt/sources.list <<'EOF'
deb http://mirrors.aliyun.com/debian/ bookworm main contrib non-free
deb http://mirrors.aliyun.com/debian/ bookworm-updates main contrib non-free
deb http://mirrors.aliyun.com/debian/ bookworm-backports main contrib non-free
deb http://mirrors.aliyun.com/debian-security bookworm-security main contrib non-free
EOF

apt update

{
  "mcpServers": {
    "fetch": {
      "command": "uvx",
      "args": [
        "mcp-server-fetch"
      ]
    }
  }
}

uv init weather

uv sync

source .venv/bin/activate 

#添加依赖
uv add "mcp[cli]" httpx

# 导入类型提示模块，用于类型注解
from typing import Any

# 导入httpx库，用于发送HTTP请求
import httpx

# 从mcp.server.fastmcp模块导入FastMCP类
# FastMCP是一个快速构建MCP（Model Control Protocol）服务器的框架
from mcp.server.fastmcp import FastMCP

# 创建FastMCP实例，命名为"weather"，日志级别设置为ERROR（只显示错误信息）
mcp = FastMCP("weather", log_level="ERROR")


# 常量定义
# NWS（National Weather Service）API的基础URL
NWS_API_BASE = "https://api.weather.gov"
# 用户代理字符串，用于标识应用程序
USER_AGENT = "weather-app/1.0"


async def make_nws_request(url: str) -> dict[str, Any] | None:
    """向NWS API发起请求并处理错误。
    
    Args:
        url: 要请求的API URL
        
    Returns:
        成功时返回解析后的JSON数据字典，失败时返回None
    """
    # 设置请求头信息
    headers = {
        "User-Agent": USER_AGENT,           # 用户代理标识
        "Accept": "application/geo+json"    # 接受的数据格式
    }
    
    # 创建异步HTTP客户端
    async with httpx.AsyncClient() as client:
        try:
            # 发起GET请求，设置超时时间为30秒
            response = await client.get(url, headers=headers, timeout=30.0)
            # 如果响应状态码不是2xx，抛出异常
            response.raise_for_status()
            # 返回解析后的JSON数据
            return response.json()
        except Exception:
            # 捕获所有异常，返回None表示请求失败
            return None


def format_alert(feature: dict) -> str:
    """将警报数据格式化为可读的字符串。
    
    Args:
        feature: 包含警报信息的字典
        
    Returns:
        格式化后的警报字符串
    """
    # 获取警报属性
    props = feature["properties"]
    # 格式化警报信息，使用get方法提供默认值防止键不存在
    return f"""
事件: {props.get('event', '未知')}
区域: {props.get('areaDesc', '未知')}
严重程度: {props.get('severity', '未知')}
描述: {props.get('description', '无描述信息')}
指示: {props.get('instruction', '无具体指示')}
"""


# 使用@mcp.tool()装饰器将函数注册为MCP工具
@mcp.tool()
async def get_alerts(state: str) -> str:
    """获取指定美国州的天气警报。
    
    Args:
        state: 两个字母的美国州代码（例如：CA, NY）
        
    Returns:
        格式化后的警报信息字符串
    """
    # 构建获取州警报的URL
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    # 发起API请求获取数据
    data = await make_nws_request(url)

    # 检查数据是否有效
    if not data or "features" not in data:
        return "无法获取警报或未找到警报。"

    # 检查是否有警报
    if not data["features"]:
        return "该州无活动警报。"

    # 格式化所有警报
    alerts = [format_alert(feature) for feature in data["features"]]
    # 用分隔符连接所有警报
    return "\n---\n".join(alerts)


# 注册为MCP工具的天气预报函数
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取指定位置的天气预报。
    
    Args:
        latitude: 位置的纬度
        longitude: 位置的经度
        
    Returns:
        格式化后的天气预报字符串
    """
    # 首先获取预报网格端点
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    # 检查点数据是否获取成功
    if not points_data:
        return "无法获取此位置的预报数据。"

    # 从点响应中获取预报URL
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    # 检查预报数据是否获取成功
    if not forecast_data:
        return "无法获取详细预报。"

    # 将时间段格式化为可读的预报
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    # 只显示接下来的5个时间段
    for period in periods[:5]:
        forecast = f"""
{period['name']}:
温度: {period['temperature']}°{period['temperatureUnit']}
风力: {period['windSpeed']} {period['windDirection']}
预报: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    # 用分隔符连接所有预报
    return "\n---\n".join(forecasts)


# 程序入口点
if __name__ == "__main__":
    # 初始化并运行服务器，使用stdio传输方式
    mcp.run(transport='stdio')

"weather": {
    "disabled": false,
    "timeout": 60,
    "command": "uv",
    "args": [
      "--directory",
      "/Users/joeygreen/PycharmProjects/weather",
      "run",
      "weather.py"
    ],
    "transportType": "stdio"
  }

{
    "name": "get_forecast",
    "description": "Get weather forecast for a location.\n\nArgs:\n    latitude: Latitude of the location\n    longitude: Longitude of the location\n",
    "inputSchema": {
        "properties": {
            "latitude": {
                "title": "Latitude",
                "type": "number"
            },
            "longitude": {
                "title": "Longitude",
                "type": "number"
            }
        },
        "required": [
            "latitude",
            "longitude"
        ],
        "title": "get_forecastArguments",
        "type": "object"
    }
}

# 注册为MCP工具的天气预报函数
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取指定位置的天气预报。
    
    Args:
        latitude: 位置的纬度
        longitude: 位置的经度
        
    Returns:
        格式化后的天气预报字符串
    """
    # 首先获取预报网格端点
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    # 检查点数据是否获取成功
    if not points_data:
        return "无法获取此位置的预报数据。"

    # 从点响应中获取预报URL
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    # 检查预报数据是否获取成功
    if not forecast_data:
        return "无法获取详细预报。"

    # 将时间段格式化为可读的预报
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    # 只显示接下来的5个时间段
    for period in periods[:5]:
        forecast = f"""
{period['name']}:
温度: {period['temperature']}°{period['temperatureUnit']}
风力: {period['windSpeed']} {period['windDirection']}
预报: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    # 用分隔符连接所有预报
    return "\n---\n".join(forecasts)

<tool_name>
<parameter1_name>value1</parameter1_name>
<parameter2_name>value2</parameter2_name>
...
</tool_name>

<read_file>
src/main.js
</read_file>

<use_mcp_tool>
<server_name>服务器名称在此</server_name>
<tool_name>工具名称在此</tool_name>

{
"param1": "value1",
"param2": "value2"
}

</use_mcp_tool>

builder = StateGraph(State)

# Initialize each node with node_secret 
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("b2", ReturnNodeValue("I'm B2"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))

# Flow
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b2")
builder.add_edge(["b2", "c"], "d")
builder.add_edge("d", END)
graph = builder.compile()

display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"state": []})

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm C to ["I'm A"]
Adding I'm B2 to ["I'm A", "I'm B", "I'm C"]
Adding I'm D to ["I'm A", "I'm B", "I'm C", "I'm B2"]
{'state': ["I'm A", "I'm B", "I'm C", "I'm B2", "I'm D"]}

def sorting_reducer(left, right):
    """ 合并并排序列表中的值"""
    # 如果 left 不是列表，则将其转换为列表
    if not isinstance(left, list):
        left = [left]

    # 如果 right 不是列表，则将其转换为列表
    if not isinstance(right, list):
        right = [right]
    
    # 合并 left 和 right 列表，然后升序排序
    return sorted(left + right, reverse=False)
class State(TypedDict):
    # sorting_reducer 函数将对 state 中的值进行排序
    state: Annotated[list, sorting_reducer]