一??Function calling 函數調用

from dotenv import load_dotenv, find_dotenvload_dotenv(find_dotenv())from openai import OpenAI
import jsonclient = OpenAI()# Example dummy function hard coded to return the same weather
# In production, this could be your backend API or an external API
def get_current_weather(location, unit="fahrenheit"):?"""Get the current weather in a given location"""if "tokyo" in location.lower():return json.dumps({"location": "Tokyo", "temperature": "10", "unit": unit})elif "san francisco" in location.lower():return json.dumps({"location": "San Francisco", "temperature": "72", "unit": unit})elif "paris" in location.lower():return json.dumps({"location": "Paris", "temperature": "22", "unit": unit})else:return json.dumps({"location": location, "temperature": "unknown"})def run_conversation():# Step 1: send the conversation and available functions to the modelmessages = [{"role": "user", "content": "What's the weather like in San Francisco, Tokyo, and Paris?"}]tools = [{"type": "function","function": {"name": "get_current_weather","description": "Get the current weather in a given location","parameters": {"type": "object","properties": {"location": {"type": "string","description": "The city and state, e.g. San Francisco, CA",},"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},},"required": ["location"],},},}]response = client.chat.completions.create(model="gpt-3.5-turbo-0125",messages=messages,tools=tools,tool_choice="auto",  # auto is default, but we'll be explicit)print(response)response_message = response.choices[0].messageprint(response_message)tool_calls = response_message.tool_callsprint(tool_calls)# Step 2: check if the model wanted to call a functionif tool_calls:# Step 3: call the function# Note: the JSON response may not always be valid; be sure to handle errorsavailable_functions = {"get_current_weather": get_current_weather,}  # only one function in this example, but you can have multiplemessages.append(response_message)  # extend conversation with assistant's reply# Step 4: send the info for each function call and function response to the modelfor tool_call in tool_calls:function_name = tool_call.function.namefunction_to_call = available_functions[function_name]function_args = json.loads(tool_call.function.arguments)function_response = function_to_call(location=function_args.get("location"),unit=function_args.get("unit"),)messages.append({"tool_call_id": tool_call.id,"role": "tool","name": function_name,"content": function_response,})  # extend conversation with function responsesecond_response = client.chat.completions.create(model="gpt-3.5-turbo-0125",messages=messages,)  # get a new response from the model where it can see the function responsereturn second_response

result = run_conversation()
resultresult.choices[0].message.content
# 'The current weather in San Francisco is 72°F, in Tokyo it is 10°C, and in Paris it is 22°C.'

二?v3-Create-Custom-Agent

2.1?Load the LLM 加載LLM

from dotenv import load_dotenv, find_dotenvload_dotenv(find_dotenv())from langchain_openai import ChatOpenAIllm = ChatOpenAI(model="gpt-4", temperature=0)

2.2?Define Tools 定義工具

