#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import os
from typing import Type, Any, TypedDict, Dict, Optional
import requests
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig, RunnablePassthrough
from langchain_core.tools import BaseTool, render_text_description_and_args
from langchain_openai import ChatOpenAI
from pydantic import Field, BaseModel
import dotenv
dotenv.load_dotenv()
# ============ 第一步:定义工具(与原生方式相同)============
class WeatherArgsSchema(BaseModel):
city: str = Field(description="需要查询天气的城市,例如:北京")
class WeatherTool(BaseTool):
"""查询指定城市的天气预报"""
name: str = "weather_query"
description: str = "当你需要查询天气信息时使用此工具"
args_schema: Type[BaseModel] = WeatherArgsSchema
def _run(self, city: str) -> str:
# 实际调用天气 API
return f"{city}今天晴天,温度 25°C"
# ============ 第二步:创建工具字典和执行函数 ============
weather_tool = WeatherTool()
tool_dict = {weather_tool.name: weather_tool}
tools = [weather_tool]
# 定义工具调用请求的类型
class ToolCallRequest(TypedDict):
name: str
arguments: Dict[str, Any]
# 定义工具执行函数
def invoke_tool(
tool_call_request: ToolCallRequest,
config: Optional[RunnableConfig] = None,
) -> str:
"""执行工具调用"""
name = tool_call_request["name"]
requested_tool = tool_dict.get(name)
return requested_tool.invoke(tool_call_request.get("arguments"), config=config)
# ============ 第三步:构建 Prompt(关键!)============
# 将工具列表渲染成文本描述
rendered_tools = render_text_description_and_args(tools)
system_prompt = f"""你是一个聊天机器人,可以访问以下工具。
以下是每个工具的名称和描述:
{rendered_tools}
根据用户输入,返回要使用的工具的名称和输入。
将您的响应作为具有`name`和`arguments`键的JSON块返回。
`arguments`应该是一个字典,其中键对应于参数名称,值对应于请求的值。"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{query}")
])
# ============ 第四步:创建执行链 ============
# 使用不支持函数调用的模型
llm = ChatOpenAI(model="moonshot-v1-8k", temperature=0)
# 链路:Prompt -> LLM -> 解析JSON -> 执行工具
chain = (
prompt
| llm
| JsonOutputParser() # 关键:解析模型输出的 JSON
| RunnablePassthrough.assign(output=invoke_tool) # 执行工具并将结果赋值给 output
)
# ============ 第五步:调用 ============
result = chain.invoke({"query": "北京今天天气怎么样?"})
print(result)
# 输出类似:
# {
# "name": "weather_query",
# "arguments": {"city": "北京"},
# "output": "北京今天晴天,温度 25°C"
# }