from langchain.agents import create_agent

from langchain_openai import ChatOpenAI

from langchain.tools import tool

 

# 加载 .env 文件中的 SILICONFLOW_API_KEY

load_dotenv()

 

# 1. 定义工具：必须要写清晰的 Docstring（函数注释）

@tool

def get_weather(city: str) -> str:

    """获取指定城市的实时天气情况。"""

    return f"{city} 的天气总是阴雨连连！"

 

# 2. 初始化 SiliconFlow 模型

llm = ChatOpenAI(

    model="deepseek-ai/DeepSeek-V3", 

    api_key=os.getenv("SILICONFLOW_API_KEY"),

    base_url="https://api.siliconflow.cn/v1",

    temperature=0.1

)

 

# 3. 创建 Agent

agent = create_agent(

    model=llm,

    tools=[get_weather],

    system_prompt="你是一个乐于助人的天气助手。"

)

 

# 4. 运行

if __name__ == "__main__":

    response = agent.invoke(

        {"messages": [{"role": "user", "content": "北京天气怎么样？"}]}

    )

    # 打印最后一条消息（AI 的回答）

from dataclasses import dataclass

from dotenv import load_dotenv

 

from langchain_openai import ChatOpenAI

from langchain.agents import create_agent

from langchain.tools import tool, ToolRuntime

from langgraph. m.suishougou.cn，checkpoint.memory import InMemorySaver

from langchain.agents.structured_output import ToolStrategy

 

load_dotenv()

 

# 1. 定义系统提示词（要求 AI 做个“谐音梗”专家）

SYSTEM_PROMPT = """你是一位专业的资深天气预报员，说话极其幽默，特别喜欢讲“谐音梗”。

 

你有权使用以下两个工具：

- get_weather_for_location: 获取特定城市的天气。

- get_user_location: 获取当前用户的所在地。

 

如果用户询问天气，请务必确认地点。如果用户说的是“我这儿”或“我这里”，请调用 get_user_location 获取他们的位置。"""

 

# 2. 定义上下文结构（存储用户信息）

@dataclass

class Context:

    """自定义运行时上下文。"""

    user_id: str

 

# 3. 定义工具

@tool

def get_weather_for_location(city: str) -> str:

    """获取指定城市的天气情况。"""

    return f"{city} 的天气总是阳光灿烂！"

 

@tool

def get_user_location(runtime: ToolRuntime[Context]) -> str:

    """根据用户 ID 获取其地理位置。"""

    user_id = runtime.context.user_id

    # 模拟数据库：用户 1 在北京，其他在珠海

    return "北京" if user_id == "1" else "珠海"

 

# 4. 配置你的模型 (SiliconFlow)

model = ChatOpenAI(

    model="deepseek-ai/DeepSeek-V3",

    api_key=os.getenv("SILICONFLOW_API_KEY"),

    base_url="https://api.siliconflow.cn/v1",

    temperature=0.7  # 稍微提高随机性，让谐音梗更有创意

)

 

# 5. 定义响应格式（汉化字段描述，让 AI 明白填什么）

@dataclass

class ResponseFormat:

    """Agent 的结构化回复格式。"""

    # 幽默的谐音梗回复

    punny_response: str

    # 任何关于天气的有趣信息（如果有的话）

    weather_conditions: str | None = None

 

# 6. 设置记忆存储

checkpointer = InMemorySaver()

 

# 7. 创建 Agent

agent = create_agent(

    model=model,

    system_prompt=SYSTEM_PROMPT,

    tools=[get_user_location, get_weather_for_location],

    context_schema=Context,

    response_format=ToolStrategy(ResponseFormat),

    checkpointer=checkpointer

)

 

# --- 运行测试 ---

 

# 设置会话 ID

config = {"configurable": {"thread_id": "test_user_001"}}

 

# 第一次提问

print("--- 第一次对话 ---")

response = agent.invoke(

    {"messages": [{"role": "user", "content": "外面天气怎么样？"}]},

    config=config,

    context=Context(user_id="1")

)

 

# 访问结构化结果

res = response['structured_response']

print(f"AI: {res.punny_response}")

print(f"详情: {res.weather_conditions}")

 