from langchain.agents import toolimport time@tool
def generate_unique_timestamp():"""生成唯一的時間戳。輸入始終為空字符串。Returns:int: 唯一的時間戳，以毫秒為單位。"""timestamp = int(time.time() * 1000)  # 獲取當前時間的毫秒級時間戳return timestampimport os@tool
def create_folder(folder_name):"""根據給定的文件夾名創建文件夾。Args:folder_name (str): 要創建的文件夾的名稱。Returns:str: 創建的文件夾的路徑。"""try:os.makedirs(os.path.join("chat_history", folder_name))  # 創建文件夾return os.path.abspath(folder_name)  # 返回創建的文件夾的絕對路徑except OSError as e:print(f"創建文件夾失敗：{e}")return Noneimport shutil@tool
def delete_temp_folder():"""刪除 chat_history 文件夾下的 temp 文件夾。輸入始終為空字符串。Returns:bool: 如果成功刪除則返回 True，否則返回 False。"""temp_folder = "chat_history/temp"  # temp 文件夾路徑try:shutil.rmtree(temp_folder)  # 遞歸刪除 temp 文件夾及其所有內容print("成功刪除 temp 文件夾。")return Trueexcept Exception as e:print(f"刪除 temp 文件夾失敗：{e}")return False@tool
def copy_chat_history(interview_id: str) -> str:"""將 chat_history/temp 文件夾中的 chat_history.txt 文件復制到 chat_history 文件夾下的以 interview_id 命名的子文件夾中。如果面試ID文件夾不存在，則返回相應的提示字符串。參數:interview_id (str): 面試的唯一標識符。返回:str: 操作結果的提示信息。"""# 確定臨時文件夾和面試文件夾路徑temp_folder = os.path.join("chat_history", "temp")interview_folder = os.path.join("chat_history", interview_id)# 檢查面試文件夾是否存在if not os.path.exists(interview_folder):return f"面試ID為 {interview_id} 的文件夾不存在。無法完成復制操作。"# 將 chat_history.txt 從臨時文件夾復制到面試文件夾source_file = os.path.join(temp_folder, 'chat_history.txt')destination_file = os.path.join(interview_folder, 'chat_history.txt')shutil.copyfile(source_file, destination_file)return f"已將 chat_history.txt 復制到面試ID為 {interview_id} 的文件夾中。"@tool
def read_chat_history(interview_id: str) -> str:"""讀取指定面試ID文件夾下的聊天記錄(chat_history.txt)內容。參數:interview_id (str): 面試的唯一標識符。返回:str: 聊天記錄的內容。"""# 確定面試文件夾路徑interview_folder = os.path.join("chat_history", interview_id)# 檢查面試文件夾是否存在if not os.path.exists(interview_folder):return f"面試ID為 {interview_id} 的文件夾不存在。無法讀取聊天記錄。"# 讀取聊天記錄文件內容chat_history_file = os.path.join(interview_folder, 'chat_history.txt')with open(chat_history_file, 'r', encoding='utf-8') as file:chat_history_content = file.read()return chat_history_content@tool
def generate_markdown_file(interview_id: str, interview_feedback: str) -> str:"""將給定的面試反饋內容生成為 Markdown 文件，并保存到指定的面試ID文件夾中。參數:interview_id (str): 面試的唯一標識符。interview_feedback (str): 面試反饋的內容。返回:str: 操作結果的提示信息。"""# 確定面試文件夾路徑interview_folder = os.path.join("chat_history", interview_id)# 檢查面試文件夾是否存在if not os.path.exists(interview_folder):return f"面試ID為 {interview_id} 的文件夾不存在。無法生成 Markdown 文件。"# 生成 Markdown 文件路徑markdown_file_path = os.path.join(interview_folder, "面試報告.md")try:# 寫入 Markdown 文件with open(markdown_file_path, 'w', encoding='utf-8') as file:# 寫入標題和面試反饋file.write("# 面試報告\n\n")file.write("## 面試反饋：\n\n")file.write(interview_feedback)file.write("\n\n")# 讀取 chat_history.txt 文件內容并寫入 Markdown 文件chat_history_file_path = os.path.join(interview_folder, "chat_history.txt")if os.path.exists(chat_history_file_path):file.write("## 面試記錄：\n\n")with open(chat_history_file_path, 'r', encoding='utf-8') as chat_file:for line in chat_file:file.write(line.rstrip('\n') + '\n\n')  # 添加換行符return f"已生成 Markdown 文件: {markdown_file_path}"except Exception as e:return f"生成 Markdown 文件時出錯: {str(e)}"

2.3 執行

from utils import parse_cv_to_md
cv_file_path = "data/cv.txt"
result = parse_cv_to_md(llm, cv_file_path)
print(result)

2.4 加載數據

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from utils import parse_md_file_to_docsfile_path = "data/cv.md"
docs = parse_md_file_to_docs(file_path)
print(len(docs))vectorstore = Chroma.from_documents(documents=docs, embedding=OpenAIEmbeddings())retriever = vectorstore.as_retriever(search_kwargs={"k": 1})retriever.get_relevant_documents("langchain")

