LangChain 1.0 Agent结构化输出、运行时参数、提示词系统综合应用

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool,ToolRuntime

from helper import WeatherAPI,BaiduTTSAPI
from dataclasses import dataclass
from common import Common

common = Common()

checkpointer = InMemorySaver()
model = common.get_model("THUDM/GLM-4-9B-0414")

system_prompt = """
你是一个天气助手，总是会用诙谐幽默的方式回答问题。
你可以访问一个工具：
get_weather:使用这个工具查询指定城市的天气情况
get_user_location:使用这个工具可以获取用户所在的城市信息

如果用户问关于天气问题，确保你知道用户的城市信息，如果你搞清楚他们所在的城市信息，调用get_weather获取天气。
"""
@dataclass
class Context:
    """
    用户运行时上下文参数
    """
    user_id: str

@dataclass
class ResponseFormat:
    """
    用户运行时返回结果格式
    """
    # 诙谐幽默的回答（必须）
    punny_response:str
    # 只有关于天气的问题会有
    weather_conditions:str |None = None

@tool
def get_user_location(runtime:ToolRuntime[Context]) -> str:
    """获取用户所在城市信息"""
    # print(runtime)
    user_id = runtime.context.user_id

    return "北京" if user_id == "1" else "上海"
@tool
def get_weather(city: str) -> str:
    """获取天气信息"""
    try:
        weather_tool = WeatherAPI()
        return weather_tool.get_weather(city)
    except Exception as e:
        return f"获取天气信息失败：{e}"

agent = create_agent(
    model=model,
    system_prompt=system_prompt,
    tools=[get_user_location,get_weather],
    checkpointer=checkpointer,
    response_format=ResponseFormat, # 返回结果格式
    context_schema=Context # 上下文参数
)
config = {
    "configurable":{"thread_id":"1"}
}
# 第一轮问答
response = agent.invoke(
    {"messages":[{"role":"user","content":"谢谢"}]},
    config=config,
    context=Context(user_id="1")
)
text = response['structured_response'].punny_response
print(text)
# 使用百度TTS读出大模型回答
tts = BaiduTTSAPI()
tts.baidu_tts_speak(text)