print("\n--- 第二次对话（测试记忆） ---")

# 第二次提问

response = agent.invoke(

    {"messages": [{"role": "user", "content": "谢谢你！我在哪里？"}]},

    config=config,

    context=Context(user_id="1")

)

 

res = response['structured_response']

用户问：“外面天气怎么样？”（没说地点）。
AI 思考：系统提示词告诉我，如果地点不明，先去查用户在哪。
动作：AI 调用 get_user_location -> 得到“北京” -> 调用 get_weather_for_location -> 得到天气。
回复：根据 ResponseFormat 吐出一个谐音梗。

第二次对话：

用户问：“谢谢你！我在哪里？”
AI 思考：由于有 thread_id 和 checkpointer，AI 记得上一次查询的结果。
动作：它不需要再次查工具，直接从记忆里提取信息并回复。

 

 

  

 

 

 

 

 

  

 

 

 

1. 使用opencv读取视频基本信息，包括视频的fps（假设是60 fps）, total_frames（假设视频为15s，则为900帧）, width（1280）和height（720）
2. 按照指定的sample_fps进行帧采样，这里指定为2，也就是每1s采2帧。原视频为15秒900帧的话，在采样后就是30帧，采样间隔就是60/2=30，也就是每30帧取1帧。 这样就得到了后续用于推理的帧序列（0, 31, 62, …, 899）
3. 动态调整分辨率。在传入的messages参数里除了video的路径外，还包括, min_pixels（指定为64 * 32 * 32 = 65,536）, total_pixels（指定为20480 * 32 * 32=20,971,520） 和 sample_fps的信息。sample_fps的作用上面已经提到。total_pixels和min_pixels用于约束调整分辨率。
   1. 假设原始视频是1280 x 720的分辨率，也就是每帧包含921,600个像素。那30帧 * 921,600 = 27,648,000总像素，大于指定的max_pixels。所以会继续对帧序列进行像素缩放
4. 调整到patch_size的整数倍。Patch size是ViT（Vision Transformer）架构的核心设计，它是将图像切分为固定大小的patch，每个patch最终作为1个token输入到transformer中。这里patch size指定为16，也就是每个token本质上是一个16x16的像素区域。所以需要基于patch_size的设置，将每帧调整到patch_size的整数倍
   1. 对齐到patch_size的整数倍后，最终分辨率为1152 x 640
5. 最后将帧列表，也就是抽取的30帧，做了分辨率调整，做了基于patch_size调整后的帧序列，转为PyTorch张量并返回。

最后可以看到video_inputs的结果为单元素的list，这单个元素里为1个tuple，包含2部分：

1. 一个shape为([30, 3, 640, 1152]) 的张量，分别代表帧id、RGB通道、缩放后的分辨率640 x 1152
2. 另一个为视频的metadata信息，包括fps和frame_indices（采样后的帧序列），以及total_num_frames

首先看video_grid_thw，这个表示的patch布局，前面提到缩放后的分辨率为640 x 1152，对应到patch=16后的patch布局则为40 x 72。但这里有个问题， 为什么采样时帧数为30，这里变成了15？

通过阅读Qwen3VLVideoProcessor的源码[1]可以了解到有一个重要参数是temporal_patch_size=2（也可以在模型的config文件里看到这个参数）。 这个参数代表的含义是：模型一次看多少帧作为一个token。也可以理解为每个video token覆盖的连续帧数。在这里指定为2，也就表示每2帧合并成1个时间patch，这样就天然包含了帧序列中运动/变化的信息。这个参数越小，能捕捉到的动作变化就越细，对应消耗的token也越多。所以这就解释了为什么输入模型的patch布局为15 * 40 * 72。

再看pixel_value_videos的shape为torch.Size([43200, 1536])：

1. 第一维的43200很好理解，就是这个视频的patch数量，也就是15 * 40 * 72 = 43,200。
2. 现在我们已知输入模型的1个vision token（也就是patch token）的组成为：2帧 * patch像素。每个patch对应的像素为：2 * 3(RGB通道) * (16*16) =1536，这就对应到了pixel_value_videos的第二维，也就是每个patch的维度。

最后再看input_ids结果，其shape为([1, 10938])，对其做decode的结果为：