目录

1. 项目结构与技术栈

2. 配置管理：config.py

3. 定义工具：weather_tool.py

4. 核心 Agent 逻辑：qwen.py

4.1. 初始化模型和工具

4.2. 实现动态系统提示 (Dynamic Prompt)

4.3. 调用与测试

5. 完整代码：

---

1. 项目结构与技术栈

本项目的核心目标是创建一个 Agent，它能够：

1. 根据用户提问，自主决定是使用内部知识库回答（如景点推荐）。

2. 在用户明确询问天气时，准确调用外部工具 (get_weather)。

3. 根据用户身份 (user 或 old)，调整 Agent 的回复风格（简洁或详细）。

技术栈：

• 大模型: 阿里通义千问 (qwen-plus)

• 框架: LangChain (用于 Agent 和工具管理)

• 工具: 心知天气API (用于天气查询)

• 配置: python-dotenv

• 环境: python 3.10.x

项目文件结构如下：

.
├── config.py           # 环境配置和API Key管理
├── qwen.py             # Agent 核心逻辑、动态 Prompt 和调用
└── tools
    └── weather_tool.py # 天气查询工具定义

第三方库：

langchain==1.0.0
langchain-community==1.0.0
langchain-core==1.0.0
dashscope
python-dotenv
requests

2. 配置管理：config.py

我们使用 python-dotenv 加载环境变量，并提供了一个自检函数来确保关键 API Key 不缺失。

# config.py
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv(override=True)

# API Keys (此处省略部分代码)
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")

# 模型配置
MODEL_NAME = "qwen-plus"
MODEL_TOP_P = 0.8

def check_required_keys():
    """检查必需的环境变量是否已加载"""
    required_keys = ["DASHSCOPE_API_KEY", "WEATHER_API_KEY"]
    missing_keys = [key for key in required_keys if not os.getenv(key)]
    
    if missing_keys:
        raise ValueError(f"缺失必需的环境变量: {', '.join(missing_keys)}")
    return True

# 在模块加载时自动检查
check_required_keys()

print("✅ 配置加载完成，必需的环境变量已验证")

3. 定义工具：weather_tool.py

我们定义了一个简单的天气查询工具 get_weather。关键点在于：

1. 使用 LangChain 的 @tool 装饰器。

2. 在 Docstring 中清晰地描述工具的用途、限制和参数，这对于 LLM 准确判断是否调用至关重要。

# tools/weather_tool.py
import os
import requests
from langchain.tools import tool
from config import WEATHER_API_KEY

@tool
def get_weather(location: str) -> str:
    """
    仅用于查询**当天**的**实时**天气情况（如温度、下雨情况）。
    
    警告：如果是要求制定旅行路线、攻略或历史信息，严禁调用此工具，请直接回答。
    
    参数:
        city (str): 必填，需要查询天气的城市名称。
    """
    # 心知天气API配置
    API_KEY = WEATHER_API_KEY
    if not API_KEY:
        return "API密钥未配置，请检查您的 .env 文件中的 WEATHER_API_KEY"
    
    # 心知天气API URL
    url = f"https://api.seniverse.com/v3/weather/now.json"
    
    # 请求参数
    params = {
        'key': API_KEY,
        'location': location,
        'language': 'zh-Hans',
        'unit': 'c'
    }
    
    try:
        # 发送请求
        response = requests.get(url, params=params, timeout=5)
        response.raise_for_status()
        data = response.json()
        
        # 解析数据
        if 'results' in data and len(data['results']) > 0:
            result = data['results'][0]
            location_name = result['location']['name']
            weather_info = result['now']
            
            return f"{location_name}今天天气：{weather_info['text']}，温度{weather_info['temperature']}°C"
    except requests.exceptions.RequestException as e:
        return f"获取天气信息失败：{str(e)}"
    except Exception as e:
        return f"解析天气数据失败：{str(e)}"

4. 核心 Agent 逻辑

4.1. 初始化模型和工具

我们使用 ChatTongyi 初始化通义千问模型。

# chat/qwen.py
from langchain_community.chat_models import ChatTongyi
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from config import DASHSCOPE_API_KEY, MODEL_NAME, MODEL_TOP_P
from tools.weather_tool import get_weather

# 初始化模型
tongyi_chat = ChatTongyi(
    model=MODEL_NAME,
    top_p=MODEL_TOP_P,
    dashscope_api_key=DASHSCOPE_API_KEY,
)

4.2. 实现动态系统提示 (Dynamic Prompt)

这是实现 Agent 个性化的关键。@dynamic_prompt 装饰器允许我们在每次 agent.invoke 调用时，根据传入的 context 动态生成不同的系统提示 (System Prompt)，从而控制 Agent 的行为。

动态系统提示：Agents - Docs by LangChain