retriever.get_relevant_documents("python")@tool
def find_most_relevant_block_from_cv(sentence: str) -> str:"""當你需要根據職位描述（JD）中的技能關鍵詞去簡歷文本中找到相關內容時，就可以調用這個函數。參數:sentence (str): 包含技能關鍵詞的句子。返回:str: 最相關的文本塊。"""try:most_relevant_docs = retriever.get_relevant_documents(sentence)print(len(most_relevant_docs))if most_relevant_docs:most_relevant_texts = [doc.page_content for doc in most_relevant_docs]most_relevant_text = "\n".join(most_relevant_texts)return most_relevant_textelse:return "未找到相關文本塊"except Exception as e:print(f"find_most_relevant_block_from_cv()發生錯誤：{e}")return "函數發生錯誤，未找到相關文本塊"tools = [generate_unique_timestamp, create_folder, copy_chat_history, read_chat_history, generate_markdown_file, find_most_relevant_block_from_cv]

2.5?Create Prompt 創建提示

from utils import *jd_file_path = "data/jd.txt"
jd_json_file_path = parse_jd_to_json(llm, jd_file_path)
jd_json_file_path

jd_json_file_path = "data/jd.json"
jd_dict = read_json(jd_json_file_path)
jd_dict

job_title = jd_dict.get('基本信息').get('職位')
job_key_skills = jd_dict.get('專業技能/知識/能力')
print(f"職位：{job_title}")
print(f"專業技能/知識/能力：{job_key_skills}")"""職位：Python工程師 (AI應用方向)
專業技能/知識/能力：['Python', 'PyTorch', 'TensorFlow', 'Numpy', 'Redis', 'MySQL', 'openai', 'langchain', 'AI開發工具集', '向量數據庫', 'Text-SQL', 'Python Web框架', 'FastAPI', 'ORM框架', 'Prompt提示詞']"""

# 棄用
system_prompt = f"""
## Role and Goals
- 你是所招崗位“{job_title}”的技術專家，同時也作為技術面試官向求職者提出技術問題，專注于考察應聘者的專業技能、知識和能力。
- 這里是當前崗位所需的專業技能、知識和能力：“{job_key_skills}”，你應該重點圍繞這些技術點提出你的問題。
- 你嚴格遵守面試流程進行面試。## Interview Workflow
1. 當應聘者說開始面試后，1.1 你要依據當前時間生成一個新的時間戳作為面試ID（只會在面試開始的時候生成面試ID，其他任何時間都不會）1.2 以該面試ID為文件夾名創建本地文件夾（只會在面試開始的時候創建以面試ID為名的文件夾，其他任何時間都不會）1.3 刪除存儲聊天記錄的臨時文件夾1.4 輸出該面試ID給應聘者，并結合當前技術點、與技術點相關的簡歷內容，提出你的第一個基礎技術問題。
2. 接收應聘者的回答后，2.1 檢查應聘者的回答是否有效2.1.1 如果是對面試官問題的正常回答（無論回答的好不好，還是回答不會，都算正常回答），就跳轉到2.2處理2.1.2 如果是與面試官問題無關的回答（胡言亂語、辱罵等），請警告求職者需要嚴肅對待面試，跳過2.2，再次向求職者提出上次的問題。2.2 如果應聘者對上一個問題回答的很好，就基于當前技術點和歷史記錄提出一個更深入一點的問題；如果應聘者對上一個問題回答的一般，就基于當前技術點和歷史記錄提出另一個角度的問題；如果應聘者對上一個問題回答的不好，就基于當前技術點和歷史記錄提出一個更簡單一點的問題；如果應聘者對上一個問題表示不會、不懂、一點也回答不了，就換一個與當前技術點不同的技術點進行技術提問。
3. 當應聘者想結束面試或當應聘者想要面試報告，3.1 從臨時文件夾里復制一份聊天記錄文件到當前面試ID文件夾下。3.2 讀取當前面試ID文件夾下的聊天記錄，基于聊天記錄、從多個角度評估應聘者的表現、生成一個詳細的面試報告。3.3 調用工具生成一個面試報告的markdown文件到當前面試ID文件夾下3.4 告知應聘者面試已結束，以及面試報告的位置。## Output Constraints
- 你發送給應聘者的信息中，一定不要解答你提出的面試問題，只需要有簡短的反饋和提出的新問題。
- 你每次提出的技術問題，都需要結合從JD里提取的技術點和與技術點相關的簡歷內容，當你需要獲取`與技術點相關的簡歷內容`時，請調用工具。
- 再一次檢查你的輸出，你一次只會問一個技術問題。
"""

