graph.add_conditional_edges(    "判断节点",           # 从哪个节点开始判断    routing_function,    # 判断逻辑    {        True: "路径A",    # 条件为真走这里        False: "路径B"    # 条件为假走这里    })

from pydantic import BaseModel, Fieldfrom typing import Optionalclass UserInfo(BaseModel):    """用户信息模型"""    name: str = Field(description="用户姓名")    age: Optional[int] = Field(description="用户年龄")    email: str = Field(description="邮箱地址")    phone: Optional[str] = Field(description="手机号")

structured_llm = llm.with_structured_output(UserInfo)result = structured_llm.invoke("我叫张三，28岁，邮箱zhangsan@qq.com")# result会是一个UserInfo对象，字段都填好了

pip install langgraph langchain-openai sqlalchemy pymysql

import osfrom typing import Union, Optional, Annotatedfrom pydantic import BaseModel, Fieldfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, END, STARTfrom langchain_core.messages import HumanMessage, AnyMessagefrom typing import TypedDictimport operator# ========== 1. 配置 ==========os.environ["OPENAI_API_KEY"] = "your-api-key-here"llm = ChatOpenAI(model="gpt-4o-mini")# ========== 2. 定义数据模型 ==========class UserInfo(BaseModel):    """用户信息，包括姓名、年龄、邮箱和手机号"""    name: str = Field(description="用户的姓名")    age: Optional[int] = Field(description="用户的年龄")    email: str = Field(description="用户的邮箱地址")    phone: Optional[str] = Field(description="用户的手机号")class ConversationalResponse(BaseModel):    """普通对话回复"""    response: str = Field(description="对用户的回复内容")class FinalResponse(BaseModel):    """最终输出，可能是用户信息或普通回复"""    final_output: Union[UserInfo, ConversationalResponse]# ========== 3. 定义图的状态 ==========class AgentState(TypedDict):    messages: Annotated[list[AnyMessage], operator.add]# ========== 4. 定义节点函数 ==========def chat_with_model(state):    """第一步：让AI理解用户输入并生成结构化输出"""    print(f"[chat_with_model] 收到消息: {state['messages'][-1].content}")        messages = state['messages']    structured_llm = llm.with_structured_output(FinalResponse)    response = structured_llm.invoke(messages)        print(f"[chat_with_model] AI判断结果: {type(response.final_output).__name__}")    return {"messages": [response]}def final_answer(state):    """处理普通对话"""    print("[final_answer] 生成普通回复")        messages = state['messages'][-1]    response = messages.final_output.response        print(f"[final_answer] 回复内容: {response}")    return {"messages": [response]}def insert_db(state):    """处理用户信息存储"""    print("[insert_db] 准备存储用户信息")        result = state['messages'][-1]    output = result.final_output        # 这里简化处理，实际项目需要真实的数据库操作    print(f"[insert_db] 存储信息: 姓名={output.name}, 年龄={output.age}, "          f"邮箱={output.email}, 手机={output.phone}")        return {"messages": [f"✅ 已成功记录您的信息：{output.name}"]}# ========== 5. 定义路由函数 ==========def generate_branch(state: AgentState):    """判断走哪条路：存储信息 or 普通回复"""    result = state['messages'][-1]    output = result.final_output        if isinstance(output, UserInfo):        print("[Router] 检测到用户信息，路由到数据库")        return "insert_db"    else:        print("[Router] 检测到普通对话，路由到回复")        return "final_answer"# ========== 6. 构建图 ==========graph_builder = StateGraph(AgentState)# 添加节点graph_builder.add_node("chat_with_model", chat_with_model)graph_builder.add_node("final_answer", final_answer)graph_builder.add_node("insert_db", insert_db)# 设置起点graph_builder.add_edge(START, "chat_with_model")# 添加条件边（核心路由逻辑）graph_builder.add_conditional_edges(    "chat_with_model",    generate_branch,    {        "insert_db": "insert_db",        "final_answer": "final_answer"    })# 设置终点graph_builder.add_edge("final_answer", END)graph_builder.add_edge("insert_db", END)# 编译图graph = graph_builder.compile()# ========== 7. 测试函数 ==========def test_agent(query):    print("\n" + "="*60)    print(f"用户输入: {query}")    print("="*60)        input_message = {"messages": [HumanMessage(content=query)]}    result = graph.invoke(input_message)        print("\n最终结果:", result["messages"][-1])    print("="*60 + "\n")# ========== 8. 运行测试 ==========if __name__ == "__main__":    # 测试1：普通对话    test_agent("你好，请介绍一下你自己")        # 测试2：用户信息    test_agent("我叫张三，28岁，邮箱zhangsan@qq.com，手机13812345678")        # 测试3：另一个普通问题    test_agent("今天天气怎么样？")

============================================================用户输入: 你好，请介绍一下你自己============================================================[chat_with_model] 收到消息: 你好，请介绍一下你自己[chat_with_model] AI判断结果: ConversationalResponse[Router] 检测到普通对话，路由到回复[final_answer] 生成普通回复[final_answer] 回复内容: 你好！我是一个AI助手...

你好！我是一个AI助手...========================================================================================================================用户输入: 我叫张三，28岁，邮箱zhangsan@qq.com，手机13812345678============================================================[chat_with_model] 收到消息: 我叫张三，28岁，邮箱zhangsan@qq.com，手机13812345678[chat_with_model] AI判断结果: UserInfo[Router] 检测到用户信息，路由到数据库[insert_db] 准备存储用户信息[insert_db] 存储信息: 姓名=张三, 年龄=28, 邮箱=zhangsan@qq.com, 手机=13812345678✅ 已成功记录您的信息：张三============================================================

{"messages": [HumanMessage(content="我叫张三...")]}

if isinstance(output, UserInfo):    return "insert_db"  # 走数据库存储else:    return "final_answer"  # 走普通回复

from langchain_core.tools import tool@tooldef search_web(query: str):    """搜索互联网获取最新信息"""    # 实际调用搜索API    return "搜索结果..."@tooldef get_weather(city: str):    """查询城市天气"""    if city == "北京":        return "北京今天16度，晴"    return "未知城市"

tools = [search_web, get_weather]llm_with_tools = llm.bind_tools(tools)

response = llm_with_tools.invoke("北京天气怎么样？")# AI会自动生成：get_weather(city="北京")

import osimport jsonimport requestsfrom typing import Optional, Annotatedfrom pydantic import BaseModel, Fieldfrom langchain_openai import ChatOpenAIfrom langchain_core.tools import toolfrom langgraph.graph import StateGraph, END, STARTfrom langgraph.prebuilt import ToolNodefrom langchain_core.messages import HumanMessage, AnyMessage, AIMessagefrom typing import TypedDictimport operator# ========== 1. 配置 ==========os.environ["OPENAI_API_KEY"] = "your-api-key-here"llm = ChatOpenAI(model="gpt-4o-mini")# ========== 2. 定义工具 ==========@tooldef search_web(query: str):    """搜索互联网获取最新信息        Args:        query: 搜索关键词    """    print(f"[Tool] 正在搜索: {query}")        # 这里用一个简化的实现    # 实际项目可以接入真实的搜索API    return f"关于'{query}'的最新搜索结果：[这里是模拟的搜索结果]"@tooldef get_weather(city: str):    """查询城市天气        Args:        city: 城市名称    """    print(f"[Tool] 正在查询天气: {city}")        weather_data = {        "北京": "北京今天16度，天气晴朗",        "上海": "上海今天20度，多云",        "深圳": "深圳今天28度，有雨"    }        return weather_data.get(city, f"抱歉，暂时没有{city}的天气信息")@tooldef save_user_info(name: str, age: int, email: str, phone: str):    """保存用户信息到数据库        Args:        name: 用户姓名        age: 用户年龄        email: 邮箱地址        phone: 手机号    """    print(f"[Tool] 正在保存用户信息: {name}")        # 实际项目这里应该是真实的数据库操作    print(f"  - 姓名: {name}")    print(f"  - 年龄: {age}")    print(f"  - 邮箱: {email}")    print(f"  - 手机: {phone}")        return f"✅ 已成功保存 {name} 的信息"# ========== 3. 创建工具节点 ==========tools = [search_web, get_weather, save_user_info]tool_node = ToolNode(tools)# 绑定工具到模型llm_with_tools = llm.bind_tools(tools)# ========== 4. 定义图的状态 ==========class AgentState(TypedDict):    messages: Annotated[list[AnyMessage], operator.add]# ========== 5. 定义节点函数 ==========def call_model(state):    """调用大模型，让它决定要不要用工具"""    print(f"\n[call_model] 收到消息: {state['messages'][-1].content}")        messages = state['messages']    response = llm_with_tools.invoke(messages)        # 检查AI是否要调用工具    if response.tool_calls:        print(f"[call_model] AI决定调用工具: {[tc['name'] for tc in response.tool_calls]}")    else:        print("[call_model] AI决定直接回答")        return {"messages": [response]}# ========== 6. 定义路由函数 ==========def should_continue(state: AgentState):    """判断是否需要调用工具"""    messages = state["messages"]    last_message = messages[-1]        # 如果AI生成了tool_calls，就去执行工具    if last_message.tool_calls:        return "tools"    # 否则直接结束    return END# ========== 7. 构建图 ==========workflow = StateGraph(AgentState)# 添加节点workflow.add_node("agent", call_model)  # AI决策节点workflow.add_node("tools", tool_node)    # 工具执行节点# 设置入口workflow.add_edge(START, "agent")# 添加条件边：AI决定后，要么调用工具，要么结束workflow.add_conditional_edges(    "agent",    should_continue,    {        "tools": "tools",        END: END    })# 工具执行完后，回到AI节点让它总结结果workflow.add_edge("tools", "agent")# 编译graph = workflow.compile()# ========== 8. 测试函数 ==========def test_agent(query):    print("\n" + "="*70)    print(f"👤 用户: {query}")    print("="*70)        result = graph.invoke(        {"messages": [HumanMessage(content=query)]},        {"recursion_limit": 10}  # 防止无限循环    )        final_answer = result["messages"][-1].content    print(f"\n🤖 助手: {final_answer}")    print("="*70 + "\n")# ========== 9. 运行测试 ==========if __name__ == "__main__":    # 测试1：普通对话    test_agent("你好，请介绍一下你自己")        # 测试2：天气查询    test_agent("北京今天天气怎么样？")        # 测试3：联网搜索    test_agent("Claude 4.5 Sonnet有什么新功能？")        # 测试4：用户信息    test_agent("我叫李四，25岁，邮箱lisi@example.com，手机13987654321")        # 测试5：复杂任务（可能调用多个工具）    test_agent("帮我查一下上海的天气，然后搜索一下最近的AI新闻")

======================================================================👤 用户: 北京今天天气怎么样？======================================================================[call_model] 收到消息: 北京今天天气怎么样？[call_model] AI决定调用工具: ['get_weather'][Tool] 正在查询天气: 北京[call_model] 收到消息: ToolMessage(content='北京今天16度，天气晴朗')[call_model] AI决定直接回答🤖 助手: 北京今天的天气是16度，天气晴朗。============================================================================================================================================👤 用户: 我叫李四，25岁，邮箱lisi@example.com，手机13987654321======================================================================[call_model] 收到消息: 我叫李四，25岁，邮箱lisi@example.com，手机13987654321[call_model] AI决定调用工具: ['save_user_info'][Tool] 正在保存用户信息: 李四  - 姓名: 李四  - 年龄: 25  - 邮箱: lisi@example.com  - 手机: 13987654321[call_model] 收到消息: ToolMessage(content='✅ 已成功保存 李四 的信息')[call_model] AI决定直接回答🤖 助手: 您的信息已经成功保存，李四！如果还有其他需要帮助的地方，请随时告诉我。======================================================================

{    "name": "get_weather",    "args": {"city": "北京"}}

def tool_node(state):    results = []    for tool_call in state["messages"][-1].tool_calls:        # 找到对应的工具        tool = tools_by_name[tool_call["name"]]        # 执行工具        result = tool.invoke(tool_call["args"])        # 包装成ToolMessage        results.append(ToolMessage(content=result, ...))    return {"messages": results}

# ❌ 不好的描述@tooldef func1(x):    """一个函数"""    pass# ✅ 好的描述@tooldef search_flights(departure: str, destination: str, date: str):    """搜索航班信息        Args:        departure: 出发城市，如"北京"        destination: 目的地城市，如"上海"          date: 日期，格式YYYY-MM-DD，如"2024-03-15"        Returns:        返回可用航班列表，包含时间、价格等信息    """    pass

@tooldef get_weather(city: str):    """查询天气"""    try:        # 实际的API调用        result = weather_api.get(city)        return result    except Exception as e:        # 返回友好的错误信息        return f"抱歉，查询{city}的天气时出错了：{str(e)}"

result = graph.invoke(    {"messages": [HumanMessage(content=query)]},    {"recursion_limit": 10}  # 最多10次迭代)

def call_tool(state):    try:        result = tool.invoke(args)        return {"messages": [result]}    except Exception as e:        error_msg = f"工具调用失败：{str(e)}"        print(f"[Error] {error_msg}")        return {"messages": [error_msg]}

from functools import lru_cachefrom datetime import datetime, timedelta# 缓存装饰器，5分钟过期CACHE = {}CACHE_EXPIRE = timedelta(minutes=5)def cached_tool(func):    """工具缓存装饰器"""    def wrapper(*args, **kwargs):        # 生成缓存key        cache_key = f"{func.__name__}_{args}_{kwargs}"                # 检查缓存        if cache_key in CACHE:            cached_time, cached_result = CACHE[cache_key]            if datetime.now() - cached_time < CACHE_EXPIRE:                print(f"[Cache] 使用缓存结果")                return cached_result                # 执行函数        result = func(*args, **kwargs)                # 存入缓存        CACHE[cache_key] = (datetime.now(), result)        return result        return wrapper@tool@cached_tooldef get_weather(city: str):    """查询天气（带缓存）"""    # 实际的API调用    pass

from concurrent.futures import ThreadPoolExecutordef execute_tools_parallel(tool_calls):    """并行执行多个工具"""    with ThreadPoolExecutor(max_workers=5) as executor:        futures = []        for tool_call in tool_calls:            future = executor.submit(                execute_single_tool,                tool_call            )            futures.append(future)                results = [f.result() for f in futures]        return results

# 根据任务复杂度选择模型def get_model(task_type):    if task_type in ["simple_chat", "weather", "todo"]:        return ChatOpenAI(model="gpt-4o-mini")  # 快且便宜    else:        return ChatOpenAI(model="gpt-4o")  # 复杂任务用完整版

def stream_response(self, user_input):    """流式输出响应"""    for chunk in self.llm.stream(user_input):        print(chunk.content, end="", flush=True)

SYSTEM_PROMPT = """重要：当用户询问天气、日程、待办等信息时，你必须调用相应的工具获取准确信息，不要自己编造答案！可用工具：- get_weather: 查询实时天气- list_todos: 查看待办事项..."""

@tooldef add_schedule(date: str, time: str):    """添加日程        Args:        date: 日期，必须是YYYY-MM-DD格式，例如"2024-03-15"        time: 时间，必须是HH:MM格式，例如"14:30"        注意：请严格按照格式要求传参！    """    # 加上格式验证    import re    if not re.match(r'\d{4}-\d{2}-\d{2}', date):        return "❌ 日期格式错误，请使用YYYY-MM-DD格式"    if not re.match(r'\d{2}:\d{2}', time):        return "❌ 时间格式错误，请使用HH:MM格式"        # 正常处理    pass

# 1. 设置最大迭代次数{"recursion_limit": 10}# 2. 工具返回清晰的信息@tooldef search_web(query):    result = api.search(query)    return f"搜索完成！结果：{result}\n请基于以上信息回答用户问题。"

# 1. 异步执行import asyncioasync def call_tool_async(tool_call):    return await tool.ainvoke(tool_call["args"])# 2. 添加加载提示def chat_with_loading(user_input):    print("🤖 小智正在思考...")    result = self.chat(user_input)    return result# 3. 用流式输出def stream_chat(user_input):    print("🤖 小智：", end="")    for chunk in model.stream(user_input):        print(chunk, end="", flush=True)

@tooldef robust_search(query: str):    """带重试机制的搜索"""    max_retries = 3        for i in range(max_retries):        try:            result = search_api.get(query, timeout=5)            return result        except TimeoutError:            if i < max

# 安装核心包pip install langgraph langchain-openai# 安装可视化工具（可选，但强烈推荐）pip install pyppeteer ipython

import osimport getpass# 设置OpenAI API Keyif not os.environ.get("OPENAI_API_KEY"):    os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入你的OpenAI API Key: ")

def calculator(x):    x = x + 1  # 第一步    y = x - 2  # 第二步    return x, yresult = calculator(10)print(result)  # (11, 9)

from langgraph.graph import StateGraph, START, END# 定义第一个节点：加法def addition(state):    """    这个函数会收到当前的state（一个字典）    从里面取出x的值，加1后返回    """    print(f"加法节点收到的state: {state}")    current_x = state["x"]    new_x = current_x + 1    # 只返回更新的部分，不用担心其他数据会丢    return {"x": new_x}# 定义第二个节点：减法def subtraction(state):    """    这个函数也会收到state    注意：这里的state已经包含了上一个节点更新后的x值    """    print(f"减法节点收到的state: {state}")    current_x = state["x"]    y = current_x - 2    # 这里我们新增了一个y，x会保留    return {"y": y}# 开始构建图# StateGraph(dict)表示我们的state是一个普通字典builder = StateGraph(dict)# 添加两个节点# 第一个参数是节点名称，第二个参数是对应的函数builder.add_node("addition", addition)builder.add_node("subtraction", subtraction)# 定义节点之间的连接关系（边）# START -> addition -> subtraction -> ENDbuilder.add_edge(START, "addition")        # 起点连到additionbuilder.add_edge("addition", "subtraction") # addition连到subtractionbuilder.add_edge("subtraction", END)        # subtraction连到终点# 编译成可执行的图graph = builder.compile()# 运行！传入初始状态initial_state = {"x": 10}final_state = graph.invoke(initial_state)print(f"\n最终状态: {final_state}")

from IPython.display import Image, display# 生成图的可视化# xray=True表示显示详细信息display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

def buggy_function(state):    # 假设state里没有"z"这个key    return {"result": state["z"] + 1}  # 💥运行时报错为了避免这种问题，实际开发中我们用TypedDict来定义State的结构：from typing_extensions import TypedDict# 定义State的结构class State(TypedDict):    x: int  # x是整数    y: int  # y也是整数# 现在用State来构建图builder = StateGraph(State)# 节点函数保持不变def addition(state: State):  # 加上类型标注    return {"x": state["x"] + 1}def subtraction(state: State):    return {"y": state["x"] - 2}# 后面的代码完全一样builder.add_node("addition", addition)builder.add_node("subtraction", subtraction)builder.add_edge(START, "addition")builder.add_edge("addition", "subtraction")builder.add_edge("subtraction", END)graph = builder.compile()# 运行result = graph.invoke({"x": 10})print(result)  # {'x': 11, 'y': 9}

import operatorfrom typing import Annotated, Listfrom typing_extensions import TypedDict# 注意这里的写法class State(TypedDict):    # Annotated[类型, Reducer函数]    messages: Annotated[List[dict], operator.add]# 测试一下def node1(state):    # 返回一个包含新消息的列表    return {"messages": [{"role": "user", "content": "你好"}]}def node2(state):    print(f"node2收到的消息: {state['messages']}")    # 再追加一条消息    return {"messages": [{"role": "assistant", "content": "你好啊"}]}builder = StateGraph(State)builder.add_node("node1", node1)builder.add_node("node2", node2)builder.add_edge(START, "node1")builder.add_edge("node1", "node2")builder.add_edge("node2", END)graph = builder.compile()# 初始状态包含一个空列表result = graph.invoke({"messages": []})print(f"\n最终消息列表: {result['messages']}")

node2收到的消息: [{'role': 'user', 'content': '你好'}]最终消息列表: [    {'role': 'user', 'content': '你好'},    {'role': 'assistant', 'content': '你好啊'}]

import osimport getpassfrom typing import Annotatedfrom typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessagefrom IPython.display import Image, display# 配置API Keyif not os.environ.get("OPENAI_API_KEY"):    os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入OpenAI API Key: ")# 定义State结构class State(TypedDict):    # add_messages是LangGraph提供的专门处理消息的Reducer    messages: Annotated[list, add_messages]# 创建GPT模型实例llm = ChatOpenAI(model="gpt-4o")# 定义聊天节点def chatbot(state: State):    """    这个函数接收包含对话历史的state    把所有消息发给GPT，获取回复    """    # state["messages"]是一个消息列表    # 直接传给模型就行    response = llm.invoke(state["messages"])        # 返回模型的回复，add_messages会自动追加到列表    return {"messages": [response]}# 构建图builder = StateGraph(State)builder.add_node("chatbot", chatbot)builder.add_edge(START, "chatbot")builder.add_edge("chatbot", END)# 编译graph = builder.compile()# 可视化看看（可选）# display(Image(graph.get_graph(xray=True).draw_mermaid_png()))# 单次对话测试def test_single():    user_message = HumanMessage(content="你好，请介绍一下你自己")    result = graph.invoke({"messages": [user_message]})    print("用户:", user_message.content)    print("AI:", result["messages"][-1].content)# 连续对话测试def chat_loop():    print("聊天机器人已启动（输入'退出'结束对话）\n")        while True:        user_input = input("你: ").strip()                if user_input == "退出":            print("再见！")            break                if not user_input:            continue                # 每次都创建新的对话        # 如果要保持对话历史，需要把messages保存下来        for event in graph.stream({"messages": [HumanMessage(content=user_input)]}):            for value in event.values():                print(f"AI: {value['messages'][-1].content}\n")# 运行if __name__ == "__main__":    # test_single()  # 测试单次对话    chat_loop()      # 开启连续对话

def chat_with_memory():    """带对话历史的聊天"""    print("聊天机器人已启动（输入'退出'结束对话）\n")        # 用这个列表保存整个对话历史    conversation_history = []        while True:        user_input = input("你: ").strip()                if user_input == "退出":            print("再见！")            break                if not user_input:            continue                # 把用户消息加入历史        conversation_history.append(HumanMessage(content=user_input))                # 用完整的历史调用模型        result = graph.invoke({"messages": conversation_history})                # 获取AI回复        ai_message = result["messages"][-1]        print(f"AI: {ai_message.content}\n")                # 把AI回复也加入历史        conversation_history.append(ai_message)# 运行带记忆的版本if __name__ == "__main__":    chat_with_memory()

from langgraph.graph.message import add_messagesfrom langchain_core.messages import HumanMessage, AIMessage# 场景1：追加新消息msg1 = [HumanMessage(content="你好", id="1")]msg2 = [AIMessage(content="你好啊", id="2")]result = add_messages(msg1, msg2)print("场景1 - 追加:", [m.content for m in result])# 输出: ['你好', '你好啊']# 场景2：更新已有消息msg1 = [HumanMessage(content="你好", id="1")]msg2 = [HumanMessage(content="你好呀", id="1")]  # 注意ID相同result = add_messages(msg1, msg2)print("场景2 - 更新:", [m.content for m in result])# 输出: ['你好呀']  # 第一条消息被更新了

def add_messages(left, right):    """    left: 已有的消息列表    right: 新来的消息列表    """    # 1. 确保都是列表    if not isinstance(left, list):        left = [left]    if not isinstance(right, list):        right = [right]        # 2. 给每条消息分配ID（如果没有的话）    for m in left:        if m.id is None:            m.id = str(uuid.uuid4())    for m in right:        if m.id is None:            m.id = str(uuid.uuid4())        # 3. 建立ID到消息的映射    left_idx_by_id = {m.id: i for i, m in enumerate(left)}        # 4. 合并：ID相同就更新，不同就追加    merged = left.copy()    for m in right:        if m.id in left_idx_by_id:            # ID存在，更新            merged[left_idx_by_id[m.id]] = m        else:            # ID不存在，追加            merged.append(m)        return merged

import osimport getpassfrom typing import Annotatedfrom typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, SystemMessage# ===== 配置部分 =====if not os.environ.get("OPENAI_API_KEY"):    os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入OpenAI API Key: ")# ===== State定义 =====class ChatState(TypedDict):    messages: Annotated[list, add_messages]# ===== 创建模型 =====llm = ChatOpenAI(    model="gpt-4o",    temperature=0.7,  # 控制回复的创造性)# ===== 定义节点 =====def chatbot_node(state: ChatState):    """处理用户消息，生成回复"""    messages = state["messages"]    response = llm.invoke(messages)    return {"messages": [response]}# ===== 构建图 =====builder = StateGraph(ChatState)builder.add_node("chatbot", chatbot_node)builder.add_edge(START, "chatbot")builder.add_edge("chatbot", END)graph = builder.compile()# ===== 主程序 =====def main():    print("=" * 50)    print("智能聊天助手 v1.0")    print("=" * 50)    print("提示：输入'退出'结束对话\n")        # 设置系统提示词    conversation = [        SystemMessage(content="你是一个友好、专业的AI助手。回答要简洁明了。")    ]        while True:        # 获取用户输入        user_input = input("👤 你: ").strip()                # 退出判断        if user_input.lower() in ["退出", "exit", "quit"]:            print("\n👋 感谢使用，再见！")            break                # 跳过空输入        if not user_input:            continue                # 添加用户消息        conversation.append(HumanMessage(content=user_input))                try:            # 调用图进行处理            result = graph.invoke({"messages": conversation})                        # 获取AI回复            ai_response = result["messages"][-1]            conversation.append(ai_response)                        # 显示回复            print(f"🤖 AI: {ai_response.content}\n")                    except Exception as e:            print(f"❌ 出错了: {str(e)}\n")            # 出错时移除最后添加的用户消息            conversation.pop()if __name__ == "__main__":    main()

# 安装必要的库# pip install langchain neo4j langchain-openai langchain-experimentalfrom langchain_core.documents import Document# 准备你的数据（可以是公司信息文档）content = """小米科技有限责任公司是一家专注于研发和推出创新技术的公司。小米推出了智能家居产品和5G技术。华为技术有限公司与清华大学建立了深度合作。华为参与了5G标准的制定。小米与美团合作开发物流解决方案。"""# 转化为Document对象documents = [Document(page_content=content)]print(f"加载了{len(documents)}份文档")

from langchain_community.graphs import Neo4jGraphfrom langchain_experimental.graph_transformers import LLMGraphTransformerfrom langchain_openai import ChatOpenAI# 连接到Neo4j数据库（使用免费的Neo4j Aura）graph = Neo4jGraph(    url="neo4j+s://your-db-uri",  # 替换为你的Neo4j URI    username="neo4j",    password="your-password",      # 替换为你的密码    database="neo4j")# 初始化LLMgraph_llm = ChatOpenAI(temperature=0, model_name="gpt-4o-mini")# 创建图转换器，定义你要提取的实体和关系类型graph_transformer = LLMGraphTransformer(    llm=graph_llm,    allowed_nodes=["公司", "产品", "技术", "教育机构", "合作伙伴"],    allowed_relationships=["推出", "参与", "合作", "位于", "开发"],)# 执行转换：从文档中提取实体和关系graph_documents = graph_transformer.convert_to_graph_documents(documents)# 将图数据导入Neo4jgraph.add_graph_documents(graph_documents)print(f"成功导入{len(graph_documents)}个图文档")print(f"提取的实体: {[node.id for node in graph_documents[0].nodes]}")print(f"提取的关系: {[(rel.source.id, rel.type, rel.target.id) for rel in graph_documents[0].relationships]}")

from langchain.chains import GraphCypherQAChainllm = ChatOpenAI(model="gpt-4o-mini", temperature=0)# 创建Cypher查询链cypher_chain = GraphCypherQAChain.from_llm(    graph=graph,    cypher_llm=llm,    qa_llm=llm,    validate_cypher=True,    verbose=True,  # 设置为True可以看到生成的Cypher查询过程    allow_dangerous_requests=True)# 测试几个问题queries = [    "小米科技有限责任公司推出了哪些创新技术？",    "华为技术有限公司与哪些教育机构建立了合作？",    "都有哪些公司在我的数据库中？"]for query in queries:    print(f"\n问题: {query}")    response = cypher_chain.invoke(query)    print(f"答案: {response['result']}")    print("-" * 50)

from langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_openai import OpenAIEmbeddingsfrom langchain_milvus import Milvus# 分割文本text_splitter = RecursiveCharacterTextSplitter(    chunk_size=250,    chunk_overlap=30)splits = text_splitter.split_documents(documents)# 创建向量嵌入embeddings = OpenAIEmbeddings(model="text-embedding-3-large")# 连接到Milvus向量数据库（使用免费的Zilliz云服务）vectorstore = Milvus.from_documents(    documents=splits,    collection_name="company_rag_milvus",    embedding=embeddings,    connection_args={        "uri": "https://your-milvus-uri",  # 替换为你的Milvus URI        "user": "your-username",        "password": "your-password",    })print(f"向量数据库已创建，包含{len(splits)}个文本块")

from langchain.prompts import PromptTemplatefrom langchain_core.output_parsers import StrOutputParser# 定义RAG提示词prompt = PromptTemplate(    template="""你是一个问答助手。使用以下检索到的上下文来回答问题。如果你不知道答案，就说你不知道。最多使用三句话，保持简洁：问题: {question}上下文: {context}答案:""",    input_variables=["question", "context"],)# 构建RAG链rag_chain = prompt | graph_llm | StrOutputParser()# 执行查询question = "我的知识库中都有哪些公司信息？"retriever = vectorstore.as_retriever(search_kwargs={"k": 2})# 检索相关文档docs = retriever.invoke(question)# 生成回答generation = rag_chain.invoke({    "context": "\n\n".join([doc.page_content for doc in docs]),    "question": question})print(f"RAG系统的回答: {generation}")

from langgraph.graph import StateGraph, MessagesState, START, ENDfrom langchain_core.messages import HumanMessagefrom typing import Literalfrom typing_extensions import TypedDictclass AgentState(MessagesState):    next: strclass Router(TypedDict):    """路由到不同的子代理"""    next: Literal["graph_kg", "vec_kg", "FINISH"]# 图数据库代理def graph_kg_agent(state: AgentState):    messages = state["messages"][-1]    response = cypher_chain.invoke(messages.content)    return {        "messages": [HumanMessage(content=response["result"], name="graph_kg")]    }# 向量数据库代理def vec_kg_agent(state: AgentState):    messages = state["messages"][-1]        retriever = vectorstore.as_retriever(search_kwargs={"k": 2})    docs = retriever.invoke(messages.content)        generation = rag_chain.invoke({        "context": "\n\n".join([doc.page_content for doc in docs]),        "question": messages.content    })        return {        "messages": [HumanMessage(content=generation, name="vec_kg")]    }# 主管代理：判断用何种方式回答def supervisor(state: AgentState):    system_prompt = """你是一个任务主管，管理两个工作者：- graph_kg: 基于知识图，擅长回答全局、综合性问题- vec_kg: 基于向量检索，擅长回答细节问题根据用户问题，判断应该使用哪个工作者。"""        messages = [{"role": "system", "content": system_prompt}] + state["messages"]        response = graph_llm.with_structured_output(Router).invoke(messages)        next_worker = response["next"]    if next_worker == "FINISH":        next_worker = END        return {"next": next_worker}# 构建图builder = StateGraph(AgentState)builder.add_node("supervisor", supervisor)builder.add_node("graph_kg", graph_kg_agent)builder.add_node("vec_kg", vec_kg_agent)# 设置工作流builder.add_edge(START, "supervisor")for worker in ["graph_kg", "vec_kg"]:    builder.add_edge(worker, "supervisor")builder.add_conditional_edges("supervisor", lambda state: state["next"])# 编译并运行multi_agent_graph = builder.compile()# 测试多代理系统test_queries = [    "都有哪些公司在我的数据库中？",  # 应该用graph_kg    "小米推出了什么技术？"            # 应该用graph_kg]for query in test_queries:    print(f"\n用户问题: {query}")    for output in multi_agent_graph.stream(        {"messages": query},        stream_mode="values"    ):        last_message = output["messages"][-1]        print(f"{last_message.name}: {last_message.content}")    print("-" * 50)

chromedriver-mac-arm64/├── chromedriver          (可执行文件)└── LICENSE.chromedriver  (许可证文件)

chmod +x chromedriver

# 方法1: 移动到系统路径(需要sudo权限)sudo mv chromedriver /usr/local/bin/# 方法2: 记住绝对路径,代码中指定/Users/yourname/Downloads/chromedriver-mac-arm64/chromedriver

pip install selenium

pip install selenium -i https://pypi.tuna.tsinghua.edu.cn/simple

pip show selenium

Name: seleniumVersion: 4.15.2

from selenium import webdriver# 创建Chrome浏览器实例driver = webdriver.Chrome()# 打开百度首页driver.get("https://www.baidu.com")# 打印页面标题print("页面标题:", driver.title)# 等待5秒(让你看到效果)import timetime.sleep(5)# 关闭浏览器driver.quit()

from selenium import webdriver

driver = webdriver.Chrome()

driver = webdriver.Chrome(executable_path='/path/to/chromedriver')

driver.get("https://www.baidu.com")

print("页面标题:", driver.title)

driver.quit()

python test.py

from selenium.webdriver.common.by import By

element = driver.find_element(By.ID, "kw")

element = driver.find_element(By.NAME, "wd")

element = driver.find_element(By.CLASS_NAME, "s_ipt")

element = driver.find_element(By.TAG_NAME, "input")

element = driver.find_element(By.LINK_TEXT, "新闻")

element = driver.find_element(By.PARTIAL_LINK_TEXT, "新")

element = driver.find_element(By.CSS_SELECTOR, "#kw")element = driver.find_element(By.CSS_SELECTOR, "input.s_ipt")element = driver.find_element(By.CSS_SELECTOR, "form > input[name='wd']")

element = driver.find_element(By.XPATH, "//input[@id='kw']")element = driver.find_element(By.XPATH, "//form//input[@name='wd']")

# 方式1: 通过ID(推荐)search_box = driver.find_element(By.ID, "kw")# 方式2: 通过NAMEsearch_box = driver.find_element(By.NAME, "wd")# 方式3: 通过CLASSsearch_box = driver.find_element(By.CLASS_NAME, "s_ipt")# 方式4: 通过CSS选择器search_box = driver.find_element(By.CSS_SELECTOR, "#kw")search_box = driver.find_element(By.CSS_SELECTOR, "input.s_ipt")# 方式5: 通过XPathsearch_box = driver.find_element(By.XPATH, "//input[@id='kw']")

driver.get("https://www.baidu.com")search_box = driver.find_element(By.ID, "kw")  # ❌ 可能报错!

import timedriver.get("https://www.baidu.com")time.sleep(5)  # 强制等待5秒search_box = driver.find_element(By.ID, "kw")

driver.implicitly_wait(10)  # 设置全局隐式等待10秒driver.get("https://www.baidu.com")search_box = driver.find_element(By.ID, "kw")  # 会自动等待元素出现

from selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECfrom selenium.webdriver.common.by import Bydriver.get("https://www.baidu.com")# 显式等待搜索框出现,最多等10秒search_box = WebDriverWait(driver, 10).until(    EC.presence_of_element_located((By.ID, "kw")))

WebDriverWait(driver, 10)  # 创建一个等待器,最多等10秒.until(...)  # 直到某个条件满足EC.presence_of_element_located((By.ID, "kw"))  # 条件:ID为kw的元素出现在DOM中常用的等待条件:# 1. 元素出现在DOM中(不一定可见)EC.presence_of_element_located((By.ID, "kw"))# 2. 元素可见EC.visibility_of_element_located((By.ID, "kw"))# 3. 元素可点击EC.element_to_be_clickable((By.ID, "su"))# 4. 页面标题包含某文字EC.title_contains("百度")# 5. 元素的文本包含某内容EC.text_to_be_present_in_element((By.ID, "result"), "搜索结果")# 6. 元素存在于DOM中(多个元素)EC.presence_of_all_elements_located((By.CLASS_NAME, "result"))

search_box = driver.find_element(By.ID, "kw")search_box.send_keys("Python自动化")

# 模拟键盘按键from selenium.webdriver.common.keys import Keyssearch_box.send_keys("Python")search_box.send_keys(Keys.ENTER)  # 按回车键search_box.send_keys(Keys.CONTROL, "a")  # Ctrl+A全选search_box.send_keys(Keys.BACK_SPACE)  # 退格键

search_btn = driver.find_element(By.ID, "su")search_btn.click()等待元素可点击:search_btn = WebDriverWait(driver, 10).until(    EC.element_to_be_clickable((By.ID, "su")))search_btn.click()

search_box.clear()  # 清空输入框内容search_box.send_keys("新的搜索词")

# 获取元素文本内容text = element.text# 获取属性值value = element.get_attribute("value")href = element.get_attribute("href")class_name = element.get_attribute("class")# 判断元素状态is_displayed = element.is_displayed()  # 是否可见is_enabled = element.is_enabled()      # 是否可用is_selected = element.is_selected()    # 是否被选中(针对checkbox)

# 滚动到页面底部driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")# 滚动到指定元素element = driver.find_element(By.ID, "footer")driver.execute_script("arguments[0].scrollIntoView();", element)# 获取页面高度height = driver.execute_script("return document.body.scrollHeight")# 修改元素属性driver.execute_script("arguments[0].setAttribute('value', '新值')", element)

from selenium import webdriverfrom selenium.webdriver.common.by import Byfrom selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECimport timedef search_in_baidu(content: str) -> str:    """    在百度搜索指定内容,并获取前3页的搜索结果        参数:        content: 搜索关键词        返回:        str: 包含3页搜索结果的文本内容    """        # 第一步:创建浏览器实例    print("正在启动Chrome浏览器...")    driver = webdriver.Chrome()        # 设置窗口大小(可选,但建议设置)    driver.set_window_size(1920, 1080)        # 第二步:访问百度首页    print("正在访问百度首页...")    driver.get("https://www.baidu.com")        try:        # 第三步:等待搜索框出现并输入关键词        print(f"正在输入搜索关键词: {content}")                # 显式等待搜索框出现(最长等待5秒)        search_box = WebDriverWait(driver, 5).until(            EC.presence_of_element_located((By.ID, "kw"))        )                # 输入搜索关键词        search_box.send_keys(content)        print(f"已输入关键词: {content}")                # 第四步:点击搜索按钮        print("正在点击搜索按钮...")                # 等待搜索按钮可点击        search_btn = WebDriverWait(driver, 5).until(            EC.element_to_be_clickable((By.ID, "su"))        )        search_btn.click()                # 第五步:等待搜索结果页面加载完成        # 搜索结果页面的标题会包含搜索关键词        WebDriverWait(driver, 5).until(            EC.title_contains(content[:10])  # 用前10个字符判断,避免关键词太长        )        print(f"搜索结果页面已加载,标题: {driver.title}")                # 用于存储所有页面的文本内容        page_text_list = []                # 第六步:循环获取前3页的内容        for page_num in range(3):            print(f"\n========== 正在处理第 {page_num + 1} 页 ==========")                        # 如果不是第一页,需要先翻页            if page_num > 0:                print(f"正在翻到第 {page_num + 1} 页...")                                try:                    # 等待分页按钮加载完成                    # 百度的分页按钮在 的容器内                    page_buttons = WebDriverWait(driver, 5).until(                        EC.presence_of_all_elements_located(                            (By.CSS_SELECTOR, ".page-inner_2jZi2>a")                        )                    )                                        # 点击对应的页码按钮                    # 第2页对应索引0,第3页对应索引1                    if page_num - 1 < len(page_buttons):                        page_buttons[page_num - 1].click()                        print(f"已点击第 {page_num + 1} 页按钮")                                                # 等待新页面加载完成                        time.sleep(2)                    else:                        print(f"没有找到第 {page_num + 1} 页的按钮,可能搜索结果不足")                        break                                        except Exception as e:                    print(f"翻页失败: {e}")                    break                        # 第七步:等待页面主体内容加载完成            WebDriverWait(driver, 10).until(                EC.presence_of_element_located((By.TAG_NAME, "body"))            )                        # 第八步:滚动页面加载所有动态内容            print("正在滚动页面加载动态内容...")                        # 记录上一次的页面高度            last_height = driver.execute_script("return document.body.scrollHeight")                        scroll_count = 0            while True:                # 滚动到页面底部                driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")                                # 等待内容加载(给页面1秒时间加载新内容)                time.sleep(1)                                # 获取新的页面高度                new_height = driver.execute_script("return document.body.scrollHeight")                                scroll_count += 1                print(f"  第 {scroll_count} 次滚动: 页面高度 {new_height}px")                                # 如果页面高度不再变化,说明已经加载完所有内容                if new_height == last_height:                    print("  页面已滚动到底部,所有内容加载完成")                    break                                # 更新高度记录                last_height = new_height                                # 防止无限循环(最多滚动10次)                if scroll_count >= 10:                    print("  已达到最大滚动次数,停止滚动")                    break                        # 第九步:获取页面文本内容            print("正在提取页面文本内容...")        # 找到页面的body元素        page_content = driver.find_element(By.TAG_NAME, "body")                # 获取整个页面的文本内容        page_text = page_content.text                # 统计文本长度        text_length = len(page_text)        print(f"第 {page_num + 1} 页内容已提取,文本长度: {text_length} 字符")                # 将内容添加到列表中,并添加页码标记        page_text_list.append(f"""==================== 第 {page_num + 1} 页 ===================={page_text}""")        # 短暂延时,避免操作过快        time.sleep(1)        print("\n所有页面处理完成!")    except Exception as e:    # 捕获并打印任何异常    print(f"发生错误: {e}")    # 可以选择截图保存错误现场    driver.save_screenshot("error_screenshot.png")    print("错误截图已保存为 error_screenshot.png")    finally:    # 第十步:无论成功还是失败,最后都要关闭浏览器    print("\n正在关闭浏览器...")    driver.quit()    print("浏览器已关闭")# 将所有页面内容合并返回return '\n'.join(page_text_list)使用示例if name == "main": # 搜索关键词 keyword = "Python自动化测试"print(f"开始搜索: {keyword}\n")# 调用搜索函数results = search_in_baidu(keyword)# 打印结果(只打印前500个字符,避免太长)print("\n搜索结果预览:")print(results[:500])print("...(后续内容省略)")# 可以选择将结果保存到文件with open("search_results.txt", "w", encoding="utf-8") as f:    f.write(results)print("\n完整结果已保存到 search_results.txt")

driver = webdriver.Chrome()driver.set_window_size(1920, 1080)

search_box = WebDriverWait(driver, 5).until(    EC.presence_of_element_located((By.ID, "kw")))search_box.send_keys(content)

if page_num > 0:  # 第一页不需要翻页    # 找到所有分页按钮    page_buttons = WebDriverWait(driver, 5).until(        EC.presence_of_all_elements_located(            (By.CSS_SELECTOR, ".page-inner_2jZi2>a")        )    )        # 点击对应的按钮    # page_num=1时,点击第0个按钮(第2页)    # page_num=2时,点击第1个按钮(第3页)    page_buttons[page_num - 1].click()

last_height = driver.execute_script("return document.body.scrollHeight")while True:    # 滚动到底部    driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")        # 等待加载    time.sleep(1)        # 获取新高度    new_height = driver.execute_script("return document.body.scrollHeight")        # 如果高度不变,说明加载完了    if new_height == last_height:        break        last_height = new_height

try:    # 主要逻辑    ...except Exception as e:    print(f"发生错误: {e}")    driver.save_screenshot("error_screenshot.png")finally:    driver.quit()

python baidu_search.py

开始搜索: Python自动化测试正在启动Chrome浏览器...正在访问百度首页...正在输入搜索关键词: Python自动化测试已输入关键词: Python自动化测试正在点击搜索按钮...搜索结果页面已加载,标题: Python自动化测试_百度搜索========== 正在处理第 1 页 ==========正在滚动页面加载动态内容...  第 1 次滚动: 页面高度 3520px  第 2 次滚动: 页面高度 4850px  第 3 次滚动: 页面高度 4850px  页面已滚动到底部,所有内容加载完成正在提取页面文本内容...第 1 页内容已提取,文本长度: 12543 字符========== 正在处理第 2 页 ==========正在翻到第 2 页...已点击第 2 页按钮正在滚动页面加载动态内容...  第 1 次滚动: 页面高度 3420px  第 2 次滚动: 页面高度 4680px  第 3 次滚动: 页面高度 4680px  页面已滚动到底部,所有内容加载完成正在提取页面文本内容...第 2 页内容已提取,文本长度: 11892 字符========== 正在处理第 3 页 ==========正在翻到第 3 页...已点击第 3 页按钮正在滚动页面加载动态内容...  第 1 次滚动: 页面高度 3380px  第 2 次滚动: 页面高度 4620px  第 3 次滚动: 页面高度 4620px  页面已滚动到底部,所有内容加载完成正在提取页面文本内容...第 3 页内容已提取,文本长度: 11654 字符所有页面处理完成!正在关闭浏览器...浏览器已关闭搜索结果预览:# ==================== 第 1 页 ====================百度首页设置登录...

from selenium.webdriver.chrome.options import Options# 配置Chrome选项chrome_options = Options()chrome_options.add_argument('--headless')  # 无头模式chrome_options.add_argument('--disable-gpu')  # 禁用GPU加速chrome_options.add_argument('--no-sandbox')  # 绕过操作系统沙箱# 创建浏览器实例时传入选项driver = webdriver.Chrome(options=chrome_options)

chrome_options = Options()chrome_options.add_argument('user-agent=Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')driver = webdriver.Chrome(options=chrome_options)

# 等待alert出现alert = WebDriverWait(driver, 10).until(EC.alert_is_present())# 获取alert文本alert_text = alert.textprint(f"Alert内容: {alert_text}")# 接受(点击确定)alert.accept()# 或者取消(点击取消)# alert.dismiss()

# 获取当前窗口句柄main_window = driver.current_window_handle# 点击链接,打开新标签页link.click()# 获取所有窗口句柄all_windows = driver.window_handles# 切换到新窗口for window in all_windows:    if window != main_window:        driver.switch_to.window(window)        break# 操作新窗口...# 切换回主窗口driver.switch_to.window(main_window)

# 切换到iframeiframe = driver.find_element(By.TAG_NAME, "iframe")driver.switch_to.frame(iframe)# 或者通过索引切换driver.switch_to.frame(0)# 或者通过name/id切换driver.switch_to.frame("iframe_name")# 在iframe中操作元素...# 切换回主页面driver.switch_to.default_content()

chrome_options = Options()# 设置下载路径prefs = {    "download.default_directory": "/path/to/download/folder",    "download.prompt_for_download": False,  # 不提示下载    "download.directory_upgrade": True,    "safebrowsing.enabled": True}chrome_options.add_experimental_option("prefs", prefs)driver = webdriver.Chrome(options=chrome_options)

from selenium.webdriver.common.action_chains import ActionChains# 找到需要悬停的元素element = driver.find_element(By.ID, "menu")# 创建动作链actions = ActionChains(driver)# 移动到元素上(悬停)actions.move_to_element(element).perform()# 等待悬停菜单出现time.sleep(1)# 点击悬停后出现的子菜单submenu = driver.find_element(By.ID, "submenu")submenu.click()

from selenium.webdriver.support.select import Select# 找到下拉框select_element = driver.find_element(By.ID, "country")# 创建Select对象select = Select(select_element)# 三种选择方式:# 1. 通过可见文本选择select.select_by_visible_text("中国")# 2. 通过value属性选择select.select_by_value("CN")# 3. 通过索引选择(从0开始)select.select_by_index(2)# 获取当前选中的选项current_option = select.first_selected_optionprint(f"当前选择: {current_option.text}")

# 获取所有cookiescookies = driver.get_cookies()print(cookies)# 获取特定cookiecookie = driver.get_cookie("cookie_name")# 添加cookiedriver.add_cookie({"name": "test", "value": "123"})# 删除cookiedriver.delete_cookie("cookie_name")# 删除所有cookiesdriver.delete_all_cookies()10. 保存和加载sessionimport pickle# 登录后保存cookiesdriver.get("https://example.com/login")# ... 执行登录操作 ...# 保存cookies到文件cookies = driver.get_cookies()with open("cookies.pkl", "wb") as f:    pickle.dump(cookies, f)# 下次访问时加载cookiesdriver.get("https://example.com")with open("cookies.pkl", "rb") as f:    cookies = pickle.load(f)    for cookie in cookies:        driver.add_cookie(cookie)# 刷新页面,cookies生效driver.refresh()

selenium.common.exceptions.NoSuchElementException: Message: no such element

# 1. 添加显式等待element = WebDriverWait(driver, 10).until(    EC.presence_of_element_located((By.ID, "element_id")))# 2. 检查是否在iframe中driver.switch_to.frame("iframe_name")# 3. 使用浏览器开发者工具验证定位器# 按F12打开开发者工具,用Elements面板检查元素# 4. 打印页面源代码检查print(driver.page_source)

selenium.common.exceptions.ElementClickInterceptedException: Message: element click intercepted

# 方法1: 滚动到元素位置element = driver.find_element(By.ID, "button")driver.execute_script("arguments[0].scrollIntoView(true);", element)time.sleep(0.5)  # 等待滚动完成element.click()# 方法2: 用JavaScript直接点击driver.execute_script("arguments[0].click();", element)# 方法3: 先关闭遮挡的弹窗close_btn = driver.find_element(By.CLASS_NAME, "popup-close")close_btn.click()time.sleep(0.5)element.click()

selenium.common.exceptions.TimeoutException: Message: timeout

# 方法1: 增加超时时间driver.set_page_load_timeout(30)  # 30秒超时# 方法2: 设置脚本执行超时driver.set_script_timeout(30)# 方法3: 使用stop加载try:    driver.get("https://example.com")except TimeoutException:    driver.execute_script("window.stop();")  # 停止加载    print("页面加载超时,已停止加载")

session not created: This version of ChromeDriver only supports Chrome version XX

pip install webdriver-managerfrom selenium import webdriverfrom webdriver_manager.chrome import ChromeDriverManagerfrom selenium.webdriver.chrome.service import Serviceservice = Service(ChromeDriverManager().install())driver = webdriver.Chrome(service=service)

from selenium.webdriver.chrome.options import Optionschrome_options = Options()# 1. 禁用自动化标志chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])chrome_options.add_experimental_option('useAutomationExtension', False)# 2. 修改webdriver属性driver = webdriver.Chrome(options=chrome_options)driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {    "source": """        Object.defineProperty(navigator, 'webdriver', {            get: () => undefined        })    """})# 3. 添加随机延时,模拟人类行为import randomtime.sleep(random.uniform(1, 3))# 4. 使用代理IPchrome_options.add_argument('--proxy-server=http://proxy_ip:port')

def search_with_cleanup():    driver = None    try:        driver = webdriver.Chrome()        # ... 执行操作 ...    except Exception as e:        print(f"错误: {e}")    finally:        if driver:            driver.quit()  # 确保关闭            # 或者使用上下文管理器from contextlib import contextmanager@contextmanagerdef get_driver():    driver = webdriver.Chrome()    try:        yield driver    finally:        driver.quit()# 使用with get_driver() as driver:    driver.get("https://www.baidu.com")    # ... 操作 ...# 自动关闭

def collect_job_info(keyword):    """采集招聘网站的职位信息"""    driver = webdriver.Chrome()    results = []        try:        driver.get("https://www.某招聘网站.com")                # 输入搜索关键词        search_box = driver.find_element(By.ID, "search")        search_box.send_keys(keyword)        search_box.send_keys(Keys.ENTER)                # 等待结果加载        WebDriverWait(driver, 10).until(            EC.presence_of_all_elements_located((By.CLASS_NAME, "job-card"))        )                # 提取职位信息        job_cards = driver.find_elements(By.CLASS_NAME, "job-card")        for card in job_cards:            job_info = {                "title": card.find_element(By.CLASS_NAME, "job-title").text,                "company": card.find_element(By.CLASS_NAME, "company-name").text,                "salary": card.find_element(By.CLASS_NAME, "salary").text,                "location": card.find_element(By.CLASS_NAME, "location").text            }            results.append(job_info)                finally:        driver.quit()        return results

def auto_fill_form(data):    """自动填写表单并提交"""    driver = webdriver.Chrome()        try:        driver.get("https://example.com/form")                # 填写姓名        driver.find_element(By.ID, "name").send_keys(data["name"])                # 选择性别        if data["gender"] == "男":            driver.find_element(By.ID, "male").click()        else:            driver.find_element(By.ID, "female").click()                # 选择城市        city_select = Select(driver.find_element(By.ID, "city"))        city_select.select_by_visible_text(data["city"])                # 填写邮箱        driver.find_element(By.ID, "email").send_keys(data["email"])                # 勾选协议        driver.find_element(By.ID, "agree").click()                # 提交表单        driver.find_element(By.ID, "submit").click()                # 等待提交成功提示        success_msg = WebDriverWait(driver, 10).until(            EC.presence_of_element_located((By.CLASS_NAME, "success"))        )        print(f"提交成功: {success_msg.text}")            finally:        driver.quit()

import scheduledef check_price():    """监控商品价格"""    driver = webdriver.Chrome()        try:        driver.get("https://example.com/product/12345")                # 获取当前价格        price_element = WebDriverWait(driver, 10).until(            EC.presence_of_element_located((By.CLASS_NAME, "price"))        )        current_price = float(price_element.text.replace("¥", ""))                # 价格阈值        target_price = 999.0                if current_price <= target_price:            print(f"价格降到了 ¥{current_price}!")            # 可以在这里发送通知(邮件、微信等)            send_notification(f"商品降价了!现价 ¥{current_price}")        else:            print(f"当前价格: ¥{current_price}, 继续等待...")                finally:        driver.quit()# 每小时检查一次schedule.every(1).hours.do(check_price)while True:    schedule.run_pending()    time.sleep(60)

# 1. 禁用图片加载(加快速度)chrome_options = Options()prefs = {"profile.managed_default_content_settings.images": 2}chrome_options.add_experimental_option("prefs", prefs)# 2. 使用缓存chrome_options.add_argument("--disk-cache-size=1073741824")  # 1GB# 3. 复用浏览器实例# 不要每次操作都创建新driver,尽量复用# 4. 并发处理(小心资源消耗)from concurrent.futures import ThreadPoolExecutorwith ThreadPoolExecutor(max_workers=3) as executor:    futures = [executor.submit(search_in_baidu, keyword) for keyword in keywords]    results = [f.result() for f in futures]

from langgraph.prebuilt import create_react_agentfrom langgraph.checkpoint.memory import MemorySaverfrom langchain_core.runnables.config import RunnableConfigfrom langchain_openai import ChatOpenAI# 第一步：初始化模型# 这里用OpenAI，你也可以换成通义千问、文心一言等llm = ChatOpenAI(    model="gpt-4o",    temperature=0,    api_key="your-api-key"  # 替换成你的API Key)# 第二步：创建内存记忆对象memory = MemorySaver()# 第三步：创建Agent，传入checkpointer参数agent = create_react_agent(    model=llm,    tools=[],  # 暂时不加工具，专注测试记忆    checkpointer=memory,  # 关键：传入记忆对象    debug=True,  # 开启调试，看详细执行过程)# 第四步：配置thread_id# 同一个thread_id代表同一个对话线程config = RunnableConfig(configurable={"thread_id": "user_123"})# 第五步：第一轮对话print("=" * 60)print("第一轮对话：用户介绍自己")print("=" * 60)res1 = agent.invoke(    input={"messages": [("user", "我是Sam，今年25岁，喜欢打篮球")]},    config=config)print("\nAgent回复：")print(res1['messages'][-1].content)# 第六步：第二轮对话 - 测试记忆print("\n" + "=" * 60)print("第二轮对话：测试Agent是否记住用户信息")print("=" * 60)res2 = agent.invoke(    input={"messages": [("user", "我叫什么名字？今年多大？")]},    config=config)print("\nAgent回复：")print(res2['messages'][-1].content)# 第七步：第三轮对话 - 测试对话历史print("\n" + "=" * 60)print("第三轮对话：询问之前聊天内容")print("=" * 60)res3 = agent.invoke(    input={"messages": [("user", "我们刚才聊了什么？")]},    config=config)print("\nAgent回复：")print(res3['messages'][-1].content)# 第八步：查看完整的消息历史print("\n" + "=" * 60)print("完整消息历史：")print("=" * 60)for i, msg in enumerate(res3['messages']):    role = "用户" if msg.type == "human" else "Agent"    print(f"{i+1}. {role}: {msg.content}")

============================================================第一轮对话：用户介绍自己============================================================[-1:checkpoint] State at the end of step -1:{'messages': []}[0:tasks] Starting 1 task for step 0:- __start__ -> {'messages': [('user', '我是Sam，今年25岁，喜欢打篮球')]}[0:writes] Finished step 0 with writes to 1 channel:- messages -> [HumanMessage(content='我是Sam，今年25岁，喜欢打篮球')][0:checkpoint] State at the end of step 0:{'messages': [HumanMessage(content='我是Sam，今年25岁，喜欢打篮球')]}[1:tasks] Starting 1 task for step 1:- agent -> {'messages': [...], 'remaining_steps': 24}[1:writes] Finished step 1 with writes to 1 channel:- messages -> [AIMessage(content='你好Sam！很高兴认识你...')]Agent回复：你好Sam！很高兴认识你。25岁正是活力满满的年纪，打篮球是个很好的运动爱好！============================================================第二轮对话：测试Agent是否记住用户信息============================================================[2:checkpoint] State at the end of step 2:{'messages': [    HumanMessage(content='我是Sam，今年25岁，喜欢打篮球'),    AIMessage(content='你好Sam！很高兴认识你...'),]}Agent回复：你叫Sam，今年25岁。

1. 读取Checkpoint（如果存在）   └─ 获取thread_id="user_123"的历史消息2. 将新消息添加到历史中   └─ messages = [历史消息...] + [新消息]3. 调用LLM生成回复   └─ LLM能看到完整的对话历史4. 保存新的Checkpoint   └─ messages = [历史消息...] + [新消息] + [AI回复]5. 返回结果

# test_memory_persistence.pyfrom langgraph.prebuilt import create_react_agentfrom langgraph.checkpoint.memory import MemorySaverfrom langchain_openai import ChatOpenAIfrom langchain_core.runnables.config import RunnableConfigimport timellm = ChatOpenAI(model="gpt-4o", api_key="your-api-key")memory = MemorySaver()agent = create_react_agent(model=llm, tools=[], checkpointer=memory)config = RunnableConfig(configurable={"thread_id": "test_persistence"})# 第一次运行print("第一次运行：存储信息")agent.invoke(input={"messages": [("user", "记住：我的密码是123456")]}, config=config)print("\n等待5秒...")time.sleep(5)# 第二次运行：在同一个程序进程中print("\n第二次查询（同一进程）：")res = agent.invoke(input={"messages": [("user", "我的密码是什么？")]}, config=config)print(res['messages'][-1].content)  # 输出：你的密码是123456print("\n" + "="*60)print("现在请手动停止程序，再重新运行...")print("你会发现记忆丢失了")print("="*60)

mkdir langgraph-redis-democd langgraph-redis-demomkdir redis

version: '3.8'services:  redis:    image: redis:7.2-alpine  # 使用最新稳定版    container_name: langgraph-redis    ports:      - "6379:6379"  # 映射到本地6379端口    volumes:      - redis_data:/data  # 数据持久化      - ./redis/redis.conf:/usr/local/etc/redis/redis.conf  # 配置文件    command: redis-server /usr/local/etc/redis/redis.conf    restart: always  # 自动重启    networks:      - langgraph-networkvolumes:  redis_data:    driver: localnetworks:  langgraph-network:    driver: bridge

# 绑定所有网络接口（生产环境建议改成具体IP）bind 0.0.0.0# 保护模式关闭（生产环境建议开启并设置密码）protected-mode no# 端口port 6379# 持久化配置save 900 1      # 900秒内至少1个key变化，触发保存save 300 10     # 300秒内至少10个key变化save 60 10000   # 60秒内至少10000个key变化# AOF持久化（更安全）appendonly yesappendfilename "appendonly.aof"# 日志级别loglevel notice# 最大内存（可选）# maxmemory 256mb# maxmemory-policy allkeys-lru

docker-compose up -d# 查看日志docker-compose logs -f redis# 看到类似输出说明启动成功：# Ready to accept connections步骤5：验证Redis连接# 进入Redis容器docker exec -it langgraph-redis redis-cli# 测试命令127.0.0.1:6379> pingPONG127.0.0.1:6379> set test "hello"OK127.0.0.1:6379> get test"hello"127.0.0.1:6379> exit

# macOSbrew install redis# 启动Redisbrew services start redis# 验证redis-cli ping# 应该返回：PONGUbuntu/Debian:sudo apt updatesudo apt install redis-serve

# 启动sudo systemctl start redis-server# 设置开机自启sudo systemctl enable redis-server

pip install langgraph-checkpoint-redispip install langchain-openaipip install langgraph# 或者使用uv（更快）uv pip install langgraph-checkpoint-redis langchain-openai langgraph

from langgraph.prebuilt import create_react_agentfrom langgraph.checkpoint.redis import RedisSaverfrom langchain_core.runnables.config import RunnableConfigfrom langchain_openai import ChatOpenAIimport time# 配置REDIS_URL = "redis://localhost:6379"  # Redis连接地址# 如果设置了密码：redis://:password@localhost:6379def demo_basic_usage():    """基础使用示例"""    print("=" * 60)    print("示例1：基础使用")    print("=" * 60)        # 初始化模型    llm = ChatOpenAI(        model="gpt-4o",        temperature=0,        api_key="your-api-key"    )        # 使用with语句确保正确关闭连接    with RedisSaver.from_conn_string(REDIS_URL) as memory:        # 初始化Redis表结构（首次运行需要）        memory.setup()                # 创建Agent        agent = create_react_agent(            model=llm,            tools=[],            checkpointer=memory,            debug=False,  # 关闭调试输出，更清晰        )                # 配置对话线程        config = RunnableConfig(configurable={"thread_id": "redis_user_001"})                # 第一轮对话        print("\n第一轮：存储用户信息")        res1 = agent.invoke(            input={"messages": [("user", "我是张三，在北京工作，是一名Python工程师")]},            config=config        )        print(f"Agent: {res1['messages'][-1].content}")                # 第二轮对话        print("\n第二轮：测试记忆")        res2 = agent.invoke(            input={"messages": [("user", "我在哪里工作？做什么的？")]},            config=config        )        print(f"Agent: {res2['messages'][-1].content}")def demo_persistence():    """测试持久化：重启程序后记忆仍然存在"""    print("\n" + "=" * 60)    print("示例2：测试持久化")    print("=" * 60)        llm = ChatOpenAI(model="gpt-4o", api_key="your-api-key")        with RedisSaver.from_conn_string(REDIS_URL) as memory:        memory.setup()        agent = create_react_agent(model=llm, tools=[], checkpointer=memory)                # 使用不同的thread_id        config = RunnableConfig(configurable={"thread_id": "persistence_test"})                # 检查是否是首次运行        print("\n请选择操作：")        print("1. 存储新信息")        print("2. 读取之前存储的信息")        choice = input("输入选择（1或2）：")                if choice == "1":            agent.invoke(                input={"messages": [("user", "记住：项目代号是 Phoenix-2024")]},                config=config            )            print("✓ 信息已存储到Redis")            print("现在你可以重启程序，选择选项2来验证持久化")                    elif choice == "2":            res = agent.invoke(                input={"messages": [("user", "项目代号是什么？")]},                config=config            )            print(f"\nAgent: {res['messages'][-1].content}")def demo_multiple_threads():    """演示多用户隔离"""    print("\n" + "=" * 60)    print("示例3：多用户对话隔离")    print("=" * 60)        llm = ChatOpenAI(model="gpt-4o", api_key="your-api-key")        with RedisSaver.from_conn_string(REDIS_URL) as memory:        memory.setup()        agent = create_react_agent(model=llm, tools=[], checkpointer=memory)                # 用户A的对话        config_a = RunnableConfig(configurable={"thread_id": "user_A"})        print("\n用户A：")        agent.invoke(            input={"messages": [("user", "我喜欢吃苹果")]},            config=config_a        )                # 用户B的对话        config_b = RunnableConfig(configurable={"thread_id": "user_B"})        print("用户B：")        agent.invoke(            input={"messages": [("user", "我喜欢吃香蕉")]},            config=config_b        )                # 验证隔离性        print("\n验证隔离性：")        print("问用户A：")        res_a = agent.invoke(            input={"messages": [("user", "我喜欢吃什么水果？")]},            config=config_a        )        print(f"Agent: {res_a['messages'][-1].content}")                print("\n问用户B：")        res_b = agent.invoke(            input={"messages": [("user", "我喜欢吃什么水果？")]},            config=config_b        )        print(f"Agent: {res_b['messages'][-1].content}")def demo_redis_inspection():    """查看Redis中的数据结构"""    print("\n" + "=" * 60)    print("示例4：查看Redis存储的数据")    print("=" * 60)        import redis        # 直接连接Redis    r = redis.from_url(REDIS_URL)        # 查看所有key    keys = r.keys("*")    print(f"\nRedis中的所有key（共{len(keys)}个）：")    for key in keys[:10]:  # 只显示前10个        key_str = key.decode('utf-8')        key_type = r.type(key).decode('utf-8')        print(f"  - {key_str} (类型: {key_type})")        # 查看特定thread的数据    thread_keys = r.keys("*redis_user_001*")    if thread_keys:        print(f"\nthread_id='redis_user_001'相关的key：")        for key in thread_keys:            print(f"  - {key.decode('utf-8')}")if __name__ == "__main__":    # 运行所有示例    demo_basic_usage()        # 等待一下让用户看清楚    time.sleep(2)        demo_persistence()        time.sleep(2)        demo_multiple_threads()        time.sleep(2)        demo_redis_inspection()

# 第一次运行python redis_saver_demo.py# 停止程序（Ctrl+C）# 再次运行，会发现记忆还在python redis_saver_demo.py

# 进入Redis CLIdocker exec -it langgraph-redis redis-cli# 或本地安装的Redisredis-cli# 查看所有key127.0.0.1:6379> KEYS *1) "checkpoint:redis_user_001:..."2) "checkpoint:user_A:..."3) "checkpoint:user_B:..."# 查看特定key的类型127.0.0.1:6379> TYPE "checkpoint:redis_user_001:..."hash# 查看hash的所有字段127.0.0.1:6379> HGETALL "checkpoint:redis_user_001:..."

- checkpoint: 序列化的checkpoint数据（包含messages等）- metadata: 元数据- parent_checkpoint_id: 父checkpoint ID（用于追溯历史）

# production_redis_config.pyfrom langgraph.checkpoint.redis import RedisSaver# 生产环境配置REDIS_CONFIG = {    "url": "redis://:your_password@redis-server:6379/0",    "decode_responses": False,    "socket_connect_timeout": 5,    "socket_timeout": 5,    "retry_on_timeout": True,    "max_connections": 50,  # 连接池大小}# 创建连接memory = RedisSaver.from_conn_string(REDIS_CONFIG["url"])# 使用连接池（性能更好）import redispool = redis.ConnectionPool(**REDIS_CONFIG)r = redis.Redis(connection_pool=pool)

# 检查Redis是否启动import redistry:    r = redis.from_url("redis://localhost:6379")    r.ping()    print("✓ Redis连接成功")except redis.ConnectionError as e:    print(f"✗ Redis连接失败: {e}")    print("请检查：")    print("1. Redis是否已启动")    print("2. 端口6379是否被占用")    print("3. 防火墙是否阻止连接")

# 查看Redis内存使用redis-cli INFO memory# 设置最大内存和淘汰策略redis-cli CONFIG SET maxmemory 256mbredis-cli CONFIG SET maxmemory-policy allkeys-lru

def clean_old_checkpoints(days=7):    """清理N天前的checkpoint"""    import redis    from datetime import datetime, timedelta        r = redis.from_url("redis://localhost:6379")    cutoff = datetime.now() - timedelta(days=days)        keys = r.keys("checkpoint:*")    cleaned = 0        for key in keys:        # 获取key的创建时间（需要启用Redis keyspace notifications）        # 这里简化处理：直接删除旧key        # 实际生产环境建议根据metadata判断        r.delete(key)        cleaned += 1        print(f"清理了 {cleaned} 个旧checkpoint")

mkdir langgraph-mongodb-democd langgraph-mongodb-demomkdir mongodb

version: '3.8'services:  mongodb:    image: mongo:7.0    container_name: langgraph-mongodb    restart: always    ports:      - "27017:27017"    environment:      # 设置管理员账号      MONGO_INITDB_ROOT_USERNAME: admin      MONGO_INITDB_ROOT_PASSWORD: admin123      # 设置默认数据库      MONGO_INITDB_DATABASE: langgraph    volumes:      # 数据持久化      - mongodb_data:/data/db      # 配置文件      - mongodb_config:/data/configdb      # 初始化脚本      - ./mongodb/init-mongo.js:/docker-entrypoint-initdb.d/init-mongo.js    networks:      - langgraph-networkvolumes:  mongodb_data:    driver: local  mongodb_config:    driver: localnetworks:  langgraph-network:    driver: bridge

// 切换到langgraph数据库db = db.getSiblingDB('langgraph');// 创建应用用户db.createUser({  user: 'langgraph_user',  pwd: 'langgraph_pass',  roles: [    {      role: 'readWrite',      db: 'langgraph'    }  ]});// 创建checkpoints集合db.createCollection('checkpoints');// 创建索引（提升查询性能）db.checkpoints.createIndex({ "thread_id": 1 });db.checkpoints.createIndex({ "checkpoint_id": 1 });db.checkpoints.createIndex({ "created_at": -1 });print('MongoDB初始化完成！');

docker-compose up -d# 查看日志docker-compose logs -f mongodb# 看到 "Waiting for connections" 说明启动成功

# 方式1：使用mongosh（MongoDB Shell）docker exec -it langgraph-mongodb mongosh# 进入shell后test> use adminadmin> db.auth('admin', 'admin123'){ ok: 1 }admin> use langgraphlanggraph> show collectionscheckpointslanggraph> exit# 方式2：使用mongo命令（旧版）docker exec -it langgraph-mongodb mongo \  -u admin \  -p admin123 \  --authenticationDatabase admin> use langgraph> show collections> exit

# macOSbrew install mongosh# 连接mongosh "mongodb://admin:admin123@localhost:27017"Ubuntu:wget -qO - https://www.mongodb.org/static/pgp/server-7.0.asc | sudo apt-key add -echo "deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu focal/mongodb-org/7.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-7.0.listsudo apt updatesudo apt install -y mongodb-mongosh

pip install langgraph-checkpoint-mongodbpip install pymongopip install langchain-openaipip install langgraph# 或使用uvuv pip install langgraph-checkpoint-mongodb pymongo langchain-openai langgraph

from langgraph.prebuilt import create_react_agentfrom langgraph.checkpoint.mongodb import MongoDBSaverfrom langchain_core.runnables.config import RunnableConfigfrom langchain_openai import ChatOpenAIfrom pymongo import MongoClientfrom datetime import datetimeimport json# MongoDB连接配置MONGODB_URI = "mongodb://langgraph_user:langgraph_pass@localhost:27017"DB_NAME = "langgraph"def demo_basic_usage():    """基础使用示例"""    print("=" * 60)    print("示例1：基础使用")    print("=" * 60)        llm = ChatOpenAI(        model="gpt-4o", temperature=0, api_key="your-api-key" )# 使用with语句管理连接with MongoDBSaver.from_conn_string(MONGODB_URI, DB_NAME) as memory:    # 创建Agent    agent = create_react_agent(        model=llm,        tools=[],        checkpointer=memory,        debug=False,    )        # 配置对话线程    config = RunnableConfig(configurable={"thread_id": "mongo_user_001"})        # 第一轮对话    print("\n第一轮：存储用户信息")    res1 = agent.invoke(        input={"messages": [("user", "我是李四，在上海工作，是产品经理，负责AI产品")]},        config=config    )    print(f"Agent: {res1['messages'][-1].content}")        # 第二轮对话    print("\n第二轮：测试记忆")    res2 = agent.invoke(        input={"messages": [("user", "我的职位是什么？在哪个城市？")]},        config=config    )    print(f"Agent: {res2['messages'][-1].content}")        # 第三轮对话    print("\n第三轮：深度提问")    res3 = agent.invoke(        input={"messages": [("user", "根据我的工作，给我推荐3本相关书籍")]},        config=config    )    print(f"Agent: {res3['messages'][-1].content}")def demo_query_history(): """演示如何查询对话历史""" print("\n" + "=" * 60) print("示例2：查询对话历史") print("=" * 60)# 直接使用pymongo查询client = MongoClient(MONGODB_URI)db = client[DB_NAME]collection = db['checkpoints']# 查询所有thread_idthread_ids = collection.distinct('thread_id')print(f"\n当前有 {len(thread_ids)} 个对话线程：")for thread_id in thread_ids[:5]:  # 只显示前5个    print(f"  - {thread_id}")# 查询特定用户的所有checkpointprint("\n查询 mongo_user_001 的对话记录：")checkpoints = collection.find(    {"thread_id": "mongo_user_001"}).sort("created_at", -1)  # 按时间倒序for i, cp in enumerate(checkpoints):    print(f"\nCheckpoint {i+1}:")    print(f"  ID: {cp.get('checkpoint_id', 'N/A')}")    print(f"  时间: {cp.get('created_at', 'N/A')}")    print(f"  消息数量: {len(cp.get('checkpoint', {}).get('channel_values', {}).get('messages', []))}")client.close()def demo_analytics(): """演示数据分析""" print("\n" + "=" * 60) print("示例3：对话数据分析") print("=" * 60)client = MongoClient(MONGODB_URI)db = client[DB_NAME]collection = db['checkpoints']# 统计1：总对话轮数total_checkpoints = collection.count_documents({})print(f"\n总checkpoint数量: {total_checkpoints}")# 统计2：活跃用户数active_users = len(collection.distinct('thread_id'))print(f"活跃用户数: {active_users}")# 统计3：每个用户的对话轮数pipeline = [    {        "$group": {            "_id": "$thread_id",            "count": {"$sum": 1},            "last_active": {"$max": "$created_at"}        }    },    {"$sort": {"count": -1}},    {"$limit": 5}]print("\n对话最多的5个用户：")for result in collection.aggregate(pipeline):    print(f"  {result['_id']}: {result['count']}轮对话, "          f"最后活跃: {result.get('last_active', 'N/A')}")# 统计4：今天的对话量from datetime import datetime, timedeltatoday_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)today_count = collection.count_documents({    "created_at": {"$gte": today_start}})print(f"\n今天的对话轮数: {today_count}")client.close()def demo_search_content(): """演示内容搜索""" print("\n" + "=" * 60) print("示例4：搜索对话内容") print("=" * 60)client = MongoClient(MONGODB_URI)db = client[DB_NAME]collection = db['checkpoints']# 先创建文本索引（首次运行需要）try:    collection.create_index([("checkpoint.channel_values.messages.content", "text")])    print("✓ 文本索引创建成功")except Exception as e:    print(f"索引已存在或创建失败: {e}")# 搜索包含特定关键词的对话keyword = "产品"print(f"\n搜索包含 '{keyword}' 的对话：")results = collection.find(    {"$text": {"$search": keyword}}).limit(3)for result in results:    print(f"\nThread ID: {result.get('thread_id')}")    messages = result.get('checkpoint', {}).get('channel_values', {}).get('messages', [])    print(f"消息数量: {len(messages)}")    if messages:        # 显示最后一条消息        last_msg = messages[-1]        print(f"最后消息: {last_msg.get('content', 'N/A')[:100]}...")client.close()def demo_export_conversations(): """导出对话记录""" print("\n" + "=" * 60) print("示例5：导出对话记录") print("=" * 60)client = MongoClient(MONGODB_URI)db = client[DB_NAME]collection = db['checkpoints']# 导出特定用户的完整对话thread_id = "mongo_user_001"checkpoints = list(collection.find(    {"thread_id": thread_id}).sort("created_at", 1))  # 按时间正序if not checkpoints:    print(f"未找到 {thread_id} 的对话记录")    client.close()    return# 提取所有消息all_messages = []for cp in checkpoints:    messages = cp.get('checkpoint', {}).get('channel_values', {}).get('messages', [])    all_messages.extend(messages)# 导出为JSON文件export_data = {    "thread_id": thread_id,    "export_time": datetime.now().isoformat(),    "total_messages": len(all_messages),    "conversations": []}for i, msg in enumerate(all_messages):    export_data["conversations"].append({        "index": i + 1,        "type": msg.get('type', 'unknown'),        "content": msg.get('content', ''),        "timestamp": msg.get('timestamp', '')    })# 保存到文件filename = f"conversation_{thread_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"with open(filename, 'w', encoding='utf-8') as f:    json.dump(export_data, f, ensure_ascii=False, indent=2)print(f"✓ 对话已导出到: {filename}")print(f"  共 {len(all_messages)} 条消息")client.close()def demo_cleanup(): """清理旧数据""" print("\n" + "=" * 60) print("示例6：数据清理") print("=" * 60)client = MongoClient(MONGODB_URI)db = client[DB_NAME]collection = db['checkpoints']# 删除7天前的数据from datetime import timedeltacutoff_date = datetime.now() - timedelta(days=7)result = collection.delete_many({    "created_at": {"$lt": cutoff_date}})print(f"✓ 清理了 {result.deleted_count} 条7天前的记录")# 删除特定用户的数据# thread_id_to_delete = "test_user"# result = collection.delete_many({"thread_id": thread_id_to_delete})# print(f"✓ 删除用户 {thread_id_to_delete} 的 {result.deleted_count} 条记录")client.close()def demo_multi_user_conversation(): """演示多用户场景""" print("\n" + "=" * 60) print("示例7：多用户对话管理") print("=" * 60)llm = ChatOpenAI(model="gpt-4o", api_key="your-api-key")with MongoDBSaver.from_conn_string(MONGODB_URI, DB_NAME) as memory:    agent = create_react_agent(model=llm, tools=[], checkpointer=memory)        # 模拟3个不同用户    users = [        {"id": "user_001", "intro": "我是工程师，喜欢Python"},        {"id": "user_002", "intro": "我是设计师，喜欢Figma"},        {"id": "user_003", "intro": "我是产品经理，喜欢研究用户需求"}    ]        # 每个用户进行对话    for user in users:        config = RunnableConfig(configurable={"thread_id": user["id"]})        print(f"\n{user['id']} 的对话：")                # 自我介绍        agent.invoke(            input={"messages": [("user", user["intro"])]},            config=config        )                # 询问建议        res = agent.invoke(            input={"messages": [("user", "根据我的职业，推荐一个学习方向")]},            config=config        )        print(f"Agent建议: {res['messages'][-1].content}")# 统计每个用户的对话print("\n" + "=" * 60)client = MongoClient(MONGODB_URI)db = client[DB_NAME]collection = db['checkpoints']for user in users:    count = collection.count_documents({"thread_id": user["id"]})    print(f"{user['id']}: {count} 条记录")client.close()if name == "main": import sysprint("LangGraph MongoDB 持久化完整示例")print("=" * 60)print("\n请选择要运行的示例：")print("1. 基础使用")print("2. 查询历史")print("3. 数据分析")print("4. 内容搜索")print("5. 导出对话")print("6. 数据清理")print("7. 多用户管理")print("0. 运行所有示例")choice = input("\n输入选择（0-7）：").strip()if choice == "1":    demo_basic_usage()elif choice == "2":    demo_query_history()elif choice == "3":    demo_analytics()elif choice == "4":    demo_search_content()elif choice == "5":    demo_export_conversations()elif choice == "6":    demo_cleanup()elif choice == "7":    demo_multi_user_conversation()elif choice == "0":    demo_basic_usage()    demo_query_history()    demo_analytics()    demo_search_content()    demo_export_conversations()    demo_multi_user_conversation()else:    print("无效的选择")

// 查找特定用户db.checkpoints.find({"thread_id": "mongo_user_001"})// 统计每个用户的对话数db.checkpoints.aggregate([  {$group: {_id: "$thread_id", count: {$sum: 1}}},  {$sort: {count: -1}}])// 查找今天的对话db.checkpoints.find({  "created_at": {    $gte: ISODate("2024-01-01T00:00:00Z")  }})

def create_indexes():    """创建性能优化索引"""    client = MongoClient(MONGODB_URI)    db = client[DB_NAME]    collection = db['checkpoints']        # 复合索引：thread_id + created_at    collection.create_index([        ("thread_id", 1),        ("created_at", -1)    ])        # 唯一索引：checkpoint_id    collection.create_index(        "checkpoint_id",        unique=True    )        # TTL索引：自动删除30天前的数据    collection.create_index(        "created_at",        expireAfterSeconds=30*24*60*60  # 30天    )        print("✓ 索引创建完成")        # 查看所有索引    indexes = collection.index_information()    print("\n当前索引：")    for name, info in indexes.items():        print(f"  - {name}: {info['key']}")        client.close()

def advanced_analytics():    """高级数据分析"""    client = MongoClient(MONGODB_URI)    db = client[DB_NAME]    collection = db['checkpoints']        # 分析1：每小时的对话量    pipeline = [        {            "$group": {                "_id": {                    "hour": {"$hour": "$created_at"},                    "date": {"$dateToString": {"format": "%Y-%m-%d", "date": "$created_at"}}                },                "count": {"$sum": 1}            }        },        {"$sort": {"_id.date": 1, "_id.hour": 1}}    ]        print("每小时对话量分布：")    for result in collection.aggregate(pipeline):        print(f"  {result['_id']['date']} {result['_id']['hour']}:00 - {result['count']}次")        # 分析2：平均对话轮数    pipeline = [        {            "$group": {                "_id": "$thread_id",                "total_turns": {"$sum": 1}            }        },        {            "$group": {                "_id": None,                "avg_turns": {"$avg": "$total_turns"},                "max_turns": {"$max": "$total_turns"},                "min_turns": {"$min": "$total_turns"}            }        }    ]        result = list(collection.aggregate(pipeline))    if result:        stats = result[0]        print(f"\n对话轮数统计：")        print(f"  平均: {stats['avg_turns']:.2f}轮")        print(f"  最多: {stats['max_turns']}轮")        print(f"  最少: {stats['min_turns']}轮")        client.close()

# 备份整个数据库docker exec langgraph-mongodb mongodump \  --username=admin \  --password=admin123 \  --authenticationDatabase=admin \  --db=langgraph \  --out=/tmp/backup# 导出备份文件到本地docker cp langgraph-mongodb:/tmp/backup ./mongodb_backup# 恢复数据库docker exec langgraph-mongodb mongorestore \  --username=admin \  --password=admin123 \  --authenticationDatabase=admin \  --db=langgraph \  /tmp/backup/langgraph

# production_mongodb_config.pyfrom pymongo import MongoClient, monitoringimport logging# 配置连接池MONGODB_CONFIG = {    "host": "mongodb://langgraph_user:langgraph_pass@mongo1:27017,mongo2:27017,mongo3:27017",    "replicaSet": "rs0",  # 副本集    "maxPoolSize": 50,    "minPoolSize": 10,    "maxIdleTimeMS": 30000,    "serverSelectionTimeoutMS": 5000,    "connectTimeoutMS": 5000,    "socketTimeoutMS": 5000,    "retryWrites": True,    "w": "majority",  # 写确认级别}# 启用监控class CommandLogger(monitoring.CommandListener):    def started(self, event):        logging.info(f"Command {event.command_name} started on {event.connection_id}")        def succeeded(self, event):        logging.info(f"Command {event.command_name} succeeded in {event.duration_micros}μs")        def failed(self, event):        logging.error(f"Command {event.command_name} failed: {event.failure}")monitoring.register(CommandLogger())# 创建客户端client = MongoClient(**MONGODB_CONFIG)

BaseCheckpointSaver (抽象基类)├── get_tuple()      # 读取checkpoint├── put()            # 保存checkpoint├── put_writes()     # 保存增量写入（可选）└── get_next_version() # 生成版本号

checkpoints/├── user_001_checkpoint_1.json├── user_001_checkpoint_2.json├── user_002_checkpoint_1.json└── user_002_checkpoint_2.json

{  "checkpoint": "base64编码的序列化数据",  "metadata": "base64编码的元数据"}

from langgraph.checkpoint.base import (    BaseCheckpointSaver,    Checkpoint,    CheckpointMetadata,    CheckpointTuple,    ChannelVersions,)from langchain_core.runnables.config import RunnableConfigfrom typing import Optional, Any, Sequencefrom pathlib import Pathimport jsonimport pickleimport base64import randomclass FileSaver(BaseCheckpointSaver[str]):    """    基于文件系统的Checkpoint持久化实现        功能：    1. 将checkpoint保存为JSON文件    2. 支持多线程（通过thread_id隔离）    3. 自动序列化/反序列化复杂对象    4. 提供版本管理    """        def __init__(        self,        *,        base_path: str = "./checkpoints",        auto_cleanup: bool = False,        max_checkpoints_per_thread: int = 10,    ) -> None:        """        初始化FileSaver                参数：            base_path: checkpoint文件存储目录            auto_cleanup: 是否自动清理旧checkpoint            max_checkpoints_per_thread: 每个thread保留的最大checkpoint数        """        super().__init__()        self.base_path = Path(base_path)        self.auto_cleanup = auto_cleanup        self.max_checkpoints_per_thread = max_checkpoints_per_thread                # 确保目录存在        self.base_path.mkdir(parents=True, exist_ok=True)        print(f"✓ FileSaver初始化完成，存储路径: {self.base_path.absolute()}")        def _get_checkpoint_path(self, thread_id: str, checkpoint_id: str) -> Path:        """        生成checkpoint文件路径                参数：            thread_id: 线程ID            checkpoint_id: checkpoint ID                    返回：            Path对象        """        # 文件名格式：{thread_id}_{checkpoint_id}.json        filename = f"{thread_id}_{checkpoint_id}.json"        return self.base_path / filename        def _serialize_data(self, obj: Any) -> str:        """        序列化Python对象为JSON可存储的字符串                使用pickle序列化 → base64编码        这样可以保存任何Python对象                参数：            obj: 要序列化的对象                    返回：            base64编码的字符串        """        try:            # 使用pickle序列化            pickled = pickle.dumps(obj)            # base64编码，方便存储为JSON            encoded = base64.b64encode(pickled).decode('ascii')            return encoded        except Exception as e:            print(f"✗ 序列化失败: {e}")            raise        def _deserialize_data(self, data: str) -> Any:        """        反序列化字符串为Python对象                base64解码 → pickle反序列化                参数：            data: base64编码的字符串                    返回：            原始Python对象        """        try:            # base64解码            decoded = base64.b64decode(data)            # pickle反序列化            obj = pickle.loads(decoded)            return obj        except Exception as e:            print(f"✗ 反序列化失败: {e}")            raise        def put(        self,        config: RunnableConfig,        checkpoint: Checkpoint,        metadata: CheckpointMetadata,        new_versions: ChannelVersions,    ) -> RunnableConfig:        """        保存checkpoint到文件                核心方法：每次Agent执行完一个步骤都会调用                参数：            config: 包含thread_id等配置            checkpoint: 当前状态快照（包含messages等）            metadata: 元数据（时间戳、步骤信息等）            new_versions: 新的channel版本号                    返回：            更新后的config        """        thread_id = config["configurable"]["thread_id"]        checkpoint_id = checkpoint["id"]                print(f"💾 保存checkpoint: thread={thread_id}, id={checkpoint_id}")                # 序列化数据        checkpoint_data = {            "checkpoint": self._serialize_data(checkpoint),            "metadata": self._serialize_data(metadata),            "thread_id": thread_id,            "checkpoint_id": checkpoint_id,        }                # 写入文件        checkpoint_path = self._get_checkpoint_path(thread_id, checkpoint_id)        try:            with open(checkpoint_path, 'w', encoding='utf-8') as f:                json.dump(checkpoint_data, f, ensure_ascii=False, indent=2)            print(f"✓ 文件已保存: {checkpoint_path.name}")        except Exception as e:            print(f"✗ 保存失败: {e}")            raise                # 自动清理旧checkpoint        if self.auto_cleanup:            self._cleanup_old_checkpoints(thread_id)                return {            "configurable": {                "thread_id": thread_id,                "checkpoint_id": checkpoint_id,            }        }        def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:        """        从文件恢复checkpoint                核心方法：每次Agent开始执行前都会调用，用于恢复历史状态                参数：            config: 包含thread_id和可选的checkpoint_id                    返回：            CheckpointTuple或None        """        thread_id = config["configurable"]["thread_id"]        checkpoint_id = config["configurable"].get("checkpoint_id", "")                # 如果没有指定checkpoint_id，找最新的        if not checkpoint_id:            checkpoint_files = list(self.base_path.glob(f"{thread_id}_*.json"))            if not checkpoint_files:                print(f"ℹ️  未找到thread={thread_id}的checkpoint")                return None                        # 按文件名排序，取最新的            checkpoint_files.sort(reverse=True)            latest_file = checkpoint_files[0]            checkpoint_id = latest_file.stem[len(thread_id)+1:]  # 提取checkpoint_id            print(f"📂 找到最新checkpoint: {latest_file.name}")                # 读取文件        checkpoint_path = self._get_checkpoint_path(thread_id, checkpoint_id)        if not checkpoint_path.exists():            print(f"✗ 文件不存在: {checkpoint_path}")            return None                try:            with open(checkpoint_path, 'r', encoding='utf-8') as f:                data = json.load(f)                        # 反序列化数据            checkpoint = self._deserialize_data(data["checkpoint"])            metadata = self._deserialize_data(data["metadata"])                        print(f"✓ 成功加载checkpoint: {len(checkpoint.get('channel_values', {}).get('messages', []))} 条消息")                        return CheckpointTuple(                config={                    "configurable": {                        "thread_id": thread_id,                        "checkpoint_id": checkpoint_id,                    }                },                checkpoint=checkpoint,                metadata=metadata,                pending_writes=[],  # 暂不支持pending writes                parent_config=None,  # 暂不支持父配置            )                except Exception as e:            print(f"✗ 加载checkpoint失败: {e}")            return None        def put_writes(        self,        config: RunnableConfig,        writes: Sequence[tuple[str, Any]],        task_id: str,        task_path: str = "",    ) -> None:        """        保存增量写入                这个方法必须实现，但我们可以选择不做任何事        （所有数据都在put方法中保存）                如果需要支持更细粒度的增量更新，可以在这里实现        """        pass  # 简化实现：不处理增量写入        def get_next_version(self, current: Optional[str], channel: Any) -> str:        """        生成下一个版本号                LangGraph用版本号来跟踪channel的变化                参数：            current: 当前版本号            channel: channel对象                    返回：            新版本号（字符串格式）        """        if current is None:            current_v = 0        elif isinstance(current, int):            current_v = currentconn_string( "redis://localhost:6379", socket_connect_timeout=5, socket_timeout=5, retry_on_timeout=True, ) memory.setup() logger.info("✓ Redis连接成功") break except Exception as e: logger.error(f"✗ Redis连接失败 (尝试 {attempt+1}/{max_retries}): {e}") if attempt == max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避agent = create_react_agent(    model=llm,    tools=[],    checkpointer=memory,    debug=False,)return agent, memorydef safe_agent_invoke(agent, input_data, config, max_retries=3): """安全的Agent调用，带错误处理"""for attempt in range(max_retries):    try:        result = agent.invoke(input=input_data, config=config)        return result, None        except TimeoutError as e:        logger.error(f"✗ 调用超时 (尝试 {attempt+1}/{max_retries}): {e}")        if attempt == max_retries - 1:            return None, "系统繁忙，请稍后重试"        time.sleep(1)        except ConnectionError as e:        logger.error(f"✗ 连接错误 (尝试 {attempt+1}/{max_retries}): {e}")        if attempt == max_retries - 1:            return None, "服务暂时不可用"        time.sleep(2)        except Exception as e:        logger.error(f"✗ 未知错误: {e}", exc_info=True)        return None, "系统错误，请联系管理员"return None, "操作失败"使用示例if name == "main": agent, memory = create_production_agent()config = {"configurable": {"thread_id": "prod_user_001"}}result, error = safe_agent_invoke(    agent,    {"messages": [("user", "你好")]},    config)if error:    print(f"错误: {error}")else:    print(f"成功: {result['messages'][-1].content}")memory.close()