# 动态系统提示
@dynamic_prompt
def user_role_prompt(request: ModelRequest) -> str:
    """Generate system prompt based on user role."""
    user_role = request.runtime.context.get("user_role", "user")
    
    # 基础设定：定义 Agent 的身份和工具使用规则
    base_prompt = (
        "你是一个资深的旅行顾问。请遵循以下规则："
        "1. 只有当用户明确询问天气时，才调用工具。"
        "2. 对于旅行路线规划、景点介绍等常规问题，请直接利用你的知识库回答。"
    )   

    if user_role == "old":
        return f"{base_prompt}, 提供详细的旅行计划。"
    elif user_role == "user":
        return f"{base_prompt}, 提供简洁的旅行建议。"

    return base_prompt

class Context(TypedDict):
    user_role: str

# 创建 Agent
agent = create_agent(
    model=tongyi_chat,
    tools=[get_weather],
    middleware=[user_role_prompt], # 注入动态提示
    context_schema=Context
)

4.3. 调用与测试

我们进行两次调用测试，验证 Agent 的两个核心能力：

测试一：知识库问答 (不调用工具)

• 输入: {"messages": [{"role": "user", "content": "北京景点推荐"}]}

• Context: {"user_role": "user"} (要求简洁建议)

Agent 会根据 System Prompt 规则 2，直接利用自身知识回答，并遵循简洁的风格。

测试二：工具调用 (Tool-Calling)

• 输入: {"messages": [{"role": "user", "content": "北京天气如何"}]}

• Context: {"user_role": "user"}

Agent 会根据 System Prompt 规则 1，准确识别“天气”关键词，触发 get_weather 工具调用，并最终返回工具结果。

运行结果示例：

# ... (第一次调用输出)
最终AI回复内容:
作为资深的旅行顾问，我为您推荐北京的几个必去景点：
1. 故宫博物院：感受明清皇家气势。
2. 天坛：体验古代祭祀文化。
3. 颐和园：欣赏皇家园林艺术。
...
模型: qwen-plus | 总消耗token数: 150 | 完成原因: stop

# ... (第二次调用输出 - 假设工具返回了结果)
最终AI回复内容:
北京今天天气：晴，温度25°C
模型: qwen-plus | 总消耗token数: 300 | 完成原因: stop

5. 完整代码：

# chat/qwen.py
import os
import sys
from typing import TypedDict
from langchain_community.chat_models import ChatTongyi
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest

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

# 导入配置（此时会自动检查环境变量）
from config import DASHSCOPE_API_KEY, MODEL_NAME, MODEL_TOP_P

print("✅ 环境变量加载成功，开始初始化 LangChain 组件...")

# 初始化模型
tongyi_chat = ChatTongyi(
    model=MODEL_NAME,
    top_p=MODEL_TOP_P,
    dashscope_api_key=DASHSCOPE_API_KEY,
)

# 导入工具
from tools.weather_tool import get_weather

# 动态系统提示
@dynamic_prompt
def user_role_prompt(request: ModelRequest) -> str:
    """Generate system prompt based on user role."""
    user_role = request.runtime.context.get("user_role", "user")
    # 基础设定：加入由于 Qwen 模型能力较强，需要明确告知它何时使用工具
    base_prompt = (
        "你是一个资深的旅行顾问。请遵循以下规则："
        "1. 只有当用户明确询问天气时，才调用工具。"
        "2. 对于旅行路线规划、景点介绍等常规问题，请直接利用你的知识库回答。"
    )   

    if user_role == "old":
        return f"{base_prompt}, 提供详细的旅行计划。"
    elif user_role == "user":
        return f"{base_prompt}, 提供简洁的旅行建议。"

    return base_prompt


class Context(TypedDict):
    user_role: str


agent = create_agent(
    model=tongyi_chat,
    tools=[get_weather],
    middleware=[user_role_prompt],
    context_schema=Context
)

# 调用代理
result = agent.invoke(
    {"messages": [{"role": "user", "content": "北京景点推荐"}]},
    context={"user_role": "user"}
)

# 提取关键字段
ai_response = result['messages'][-1]

# 打印关键字段
print("最终AI回复内容:")
print(ai_response.content)

# 在一行中展示关键字段
print(f"模型: {ai_response.response_metadata['model_name']} | "
      f"总消耗token数: {ai_response.response_metadata['token_usage']['total_tokens']} | "
      f"完成原因: {ai_response.response_metadata['finish_reason']}")

# 调用代理
result = agent.invoke(
    {"messages": [{"role": "user", "content": "北京天气如何"}]},
    context={"user_role": "user"}
)

# 提取关键字段
ai_response = result['messages'][-1]

# 打印关键字段
print("最终AI回复内容:")
print(ai_response.content)

# 在一行中展示关键字段
print(f"模型: {ai_response.response_metadata['model_name']} | "
      f"总消耗token数: {ai_response.response_metadata['token_usage']['total_tokens']} | "
      f"完成原因: {ai_response.response_metadata['finish_reason']}")