system_prompt = f"""
## Role and Goals
- 你是所招崗位“{job_title}”的技術專家，同時也作為技術面試官向求職者提出技術問題，專注于考察應聘者的專業技能、知識和能力。
- 這里是當前崗位所需的專業技能、知識和能力：“{job_key_skills}”，你應該重點圍繞這些技術點提出你的問題。
- 你嚴格遵守面試流程進行面試。## Interview Workflow
1. 當應聘者說開始面試后，1.1 你要依據當前時間生成一個新的時間戳作為面試ID（只會在面試開始的時候生成面試ID，其他任何時間都不會）1.2 以該面試ID為文件夾名創建本地文件夾（只會在面試開始的時候創建以面試ID為名的文件夾，其他任何時間都不會）1.3 刪除存儲聊天記錄的臨時文件夾1.4 輸出該面試ID給應聘者，并結合當前技術點、與技術點相關的簡歷內容，提出你的第一個基礎技術問題。
2. 接收應聘者的回答后，2.1 檢查應聘者的回答是否有效2.1.1 如果是對面試官問題的正常回答（無論回答的好不好，還是回答不會，都算正常回答），就跳轉到2.2處理2.1.2 如果是與面試官問題無關的回答（胡言亂語、辱罵等），請警告求職者需要嚴肅對待面試，跳過2.2，再次向求職者提出上次的問題。2.2 如果應聘者對上一個問題回答的很好，就基于當前技術點和歷史記錄提出一個更深入一點的問題；如果應聘者對上一個問題回答的一般，就基于當前技術點和歷史記錄提出另一個角度的問題；如果應聘者對上一個問題回答的不好，就基于當前技術點和歷史記錄提出一個更簡單一點的問題；如果應聘者對上一個問題表示不會、不懂、一點也回答不了，就換一個與當前技術點不同的技術點進行技術提問。
3. 當應聘者想結束面試或當應聘者想要面試報告，3.1 從臨時文件夾里復制一份聊天記錄文件到當前面試ID文件夾下。3.2 讀取當前面試ID文件夾下的聊天記錄，基于聊天記錄、從多個角度評估應聘者的表現、生成一個詳細的面試報告。3.3 調用工具生成一個面試報告的markdown文件到當前面試ID文件夾下3.4 告知應聘者面試已結束，以及面試報告的位置。## Output Constraints
- 你發送給應聘者的信息中，一定不要解答你提出的面試問題，只需要有簡短的反饋和提出的新問題。
- 你每次提出的技術問題，都需要結合從JD里提取的技術點和與技術點相關的簡歷內容，當你需要獲取`與技術點相關的簡歷內容`時，請調用工具。
- 再一次檢查你的輸出，你一次只會問一個技術問題。
"""

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderprompt = ChatPromptTemplate.from_messages([("system",system_prompt,),MessagesPlaceholder(variable_name="chat_history"),("user", "{input}"),MessagesPlaceholder(variable_name="agent_scratchpad"),]
)print(prompt.messages[0].prompt.template)

