Question: input question to answer Thought: consider previous and subsequent steps Action: ``` $JSON_BLOB ``` Observation: action result ... (repeat Thought/Action/Observation N times) Thought: I know what to respond Action: ``` {{ "action": "Final Answer", "action_input": "Final response to human" }} ```
Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation:. Thought:
def _get_llm_math(llm: BaseLanguageModel) -> BaseTool: return Tool( name="Calculator", description="Useful for when you need to answer questions about math.", func=LLMMathChain.from_llm(llm=llm).run, coroutine=LLMMathChain.from_llm(llm=llm).arun, )
我们看看此时的prompt
Respond to the human as helpfully and accurately as possible. You have access to the following tools:
Calculator: Useful for when you need to answer questions about math., args: {{'tool_input': {{'type': 'string'}}}} get current time: 用来获取当前时间. input should be 'now', args: {{'tool_input': {{'type': 'string'}}}}
Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).
Valid "action" values: "Final Answer" or Calculator, get current time
Question: input question to answer Thought: consider previous and subsequent steps Action: ``` $JSON_BLOB ``` Observation: action result ... (repeat Thought/Action/Observation N times) Thought: I know what to respond Action: ``` {{ "action": "Final Answer", "action_input": "Final response to human" }} ```
Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation:. Thought:
相比上一个例子，多了一个名叫Calculator的prompt: Calculator: Useful for when you need to answer questions about math., args: {{'tool_input': {{'type': 'string'}}}}
def getLocationId(city): d = collections.defaultdict(str) try: df = pd.read_csv("./data/datasets/virus/China-City-List-latest.csv", encoding='utf-8') except Exception as e: print(e) for i, row in df.iterrows(): d[row['Location_Name_ZH']] = row['Location_ID'] return d[city] if city in d else ''
def get_weather(location): key = "你的和风天气API key" id = getLocationId(location) if not id: return "没有这个城市" base_url = 'https://devapi.qweather.com/v7/weather/now?' params = {'location': id, 'key': key, 'lang': 'zh'} response = requests.get(base_url, params=params) data = response.json() if data["code"] != "200": return "没有这个城市的天气情况" return get_weather_info(data)
def get_weather_info(info): if info["code"] != "200": return "没有这个城市的天气情况" # result = f'现在天气{info["hourly"][0]["text"]}，温度 {info["hourly"][0]["temp"]} 度, 未来 24 小时天气{info["hourly"][-1]["text"]}，温度 {info["hourly"][-1]["temp"]} 度。'
class StructuredTool(BaseTool): """Tool that can operate on any number of inputs."""
description: str = "" args_schema: Type[BaseModel] = Field(..., description="The tool schema.") """The input arguments' schema.""" func: Optional[Callable[..., Any]] """The function to run when the tool is called.""" coroutine: Optional[Callable[..., Awaitable[Any]]] = None """The asynchronous version of the function."""
Calculator: Useful for when you need to answer questions about math., args: {{'tool_input': {{'type': 'string'}}}} get current time: 用来获取当前时间. input should be 'now'。当需要获取今天、明天、后天等的日期时，你应该调用此函数获取今天的日期, args: {{'tool_input': {{'type': 'string'}}}} get current weather: 用来获取当地当天的天气信息，输入应该是城市名称, args: {{'tool_input': {{'type': 'string'}}}} get future weather: 用来获取当地今天和未来六天的天气信息。, args: {{'location': {{'title': 'Location', 'description': '城市名称', 'type': 'string'}}, 'date': {{'title': 'Date', 'description': '日期，格式：yyyy-mm-dd，如：2021-11-15', 'type': 'string'}}}}
Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).
Valid "action" values: "Final Answer" or Calculator, get current time, get current weather, get future weather
import collections import random import requests import datetime import pandas as pd from langchain.tools import Tool, StructuredTool from langchain.agents import initialize_agent from langchain.chat_models import ChatOpenAI from langchain.agents import load_tools from langchain.agents import AgentType
from pydantic import BaseModel, Field from serpapi.baidu_search import BaiduSearch
def getLocationId(city): d = collections.defaultdict(str) try: df = pd.read_csv("./data/datasets/virus/China-City-List-latest.csv", encoding='utf-8') except Exception as e: print(e) for i, row in df.iterrows(): d[row['Location_Name_ZH']] = row['Location_ID'] return d[city] if city in d else ''
def get_weather(location): key = "你的和风天气API key" id = getLocationId(location) if not id: return "没有这个城市" base_url = 'https://devapi.qweather.com/v7/weather/now?' params = {'location': id, 'key': key, 'lang': 'zh'} response = requests.get(base_url, params=params) data = response.json() if data["code"] != "200": return "没有这个城市的天气情况" return get_weather_info(data)
class FutureWeatherInput(BaseModel): location: str = Field(description="城市名称") date: str = Field(description="日期，格式：yyyy-mm-dd，如：2021-11-15")
def get_future_weather(location, date): key = "你的和风天气API key" id = getLocationId(location) if not id: return "没有这个城市" base_url = 'https://devapi.qweather.com/v7/weather/7d?' params = {'location': id, 'key': key, 'lang': 'zh'} response = requests.get(base_url, params=params) data = response.json() if data["code"] != "200": return "没有这个城市的天气情况"
result = {} daily = data["daily"] for item in daily: fxDate = item["fxDate"]
#定义获取当前时间 time_tool = Tool( name='get current time', func= get_time, description="用来获取当前时间. input should be 'now'。当需要获取今天、明天、后天等的日期时，你应该调用此函数获取今天的日期""" )
weather_tool = Tool( name='get current weather', func= get_weather, description="用来获取当地当天的天气信息，输入应该是城市名称""" )
