LangChain和LangGraph

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

import openai
import re
import httpx
import os
from dotenv import load_dotenv

_ = load_dotenv()
from openai import OpenAI

# ================= 改动开始 =================
# 使用 DeepSeek 的 API 地址和你的 API Key
# 建议在 .env 文件中设置 DEEPSEEK_API_KEY
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")  # 如果没有设置，会读取 OPENAI_API_KEY 兼容
if not DEEPSEEK_API_KEY:
    raise ValueError("请在环境变量中设置 DEEPSEEK_API_KEY")

client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com/v1"      # DeepSeek 官方 API 地址
)
# ================= 改动结束 =================

# 测试调用（可选）
chat_completion = client.chat.completions.create(
    model="deepseek-chat",          # 使用 DeepSeek 模型
    messages=[{"role": "user", "content": "Hello world"}]
)
print(chat_completion.choices[0].message.content)


class Agent:
    def __init__(self, system=""):
        self.system = system
        self.messages = []
        if self.system:
            self.messages.append({"role": "system", "content": system})

    def __call__(self, message):
        self.messages.append({"role": "user", "content": message})
        result = self.execute()
        self.messages.append({"role": "assistant", "content": result})
        return result

    def execute(self):
        completion = client.chat.completions.create(
            model="deepseek-chat",          # 统一使用 DeepSeek 模型
            temperature=0,
            messages=self.messages
        )
        return completion.choices[0].message.content


prompt = """
You run in a loop of Thought, Action, PAUSE, Observation.
At the end of the loop you output an Answer
Use Thought to describe your thoughts about the question you have been asked.
Use Action to run one of the actions available to you - then return PAUSE.
Observation will be the result of running those actions.

Your available actions are:

calculate:
e.g. calculate: 4 * 7 / 3
Runs a calculation and returns the number - uses Python so be sure to use floating point syntax if necessary

average_dog_weight:
e.g. average_dog_weight: Collie
returns average weight of a dog when given the breed

Example session:

Question: How much does a Bulldog weigh?
Thought: I should look the dogs weight using average_dog_weight
Action: average_dog_weight: Bulldog
PAUSE

You will be called again with this:

Observation: A Bulldog weights 51 lbs

You then output:

Answer: A bulldog weights 51 lbs
""".strip()


def calculate(what):
    return eval(what)


def average_dog_weight(name):
    if name in "Scottish Terrier":
        return("Scottish Terriers average 20 lbs")
    elif name in "Border Collie":
        return("a Border Collies average weight is 37 lbs")
    elif name in "Toy Poodle":
        return("a toy poodles average weight is 7 lbs")
    else:
        return("An average dog weights 50 lbs")


known_actions = {
    "calculate": calculate,
    "average_dog_weight": average_dog_weight
}

abot = Agent(prompt)
result = abot("How much does a toy poodle weigh?")
print(result)
result = average_dog_weight("Toy Poodle")
result
next_prompt = "Observation: {}".format(result)
abot(next_prompt)
abot.messages

abot = Agent(prompt)
question = """I have 2 dogs, a border collie and a scottish terrier. \
What is their combined weight"""
abot(question)
next_prompt = "Observation: {}".format(average_dog_weight("Border Collie"))
print(next_prompt)
abot(next_prompt)
next_prompt = "Observation: {}".format(average_dog_weight("Scottish Terrier"))
print(next_prompt)
abot(next_prompt)
next_prompt = "Observation: {}".format(eval("37 + 20"))
print(next_prompt)
next_prompt = "Observation: {}".format(eval("37 + 20"))
print(next_prompt)
abot(next_prompt)


action_re = re.compile('^Action: (\w+): (.*)$')   # python regular expression to selection action


def query(question, max_turns=5):
    i = 0
    bot = Agent(prompt)
    next_prompt = question
    while i < max_turns:
        i += 1
        result = bot(next_prompt)
        print(result)
        actions = [
            action_re.match(a)
            for a in result.split('\n')
            if action_re.match(a)
        ]
        if actions:
            # There is an action to run
            action, action_input = actions[0].groups()
            if action not in known_actions:
                raise Exception("Unknown action: {}: {}".format(action, action_input))
            print(" -- running {} {}".format(action, action_input))
            observation = known_actions[action](action_input)
            print("Observation:", observation)
            next_prompt = "Observation: {}".format(observation)
        else:
            return


question = """I have 2 dogs, a border collie and a scottish terrier. \
What is their combined weight"""
query(question)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from dotenv import load_dotenv
_ = load_dotenv()
from openai import OpenAI
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
if not DEEPSEEK_API_KEY:
    raise ValueError("请在环境变量中设置 DEEPSEEK_API_KEY")
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com/v1"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

class Agent:
    def __init__(self, system=""):
        self.system = system
        self.messages = []
        if self.system:
            self.messages.append({"role": "system", "content": system})
    def __call__(self, message):
        self.messages.append({"role": "user", "content": message})
        result = self.execute()
        self.messages.append({"role": "assistant", "content": result})
        return result
    def execute(self):
        completion = client.chat.completions.create(
            model="deepseek-chat",
            temperature=0,
            messages=self.messages
        )
        return completion.choices[0].message.content

1
2
3
4

prompt = """
You run in a loop of Thought, Action, PAUSE, Observation...
...（详细内容）
""".strip()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def calculate(what):
    return eval(what)
def average_dog_weight(name):
    if name in "Scottish Terrier": 
        return("Scottish Terriers average 20 lbs")
    elif name in "Border Collie":
        return("a Border Collies average weight is 37 lbs")
    elif name in "Toy Poodle":
        return("a toy poodles average weight is 7 lbs")
    else:
        return("An average dog weights 50 lbs")
known_actions = {
    "calculate": calculate,
    "average_dog_weight": average_dog_weight
}

1
2
3
4
5
6

abot = Agent(prompt)
result = abot("How much does a toy poodle weigh?")
print(result)                     # 模型输出 Action 和 PAUSE
result = average_dog_weight("Toy Poodle")   # 手工执行工具
next_prompt = "Observation: {}".format(result)
abot(next_prompt)                 # 将 Observation 喂给 Agent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

action_re = re.compile('^Action: (\w+): (.*)$')
def query(question, max_turns=5):
    i = 0
    bot = Agent(prompt)
    next_prompt = question
    while i < max_turns:
        i += 1
        result = bot(next_prompt)
        print(result)
        actions = [
            action_re.match(a) 
            for a in result.split('\n') 
            if action_re.match(a)
        ]
        if actions:
            action, action_input = actions[0].groups()
            if action not in known_actions:
                raise Exception("Unknown action: {}: {}".format(action, action_input))
            print(" -- running {} {}".format(action, action_input))
            observation = known_actions[action](action_input)
            print("Observation:", observation)
            next_prompt = "Observation: {}".format(observation)
        else:
            return

1
2
3

question = """I have 2 dogs, a border collie and a scottish terrier. \
What is their combined weight"""
query(question)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

from dotenv import load_dotenv
_ = load_dotenv()
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
import os
 

# ================= DeepSeek 配置开始 =================
# 从环境变量获取 DeepSeek API Key（建议在 .env 中设置 DEEPSEEK_API_KEY）
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
if not DEEPSEEK_API_KEY:
    raise ValueError("请在环境变量中设置 DEEPSEEK_API_KEY")  
# 创建 ChatOpenAI 实例，指向 DeepSeek 的 API 地址
# 注意：模型名称使用 deepseek-chat 或 deepseek-coder
deepseek_model = ChatOpenAI(
    model="deepseek-chat",                # DeepSeek 对话模型
    openai_api_key=DEEPSEEK_API_KEY,      # 注意参数名是 openai_api_key（兼容 OpenAI）
    openai_api_base="https://api.deepseek.com/v1",  # DeepSeek 官方地址
    temperature=0                         # 可选：降低随机性
)

# ================= DeepSeek 配置结束 =================