2.6?Bind tools to LLM 將工具綁定到LLM

llm_with_tools = llm.bind_tools(tools)llm_with_tools.kwargs['tools']

2.7?Create the Agent 創建代理

from langchain.agents.format_scratchpad.openai_tools import (format_to_openai_tool_messages,
)
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParseragent = ({"input": lambda x: x["input"],"agent_scratchpad": lambda x: format_to_openai_tool_messages(x["intermediate_steps"]),"chat_history": lambda x: x["chat_history"],}| prompt| llm_with_tools| OpenAIToolsAgentOutputParser()
)

2.8?Run the agent 運行代理

from langchain.agents import AgentExecutoragent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)from utils import save_chat_historyfrom langchain_core.messages import AIMessage, HumanMessagechat_history = []user_input = "開始面試"
print(user_input)while True:result = agent_executor.invoke({"input": user_input, "chat_history": chat_history})print(result['output'])chat_history.extend([HumanMessage(content=user_input),AIMessage(content=result["output"]),])# 存儲聊天記錄到臨時文件夾temp_folder = "chat_history/temp"  # 臨時文件夾名稱os.makedirs(temp_folder, exist_ok=True)  # 創建臨時文件夾，如果不存在則創建save_chat_history(chat_history, temp_folder)# 獲取用戶下一條輸入user_input = input("user: ")# 檢查用戶輸入是否為 "exit"if user_input == "exit":print("用戶輸入了 'exit'，程序已退出。")break

2.9 urls.py

