Skip to content

Instantly share code, notes, and snippets.

@rajivmehtaflex
Last active August 1, 2024 05:17
Show Gist options
  • Select an option

  • Save rajivmehtaflex/8295cafdbb18a602b7c4fe90579a65e8 to your computer and use it in GitHub Desktop.

Select an option

Save rajivmehtaflex/8295cafdbb18a602b7c4fe90579a65e8 to your computer and use it in GitHub Desktop.
This GIST conain Langchain related references.
Display the source blob
Display the rendered blob
Raw
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "1864648c-bed7-4bbc-b083-9622f47e26b7",
"metadata": {},
"outputs": [],
"source": [
"import transformers\n",
"import torch\n",
"import gc"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "535201b2-95b7-4a5c-aa1b-5f95f9e695ad",
"metadata": {},
"outputs": [],
"source": [
"model_name = \"teknium/OpenHermes-2.5-Mistral-7B\"\n",
"\n",
"def delete_model():\n",
" global model\n",
" global tokenizer\n",
" model = None\n",
" tokenizer = None\n",
" gc.collect()\n",
" torch.cuda.empty_cache()\n",
"\n",
"def load_model(model_name: str):\n",
" tokenizer = transformers.AutoTokenizer.from_pretrained(model_name)\n",
"\n",
" with torch.device(\"cuda:0\"):\n",
" model = transformers.AutoModelForCausalLM.from_pretrained(model_name, torch_dtype=torch.bfloat16).eval()\n",
" \n",
" return tokenizer, model\n",
"\n",
"tokenizer, model = load_model(model_name)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "88eebb0f-f063-4138-af27-16c249905698",
"metadata": {},
"outputs": [],
"source": [
"import inspect\n",
"import json\n",
"from typing import get_type_hints\n",
"\n",
"class Article:\n",
" pass\n",
"\n",
"class Weather:\n",
" pass\n",
"\n",
"class Directions:\n",
" pass\n",
"\n",
"def calculate_mortgage_payment(loan_amount: int, interest_rate: float, loan_term: int) -> float:\n",
" \"\"\"Get the monthly mortgage payment given an interest rate percentage.\"\"\"\n",
" \n",
" # TODO: you must implement this to actually call it later\n",
" pass\n",
"\n",
"def get_article_details(title: str, authors: list[str], short_summary: str, date_published: str, tags: list[str]) -> Article:\n",
" '''Get article details from unstructured article text.\n",
"date_published: formatted as \"MM/DD/YYYY\"'''\n",
" \n",
" # TODO: you must implement this to actually call it later\n",
" pass\n",
"\n",
"def get_weather(zip_code: str) -> Weather:\n",
" \"\"\"Get the current weather given a zip code.\"\"\"\n",
" \n",
" # TODO: you must implement this to actually call it later\n",
" pass\n",
"\n",
"def get_directions(start: str, destination: str) -> Directions:\n",
" \"\"\"Get directions from Google Directions API.\n",
"start: start address as a string including zipcode (if any)\n",
"destination: end address as a string including zipcode (if any)\"\"\"\n",
" \n",
" # TODO: you must implement this to actually call it later\n",
" pass\n",
"\n",
"def get_type_name(t):\n",
" name = str(t)\n",
" if \"list\" in name or \"dict\" in name:\n",
" return name\n",
" else:\n",
" return t.__name__\n",
"\n",
"def serialize_function_to_json(func):\n",
" signature = inspect.signature(func)\n",
" type_hints = get_type_hints(func)\n",
"\n",
" function_info = {\n",
" \"name\": func.__name__,\n",
" \"description\": func.__doc__,\n",
" \"parameters\": {\n",
" \"type\": \"object\",\n",
" \"properties\": {}\n",
" },\n",
" \"returns\": type_hints.get('return', 'void').__name__\n",
" }\n",
"\n",
" for name, _ in signature.parameters.items():\n",
" param_type = get_type_name(type_hints.get(name, type(None)))\n",
" function_info[\"parameters\"][\"properties\"][name] = {\"type\": param_type}\n",
"\n",
" return json.dumps(function_info, indent=2)\n",
"\n",
"print(serialize_function_to_json(get_article_details))"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d2574b08-bed3-4b92-b768-493f57de8499",
"metadata": {},
"outputs": [],
"source": [
"import xml.etree.ElementTree as ET\n",
"import re\n",
"\n",
"def extract_function_calls(completion):\n",
" completion = completion.strip()\n",
" pattern = r\"(<multiplefunctions>(.*?)</multiplefunctions>)\"\n",
" match = re.search(pattern, completion, re.DOTALL)\n",
" if not match:\n",
" return None\n",
" \n",
" multiplefn = match.group(1)\n",
" root = ET.fromstring(multiplefn)\n",
" functions = root.findall(\"functioncall\")\n",
" return [json.loads(fn.text) for fn in functions]"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4059915c-ce53-4178-9707-a930428254ab",
"metadata": {},
"outputs": [],
"source": [
"def generate_hermes(prompt):\n",
" fn = \"\"\"{\"name\": \"function_name\", \"arguments\": {\"arg_1\": \"value_1\", \"arg_2\": value_2, ...}}\"\"\"\n",
" prompt = f\"\"\"<|im_start|>system\n",
"You are a helpful assistant with access to the following functions:\n",
"\n",
"{serialize_function_to_json(get_weather)}\n",
"\n",
"{serialize_function_to_json(calculate_mortgage_payment)}\n",
"\n",
"{serialize_function_to_json(get_directions)}\n",
"\n",
"{serialize_function_to_json(get_article_details)}\n",
"\n",
"To use these functions respond with:\n",
"<multiplefunctions>\n",
" <functioncall> {fn} </functioncall>\n",
" <functioncall> {fn} </functioncall>\n",
" ...\n",
"</multiplefunctions>\n",
"\n",
"Edge cases you must handle:\n",
"- If there are no functions that match the user request, you will respond politely that you cannot help.<|im_end|>\n",
"<|im_start|>user\n",
"{prompt}<|im_end|>\n",
"<|im_start|>assistant\"\"\"\n",
"\n",
" tokens = tokenizer(prompt, return_tensors=\"pt\").to(model.device)\n",
" input_size = tokens.input_ids.numel()\n",
" with torch.inference_mode():\n",
" generated_tokens = model.generate(**tokens, use_cache=True, do_sample=True, temperature=0.2, top_p=1.0, top_k=0, max_new_tokens=512, eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.eos_token_id)\n",
"\n",
" return tokenizer.decode(generated_tokens.squeeze()[input_size:], skip_special_tokens=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "19b083de-1a97-4d48-9202-e1902ed3b041",
"metadata": {},
"outputs": [],
"source": [
"prompts = [\n",
" \"What's the weather in 10001?\",\n",
" \"Determine the monthly mortgage payment for a loan amount of $200,000, an interest rate of 4%, and a loan term of 30 years.\",\n",
" \"What's the current exchange rate for USD to EUR?\"\n",
"]\n",
"\n",
"for prompt in prompts:\n",
" completion = generate_hermes(prompt)\n",
" functions = extract_function_calls(completion)\n",
"\n",
" if functions:\n",
" print(functions)\n",
" else:\n",
" print(completion.strip())\n",
" print(\"=\"*100)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ed44dd0f-0370-47df-a088-2f0f527b5622",
"metadata": {},
"outputs": [],
"source": [
"prompts = [\n",
" \"What's the weather in 05751?\",\n",
" \"I'm planning a trip to Killington, Vermont (05751) from Hoboken, NJ (07030). Can you get me weather for both locations and directions?\",\n",
" \"What's the current exchange rate for USD to EUR?\"\n",
"]\n",
"\n",
"for prompt in prompts:\n",
" completion = generate_hermes(prompt)\n",
" functions = extract_function_calls(completion)\n",
"\n",
" if functions:\n",
" for function in functions:\n",
" print(function[\"name\"])\n",
" print(function[\"arguments\"])\n",
" else:\n",
" print(completion.strip())\n",
"\n",
" print(\"=\"*100)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.12"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
from langchain_ollama.chat_models import ChatOllama
from langchain_core.tools import tool
from typing import Any, Dict, Optional, TypedDict
from langchain_core.runnables import RunnableConfig
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import render_text_description
model=ChatOllama(model='gemma2:2b',temperature=0.3)
@tool
def multiply(x: float, y: float) -> float:
"""Multiply two numbers together."""
return x * y
@tool
def add(x: int, y: int) -> int:
"Add two numbers."
return x + y
class ToolCallRequest(TypedDict):
"""A typed dict that shows the inputs into the invoke_tool function."""
name: str
arguments: Dict[str, Any]
def invoke_tool(tool_call_request: ToolCallRequest, config: Optional[RunnableConfig] = None):
tool_name_to_tool = {tool.name: tool for tool in tools}
name = tool_call_request["name"]
requested_tool = tool_name_to_tool[name]
return requested_tool.invoke(tool_call_request["arguments"], config=config)
tools = [multiply, add]
rendered_tools = render_text_description(tools)
system_prompt = f"""\
You are an assistant that has access to the following set of tools.
Here are the names and descriptions for each tool:
{rendered_tools}
Given the user input, return the name and input of the tool to use.
Return your response as a JSON blob with 'name' and 'arguments' keys.
The `arguments` should be a dictionary, with keys corresponding
to the argument names and the values corresponding to the requested values.
"""
prompt = ChatPromptTemplate.from_messages(
[("system", system_prompt), ("user", "{input}")]
)
chain = prompt | model | JsonOutputParser() | invoke_tool
response=chain.invoke({"input": "what's thirteen times 4"})
response
Display the source blob
Display the rendered blob
Raw
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment