1
from langchain_openai import ChatOpenAI  # 调用 OpenAI 的聊天模型接口
2
from langchain.memory import ConversationBufferMemory  # 简单记忆模块
3
from langchain.agents import initialize_agent, AgentType  # 初始化代理
4
from langchain.tools import BaseTool  # 工具类
5
import requests  # HTTP 请求库
6
from typing import Optional  # 类型提示
7
import os
8
from dotenv import load_dotenv  # 环境变量加载库
9

10
# 加载环境变量
11
load_dotenv()
12

13
# 从环境变量获取配置信息
14
api_key = os.getenv("OPENAI_API_KEY")
15
base_url = os.getenv("OPENAI_BASE_URL")
16
model = os.getenv("MODEL_NAME")
17
qweather_api_key = os.getenv("QWEATHER_API_KEY")
18

19
# 初始化大语言模型
20
llm = ChatOpenAI(
21
    base_url=base_url,
22
    api_key=api_key,
23
    model=model,
24
    temperature=0,  # 设置温度为0，确保输出一致性
25
)

1
# 自定义天气查询工具
2
class WeatherTool(BaseTool):
3
    name: str = "weather_query"
4
    description: str = "查询指定城市的天气情况，通过输入城市名称获取当日的天气信息。"
5

6
    def _run(self, city: str, run_manager: Optional[object] = None) -> str:
7
        """
8
        工具的核心执行逻辑
9
        分为两步：1) 根据城市名获取城市ID，2) 使用城市ID查询天气
10
        """
11
        # 第一步：获取城市的 location ID
12
        city_id_res_url = f"https://geoapi.qweather.com/v2/city/lookup?location={city}&key={qweather_api_key}"
13
        city_id_res = requests.get(city_id_res_url)
14

15
        if city_id_res.status_code == 200:
16
            city_id_data = city_id_res.json()
17
            if city_id_data['code'] == '200' and city_id_data['location']:
18
                city_id = city_id_data['location'][0]['id']
19
            else:
20
                return f"无法获取 {city} 的地理信息，请检查城市名称是否正确。"
21
        else:
22
            return f"地理信息查询失败，状态码: {city_id_res.status_code}"
23

24
        # 第二步：使用城市ID查询实时天气
25
        weather_url = f"https://devapi.qweather.com/v7/weather/now?location={city_id}&key={qweather_api_key}"
26
        weather_res = requests.get(weather_url)
27

28
        if weather_res.status_code == 200:
29
            weather_data = weather_res.json()
30
            if weather_data['code'] == '200':
31
                weather = weather_data['now']
32
                weather_info = (
33
                    f"{city} 的当前天气情况：{weather['text']}，"
34
                    f"温度：{weather['temp']}°C，"
35
                    f"湿度：{weather['humidity']}%，"
36
                    f"风向：{weather['windDir']}，"
37
                    f"风力：{weather['windScale']}级"
38
                )
39
                return weather_info
40
            else:
41
                return f"无法获取 {city} 的天气信息，错误代码：{weather_data['code']}"
42
        else:
43
            return f"天气查询失败，状态码: {weather_res.status_code}"
44

45
    def _arun(self, city: str, run_manager: Optional[object] = None) -> str:
46
        """异步执行方法（本例中未实现）"""
47
        raise NotImplementedError("异步运行方法未实现")

1
# 初始化记忆模块
2
memory = ConversationBufferMemory(
3
    memory_key="chat_history",  # 内存中存储对话历史的键名
4
    return_messages=True,       # 返回消息格式而非字符串
5
    k=5                        # 保留最近5轮对话
6
)

1
# 工具列表配置
2
tools = [WeatherTool()]
3

4
# 系统提示设计
5
system_message = """你是一个专业的天气助手，擅长跟踪和维护对话的上下文。
6

7
核心能力：
8
1. 查询任何城市的实时天气信息
9
2. 基于天气条件提供穿衣、出行建议
10
3. 维持对话上下文，理解用户的隐式引用
11

12
对话原则：
13
- 当用户提到某个城市后，在后续对话中保持对该城市的引用
14
- 如果用户询问"这个城市"、"那里"等代词，请参考之前提到的城市
15
- 基于天气信息主动提供实用的生活建议
16
- 保持友好、专业的对话风格"""
17

18
# 初始化智能代理
19
agent = initialize_agent(
20
    tools=tools,
21
    llm=llm,
22
    agent_type=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
23
    memory=memory,
24
    verbose=True,  # 启用详细输出，便于调试
25
    agent_kwargs={"system_message": system_message}
26
)

1
# 第一轮对话：查询天气
2
print("=== 第一轮对话：天气查询 ===")
3
weather_response = agent.invoke("今天武汉的天气怎么样？")
4
print("Agent 响应:", weather_response['output'])
5

6
# 查看记忆内容
7
print("\n对话历史记录:")
8
print(memory.load_memory_variables({}))

1
# 第二轮对话：基于天气的建议
2
print("\n=== 第二轮对话：生活建议 ===")
3
advice_response = agent.invoke("根据刚才的天气情况，今天适合穿什么衣服？需要带伞吗？")
4
print("Agent 响应:", advice_response['output'])

1
# 第三轮对话：测试上下文理解
2
print("\n=== 第三轮对话：上下文理解 ===")
3
context_response = agent.invoke("这个地方有什么好玩的景点推荐？")
4
print("Agent 响应:", context_response['output'])
5

6
# 查看完整对话历史
7
print("\n=== 完整对话历史 ===")
8
chat_history = memory.load_memory_variables({})["chat_history"]
9
for i, message in enumerate(chat_history):
10
    print(f"{i+1}. {message.type}: {message.content}")

1
from langchain.memory import ConversationBufferWindowMemory
2

3
# 滑动窗口记忆：只保留最近N轮对话
4
window_memory = ConversationBufferWindowMemory(
5
    memory_key="chat_history",
6
    return_messages=True,
7
    k=3  # 只保留最近3轮对话
8
)
9

10
# 可以替换原有的 memory 参数
11
agent_with_window = initialize_agent(
12
    tools=tools,
13
    llm=llm,
14
    agent_type=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
15
    memory=window_memory,
16
    verbose=True
17
)

1
from langchain_community.chat_message_histories import ChatMessageHistory
2
from langchain_core.runnables import RunnableWithMessageHistory
3
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
4

5
# 定义现代化的提示模板
6
modern_prompt = ChatPromptTemplate.from_messages([
7
    ("system", """你是一个专业的天气助手，可以查询天气信息并提供生活建议。
8
当需要查询天气时，请使用 weather_query 工具。"""),
9
    MessagesPlaceholder(variable_name="history"),
10
    ("human", "{input}"),
11
])
12

13
# 创建链式调用
14
chain = modern_prompt | llm
15

16
# 消息历史管理
17
message_histories = {}
18

19
def get_message_history(session_id):
20
    if session_id not in message_histories:
21
        message_histories[session_id] = ChatMessageHistory()
22
    return message_histories[session_id]
23

24
# 带历史记录的链
25
chain_with_history = RunnableWithMessageHistory(
26
    chain,
27
    get_message_history,
28
    input_messages_key="input",
29
    history_messages_key="history",
30
)