import os
import json
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_core.messages import AIMessage, HumanMessagedef save_chat_history(chat_history, folder_name):"""將聊天記錄存儲到指定的文件夾下的chat_history.txt文件中。Args:chat_history (list): 聊天記錄列表，每個元素是一個AIMessage或HumanMessage對象。folder_name (str): 聊天記錄文件夾的名稱。Returns:str: 保存的文件路徑，如果保存失敗則返回None。"""try:file_path = os.path.join(folder_name, "chat_history.txt")  # chat_history.txt文件路徑with open(file_path, "w", encoding="utf-8") as file:for message in chat_history:if isinstance(message, AIMessage):speaker = "面試官"elif isinstance(message, HumanMessage):speaker = "應聘者"else:continue  # 忽略不是面試官或應聘者的消息file.write(f"{speaker}: {message.content}\n")  # 將每條聊天記錄寫入文件，每條記錄占一行return file_path  # 返回保存的文件路徑except Exception as e:print(f"保存聊天記錄失敗：{e}")return Nonedef parse_jd_to_json(llm, jd_file_path: str):"""將給定的 JD 文件內容解析為 JSON，并存儲到指定路徑下。參數:llm: 大模型。jd_file_path (str): JD 文件的路徑。返回:str: 存儲的 JSON 文件路徑。"""try:with open(jd_file_path, 'r', encoding='utf-8') as jd_file:jd_content = jd_file.read().strip()template = """
基于JD文本，按照約束，生成以下格式的 JSON 數據：
{{"基本信息": {{"職位": "職位名稱","薪資": "薪資范圍","地點": "工作地點","經驗要求": "經驗要求","學歷要求": "學歷要求","其他":""}},"崗位職責": {{"具體職責": ["職責1", "職責2", ...]}},"崗位要求": {{"學歷背景": "學歷要求","工作經驗": "工作經驗要求","技能要求": ["技能1", "技能2", ...],"個人特質": ["特質1", "特質2", ...],}},"專業技能/知識/能力": ["技能1", "技能2", ...],"其他信息": {{}}
}}JD文本：
[{jd_content}]約束：
1、除了`專業技能/知識/能力`鍵，其他鍵的值都從原文中獲取。
2、保證JSON里的值全面覆蓋JD原文，不遺漏任何原文，不知如何分類就放到`其他信息`里。
3、`專業技能/知識/能力`鍵對應的值要求從JD全文中（尤其是崗位職責、技能要求部分）提取總結關鍵詞或關鍵短句，不能有任何遺漏的硬技能。JSON：
"""parser = JsonOutputParser()prompt = PromptTemplate(template=template,input_variables=["jd_content"],partial_variables={"format_instructions": parser.get_format_instructions()},)print(prompt.template)chain = prompt | llm | parserresult = chain.invoke({"jd_content": jd_content})# 打印print(result)print(type(result))print(result['專業技能/知識/能力'])# 存儲到 data 目錄下output_file_path = "data/jd.json"with open(output_file_path, 'w', encoding='utf-8') as output_file:json.dump(result, output_file, ensure_ascii=False)print(f"已存儲最終 JSON 文件到 {output_file_path}")return output_file_pathexcept Exception as e:print(f"解析 JD 文件時出錯: {str(e)}")return Nonedef read_json(file_path: str) -> dict:"""讀取 JSON 文件并返回其內容。參數:file_path (str): JSON 文件的路徑。返回:dict: JSON 文件的內容。"""try:with open(file_path, 'r', encoding='utf-8') as json_file:data = json.load(json_file)return dataexcept Exception as e:print(f"讀取 JSON 文件時出錯: {str(e)}")return {}from langchain.text_splitter import MarkdownHeaderTextSplitter
from langchain.schema import Documentdef parse_cv_to_md(llm, cv_file_path: str):"""將給定的簡歷文件內容解析為 JSON，并存儲到指定路徑下。參數:llm: 大模型。cv_file_path (str): 簡歷文件的路徑。返回:str: 存儲的 Markdown 文件路徑。"""try:with open(cv_file_path, 'r', encoding='utf-8') as cv_file:cv_content = cv_file.read().strip()template = """
基于簡歷文本，按照約束，轉換成Markdown格式：簡歷文本：
[{cv_content}]約束：
1、只用一級標題和二級標題分出來簡歷的大塊和小塊
2、一級標題只有這些：個人信息、教育經歷、工作經歷、項目經歷、校園經歷、職業技能、曾獲獎項、興趣愛好、自我評價、其他信息。Markdown：
"""parser = StrOutputParser()prompt = PromptTemplate(template=template,input_variables=["cv_content"])print(prompt.template)chain = prompt | llm | parserresult = chain.invoke({"cv_content": cv_content})# 打印print(result)print(type(result))# 存儲到 data 目錄下output_file_path = "data/cv.md"with open(output_file_path, 'w', encoding='utf-8') as output_file:output_file.write(result.strip("```"))output_file.write("\n\n")print(f"已存儲最終 Markdown 文件到 {output_file_path}")return output_file_pathexcept Exception as e:print(f"解析 CV 文件時出錯: {str(e)}")return Nonedef parse_md_file_to_docs(file_path):with open(file_path, 'r', encoding='utf-8') as file:markdown_text = file.read()docs = []headers_to_split_on = [("#", "Title 1"),("##", "Title 2"),# ("###", "Title 3"),# ("###", "Title 4")]markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)split_docs = markdown_splitter.split_text(markdown_text)for split_doc in split_docs:metadata = split_doc.metadatatitle_str = f"# {metadata.get('Title 1', 'None')}\n## {metadata.get('Title 2', 'None')}\n"page_content = title_str + split_doc.page_content.strip()doc = Document(page_content=page_content,metadata=metadata)docs.append(doc)return docsif __name__ == "__main__":from dotenv import load_dotenv, find_dotenvload_dotenv(find_dotenv())from langchain_openai import ChatOpenAIllm = ChatOpenAI(model="gpt-4", temperature=0) # gpt-3.5-turbo# # jd# jd_file_path = "data/jd.txt"# result = parse_jd_to_json(llm, jd_file_path)# print(result)# cvcv_file_path = "data/cv.txt"result = parse_cv_to_md(llm, cv_file_path)print(result)