# 搜索工具（与模型无关）
tool = TavilySearch(max_results=4)
print(type(tool))
print(tool.name)

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
class Agent:
    def __init__(self, model, tools, system=""):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges(
            "llm",
            self.exists_action,
            {True: "action", False: END}
        )
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        self.graph = graph.compile()
        self.tools = {t.name: t for t in tools}

        # 模型绑定工具（同样兼容 DeepSeek，因为 DeepSeek 支持 function calling）

        self.model = model.bind_tools(tools)
    def exists_action(self, state: AgentState):
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

    def take_action(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []

        for t in tool_calls:
            print(f"Calling: {t}")
            if not t['name'] in self.tools:
                print("\n ....bad tool name....")
                result = "bad tool name, retry"
            else:
                result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print("Back to the model!")
        return {'messages': results}

# 系统提示（可根据 DeepSeek 调整，但原 prompt 也适用）
prompt = """You are a smart research assistant. Use the search engine to look up information. \
You are allowed to make multiple calls (either together or in sequence). \
Only look up information when you are sure of what you want. \
If you need to look up some information before asking a follow up question, you are allowed to do that!
"""
  
# ================= 使用 DeepSeek 模型创建 Agent =================

# 对于简单任务（如天气查询），使用 deepseek-chat 即可
abot = Agent(deepseek_model, [tool], system=prompt)

# 可选：可视化图结构（需要 IPython，如果没有可注释）
# from IPython.display import Image
# Image(abot.graph.get_graph().draw_png())

# 测试1：单个城市天气
messages = [HumanMessage(content="What is the weather in sf?")]
result = abot.graph.invoke({"messages": messages})
print("Final answer:", result['messages'][-1].content)

# 测试2：两个城市天气
messages = [HumanMessage(content="What is the weather in SF and LA?")]
result = abot.graph.invoke({"messages": messages})
print("Final answer:", result['messages'][-1].content)

# 测试3：多步推理问题（需要更好性能，仍用 deepseek-chat，也可以用 deepseek-coder）
query = "Who won the super bowl in 2024? In what state is the winning team headquarters located? \
What is the GDP of that state? Answer each question."
messages = [HumanMessage(content=query)]
# 这里如果希望使用更强的模型，DeepSeek 目前主要提供 deepseek-chat，已经足够
abot = Agent(deepseek_model, [tool], system=prompt)
result = abot.graph.invoke({"messages": messages})
print("Multi-step answer:", result['messages'][-1].content)

1
2
3
4
5
6

deepseek_model = ChatOpenAI(
    model="deepseek-chat",
    openai_api_key=DEEPSEEK_API_KEY,
    openai_api_base="https://api.deepseek.com/v1",
    temperature=0
)

1
2
3
4
5
6

from concurrent.futures import ThreadPoolExecutor
def take_action(self, state):
    tool_calls = state['messages'][-1].tool_calls
    with ThreadPoolExecutor() as executor:
        results = list(executor.map(self._invoke_tool, tool_calls))
    return {'messages': results}

1

result = f"Error: Tool '{t['name']}' not found. Available tools: {list(self.tools.keys())}"

1
2

def __init__(self, ..., max_iterations=10):
    self.max_iterations = max_iterations

1
2

if not os.getenv("TAVILY_API_KEY"):
    raise ValueError("请设置 TAVILY_API_KEY 环境变量")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# 导入 dotenv 中的 load_dotenv 函数，用于加载 .env 文件中的环境变量
# pip install python-dotenv tavily-python 提前导入相关python依赖
# 搜索工具初始连接
from dotenv import load_dotenv


# 导入 os 模块，用于读取操作系统的环境变量

import os
  
# 从 tavily 库中导入 TavilyClient 类，用于调用 Tavily 搜索 API
from tavily import TavilyClient
  
# 调用 load_dotenv()，它会自动查找项目根目录下的 .env 文件，并将其中的键值对加载到环境变量中
load_dotenv()
# 从环境变量中获取 TAVILY_API_KEY 的值（即你注册 Tavily 后获得的真实 API Key）
# 这里使用 os.getenv()，它等同于 os.environ.get()，如果环境变量不存在则返回 None

api_key = os.getenv("TAVILY_API_KEY")

# 检查 API Key 是否成功读取，如果为空则主动抛出异常，让程序停止并提示错误信息

if not api_key:

    raise ValueError("未找到 TAVILY_API_KEY，请检查 .env 文件")  

# 使用读取到的 API Key 创建 TavilyClient 对象，该对象负责发起搜索请求
client = TavilyClient(api_key=api_key)
# 调用 client.search() 方法向 Tavily API 发起搜索请求
# 参数 "What is in Nvidia's new Blackwell GPU?" 是查询的问题
# include_answer=True 表示希望 Tavily 在返回结果中直接提供一个总结性的答案（而非只返回原始网页片段）
result = client.search("What is in Nvidia's new Blackwell GPU?",
                       include_answer=True) 
# 从返回的 result 字典中提取 "answer" 字段，并使用 print() 函数将其输出到控制台
# 注意：在 VSCode 中运行 Python 脚本时，必须显式使用 print() 才能看到输出

print(result["answer"])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

from ddgs import DDGS
import requests
from bs4 import BeautifulSoup
import re

city = "Bei Jing"
query = f"""
    what is your current weather in {city}?
    Should I travel there today?
"""

# ② 代理设置（如果需要）—— 使用字符串而不是字典
proxy = "http://127.0.0.1:7897"   # 改成你实际代理地址
ddg = DDGS(proxy=proxy)           # ✅ 传入字符串
 
# 如果没有代理，就用 ddg = DDGS()

# def search(query, max_results=3):
#     # 使用 backend="html" 或 "lite"
#     results = ddg.text(query, max_results=max_results, backend="html")
#     return [item["href"] for item in results]  
# urls = search(query)
# for url in urls:
#     print(url) 

def search(query, max_results=6):
    try:
        results = ddg.text(query, max_results=max_results)
        return [i["href"] for i in results]
    except Exception as e:
        print(f"returning previous results due to exception reaching ddg.")
        results = [ # cover case where DDG rate limits due to high deeplearning.ai volume
        "https://weather.com/weather/today/l/USCA0987:1:US",         "https://weather.com/weather/hourbyhour/l/54f9d8baac32496f6b5497b4bf7a277c3e2e6cc5625de69680e6169e7e38e9a8",
        ]
        return results  
  
for i in search(query):
    print(i)
def scrape_weather_info(url):
    """Scrape content from the given URL"""
    if not url:
        return "Weather information could not be found."
    # fetch data
    headers = {'User-Agent': 'Mozilla/5.0'}
    response = requests.get(url, headers=headers)
    if response.status_code != 200:
        return "Failed to retrieve the webpage."
    # parse result
    soup = BeautifulSoup(response.text, 'html.parser')
    return soup 
url = search(query)[0]  

# scrape first wesbsite
soup = scrape_weather_info(url)
# print(f"Website: {url}\n\n")
# print(str(soup.body)[:50000])  
weather_data = []
for tag in soup.find_all(['h1', 'h2', 'h3', 'p']):
    text = tag.get_text(" ", strip=True)
    weather_data.append(text)
  
# combine all elements into a single string
weather_data = "\n".join(weather_data) 
# remove all spaces from the combined text
weather_data = re.sub(r'\s+', ' ', weather_data)
print(f"Website: {url}\n\n")
print(weather_data)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

from ddgs import DDGS
import requests
from bs4 import BeautifulSoup
import re
from dotenv import load_dotenv
import os
from tavily import TavilyClient
import json
from pygments import highlight,lexers,formatters

load_dotenv()
api_key = os.getenv("TAVILY_API_KEY")

if not api_key:
    raise ValueError("未找到 TAVILY_API_KEY，请检查 .env 文件")
  
client = TavilyClient(api_key=api_key)

city = "Bei Jing"

query = f"""
    what is your current weather in {city}?
    Should I travel there today?
"""
result = client.search(query,max_results=1)  

data =result["results"][0]["content"]
# print(data)
# parse JSON

parsed_json = json.loads(data.replace("'", '"'))

# pretty print JSON with syntax highlighting
formatted_json = json.dumps(parsed_json, indent=4)
colorful_json = highlight(formatted_json,
                          lexers.JsonLexer(),
                          formatters.TerminalFormatter()) 
print(colorful_json)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

"""
基于 DeepSeek 模型 + Tavily 搜索工具的 LangGraph 智能体示例
支持多轮对话和工具调用，使用内存检查点保存线程状态。
"""
# ======================= 1. 导入依赖 =======================

import os
from dotenv import load_dotenv
# 加载 .env 文件中的环境变量（API 密钥等）
_ = load_dotenv()  
# LangGraph 核心组件
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver  # 内存检查点（简便，无需 SQLite） 
# 类型定义
from typing import TypedDict, Annotated
import operator 
# LangChain 消息类型
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
# DeepSeek 模型（OpenAI 兼容）
from langchain_openai import ChatOpenAI
# Tavily 搜索工具（新版包，需安装 langchain-tavily）
from langchain_tavily import TavilySearch  # 替代已弃用的 TavilySearchResults

# ======================= 2. 定义状态结构 =======================
class AgentState(TypedDict):
    """智能体的状态结构，使用 Annotated 实现消息自动累加"""
    messages: Annotated[list[AnyMessage], operator.add]  
# ======================= 3. 初始化工具 =======================
# 创建 Tavily 搜索工具实例，每次返回最多 2 条结果
tool = TavilySearch(max_results=2) 
# ======================= 4. 定义 Agent 类 =======================
class Agent:
    """
    支持工具调用的智能体封装类
    使用 LangGraph 构建执行图，包含 LLM 节点和动作节点，
    并自动路由：无工具调用则结束，有调用则执行工具后返回 LLM。
    """  
    def __init__(self, model, tools, checkpointer, system=""):
        """
        参数:
            model: 绑定工具后的 LLM 模型实例
            tools: 工具列表（每个工具应有 name, invoke 等方法）
            checkpointer: 检查点保存器（例如 MemorySaver）
            system: 系统提示词（可选）
        """
        self.system = system
        # 构建 StateGraph，状态类型为 AgentState
        graph = StateGraph(AgentState)

        # 添加两个节点
        graph.add_node("llm", self.call_openai)      # LLM 推理节点
        graph.add_node("action", self.take_action)   # 工具执行节点

        # 添加条件边：从 llm 节点出发，根据 exists_action 决定跳转到 action 或 END
        graph.add_conditional_edges(
            "llm",
            self.exists_action,
            {True: "action", False: END}
        )
        # 添加普通边：action 执行完毕后回到 llm 节点
        graph.add_edge("action", "llm")
        # 设置入口点为 llm
        graph.set_entry_point("llm")
        # 编译图，传入检查点保存器（支持多线程对话记忆）
        self.graph = graph.compile(checkpointer=checkpointer)
        # 将工具列表转换为 {工具名: 工具对象} 映射，便于快速调用
        self.tools = {t.name: t for t in tools}
        # 模型绑定工具（使模型能够生成 tool_calls）
        self.model = model.bind_tools(tools)
  
    def call_openai(self, state: AgentState):
        """LLM 节点：调用模型生成回复（可能包含工具调用请求）"""
        messages = state['messages']
        # 如果设置了系统提示词，将其插入到消息列表最前面
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        # 调用模型
        response = self.model.invoke(messages)
        # 返回新消息（将被自动添加到状态中的 messages 列表）
        return {'messages': [response]} 

    def exists_action(self, state: AgentState):
        """条件判断函数：检查最后一条消息是否包含工具调用"""
        last_message = state['messages'][-1]
        # 如果有 tool_calls 属性且非空，则返回 True
        return len(last_message.tool_calls) > 0

    def take_action(self, state: AgentState):
        """动作节点：执行工具调用，并将结果封装为 ToolMessage 返回"""
        last_message = state['messages'][-1]
        tool_calls = last_message.tool_calls 
        results = []
        for tool_call in tool_calls:
            print(f"调用工具: {tool_call}")   # 调试输出
            # 根据工具名获取工具实例，并调用其 invoke 方法
            tool_result = self.tools[tool_call['name']].invoke(tool_call['args'])
            # 构造 ToolMessage，必须包含 tool_call_id 和原工具名
            results.append(
                ToolMessage(
                    tool_call_id=tool_call['id'],
                    name=tool_call['name'],
                    content=str(tool_result)
                )
            )
        print("工具执行完毕，返回模型继续推理...")
        return {'messages': results}

# ======================= 5. 配置模型和检查点 =======================
# 设置系统提示词（指导智能体合理使用搜索工具）

prompt = """You are a smart research assistant. Use the search engine to look up information. \
You are allowed to make multiple calls (either together or in sequence). \
Only look up information when you are sure of what you want. \
If you need to look up some information before asking a follow up question, you are allowed to do that!

"""
# 创建 DeepSeek 模型实例（使用与 OpenAI 兼容的接口）
model = ChatOpenAI(
    model="deepseek-chat",                     # DeepSeek 对话模型
    base_url="https://api.deepseek.com/v1",    # DeepSeek API 地址
    api_key=os.getenv("DEEPSEEK_API_KEY")      # 从环境变量读取 API Key
)
# 创建内存检查点保存器（程序重启后记忆丢失，但无需 SQLite 依赖）
memory = MemorySaver()
# 实例化智能体
abot = Agent(model, [tool], system=prompt, checkpointer=memory)
# ======================= 6. 多线程对话示例 =======================
# 定义线程配置（每个 thread_id 代表一个独立的对话会话）
thread_1 = {"configurable": {"thread_id": "1"}}
thread_2 = {"configurable": {"thread_id": "2"}}
# ---- 线程1：询问旧金山天气 ----
print("=== 线程1：询问旧金山天气 ===")
messages = [HumanMessage(content="What is the weather in sf?")]
for event in abot.graph.stream({"messages": messages}, thread_1):
    # 遍历流式输出的事件，打印每个节点的消息
    for node_name, node_output in event.items():
        print(f"[{node_name}] 输出: {node_output['messages'][-1].content}")  

# ---- 线程1：继续询问洛杉矶天气（利用记忆） ----
print("\n=== 线程1：询问洛杉矶天气（同一会话） ===")
messages = [HumanMessage(content="What about in la?")]
for event in abot.graph.stream({"messages": messages}, thread_1):
    for node_name, node_output in event.items():
        print(f"[{node_name}] 输出: {node_output['messages'][-1].content}")


# ---- 线程1：比较两个城市温度 ----
print("\n=== 线程1：询问哪个更暖和（同一会话） ===")
messages = [HumanMessage(content="Which one is warmer?")]
for event in abot.graph.stream({"messages": messages}, thread_1):
    for node_name, node_output in event.items():
        print(f"[{node_name}] 输出: {node_output['messages'][-1].content}")

# ---- 线程2：新会话，同样的问题（无上下文记忆） ----
print("\n=== 线程2：新会话询问哪个更暖和（无历史） ===")
messages = [HumanMessage(content="Which one is warmer?")]
for event in abot.graph.stream({"messages": messages}, thread_2):
    for node_name, node_output in event.items():
        print(f"[{node_name}] 输出: {node_output['messages'][-1].content}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

"""
基于 DeepSeek 模型 + Tavily 搜索工具的 LangGraph 智能体
- 使用异步 SQLite 检查点（支持多线程记忆）
- 使用 astream_events 实现流式输出
- 修正导入路径为 langgraph.checkpoint.sqlite.aio
"""

import os
import asyncio
from dotenv import load_dotenv
_ = load_dotenv()
  

# 正确的异步 SQLite 检查点导入路径
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch  # 需安装 langchain-tavily

# ======================= 状态定义 =======================
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
# ======================= 工具 =======================

tool = TavilySearch(max_results=2)  
# ======================= Agent 类 =======================
class Agent:
    def __init__(self, model, tools, checkpointer, system=""):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        # 编译时传入有效的 checkpointer 实例
        self.graph = graph.compile(checkpointer=checkpointer)
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        response = self.model.invoke(messages)
        return {'messages': [response]}
  
    def exists_action(self, state: AgentState):
        last_message = state['messages'][-1]
        return len(last_message.tool_calls) > 0

    def take_action(self, state: AgentState):
        last_message = state['messages'][-1]
        tool_calls = last_message.tool_calls
        results = []
        for t in tool_calls:
            print(f"调用工具: {t}")
            result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print("工具执行完毕，返回模型继续推理...")
        return {'messages': results}

# ======================= 系统提示词 =======================
prompt = """You are a smart research assistant. Use the search engine to look up information. \
You are allowed to make multiple calls (either together or in sequence). \
Only look up information when you are sure of what you want. \
If you need to look up some information before asking a follow up question, you are allowed to do that!
""" 
# ======================= 模型配置（DeepSeek） =======================
model = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY")
)


# ======================= 异步主函数 =======================
async def main():
    # 重要：AsyncSqliteSaver.from_conn_string 返回异步上下文管理器，
    # 必须用 async with 才能获得真正的 checkpointer 实例
    async with AsyncSqliteSaver.from_conn_string(":memory:") as memory:
        abot = Agent(model, [tool], system=prompt, checkpointer=memory)
        thread = {"configurable": {"thread_id": "4"}}
        messages = [HumanMessage(content="What is the weather in SF?")]
        # 流式输出
        async for event in abot.graph.astream_events(
            {"messages": messages}, thread, version="v1"
        ):
            kind = event["event"]
            if kind == "on_chat_model_stream":
                content = event["data"]["chunk"].content
                if content:  # 过滤掉工具调用时的空内容
                    print(content, end="|")
        print("\n")  # 换行结束
        
        # 可选：同一线程继续对话，验证记忆功能

        print("继续询问洛杉矶天气：")

        messages2 = [HumanMessage(content="What about in LA?")]

        async for event in abot.graph.astream_events(
            {"messages": messages2}, thread, version="v1"
        ):
            if event["event"] == "on_chat_model_stream":
                content = event["data"]["chunk"].content
                if content:
                    print(content, end="|")
        print("\n")

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

Let| me| look| up| the| current| weather| in| San| Francisco|.|调用工具: {'name': 'tavily_search', 'args': {'query': 'current weather in San Francisco', 'search_depth': 'basic'}, 'id': 'call_00_QaGKAAhxxDmgUExKL9Md7517', 'type': 'tool_call'}
工具执行完毕，返回模型继续推理...
Let| me| try| again|.|调用工具: {'name': 'tavily_search', 'args': {'query': 'San Francisco weather today', 'search_depth': 'basic'}, 'id': 'call_00_0t5rBYaVJrV0Iry7XtIJ8882', 'type': 'tool_call'}
工具执行完毕，返回模型继续推理...
Here|'s| the| current| weather| in| San| Francisco| as| of| late| evening| on| **|May| |8|,| |202|6|**|:

|🌡|️| **|Temperature|**:| |54|°|F| (|12|.|2|°|C|),| but| feels| like| **|52|.|6|°|F| (|11|.|4|°|C|)|**
|⛅| **|Conditions|**:| Part|ly| cloudy|
|💨| **|Wind|**:| WSW| at| |5|.|6| mph| (|gust|ing| up| to| |8|.|6| mph|)
|💧| **|Hum|idity|**:| |86|%
|☁|️| **|Cloud| cover|**:| |50|%
|🌧|️| **|Ch|ance| of| rain|**:| |0|%
|👁|️| **|Visibility|**:| |9| miles|

|It|'s| a| cool|,| partly| cloudy| evening| in| SF| —| typical| for| this| time| of| year|!| If| you|'re| planning| for| tomorrow| (|May| |9|),| it|'s| expected| to| be| sunny| with| a| high| around| **|66|°|F| (|19|°|C|)**| in| the| afternoon|.|

继续询问洛杉矶天气：
调用工具: {'name': 'tavily_search', 'args': {'query': 'current weather in Los Angeles', 'search_depth': 'basic'}, 'id': 'call_00_DU0izSYvzKayq7bI5btZ5735', 'type': 'tool_call'}
工具执行完毕，返回模型继续推理...
Here|'s| the| current| weather| in| **|Los| Angeles|**| as| of| the| same| time| (|even|ing| of| May| |8|,| |202|6|):

|🌡|️| **|Temperature|**:| |61|°|F| (|16|.|1|°|C|),| feels| like| **|61|°|F|**
|☁|️| **|Conditions|**:| Over|cast| (|cloud|y|)
|💨| **|Wind|**:| South| at| |3|.|1| mph|
|💧| **|Hum|idity|**:| |83|%
|☁|️| **|Cloud| cover|**:| |100|%| (|full| over|cast|)
|🌧|️| **|Ch|ance| of| rain|**:| |0|%
|👁|️| **|Visibility|**:| |8| miles|

|**|Quick| comparison| —| SF| vs| LA|:**

||| || San| Francisco| || Los| Angeles| |
||---|---||---||
||| 🌡|️| Temp| || |54|°|F| (|12|°|C|)| || |61|°|F| (|16|°|C|)| |
||| ☁|️| Sky| || Part|ly| cloudy| || Over|cast| |
||| 💨| Wind| || WSW| |5|.|6| mph| || S| |3|.|1| mph| |
||| 💧| Humidity| || |86|%| || |83|%| |

|LA| is| about| **|7|°|F| warmer|**| than| SF| tonight|,| though| it|'s| more| over|cast|.| Both| are| dry| with| no| rain| expected|!|