ok, not exactly what I want. I have a specific chain see below
import requests
from langchain.llms.base import LLM
from typing import Optional, List
from sentence_transformers import SentenceTransformer
from langchain_community.vectorstores import FAISS
import faiss
from langchain_core.runnables import RunnableLambda
from langsmith import Client
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.tools import Tool
from langchain.prompts import PromptTemplate
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.agents import initialize_agent, AgentType
from langchain.schema.runnable import RunnableMap, RunnableSequence
import numpy as np
import os
import yaml
from Logger import Logger
import sys
from embed_rag_data import load_texts_from_folder
from HF_API_Wrapper import HuggingFaceAPI
============================================================================
model_url = “https://api-inference.huggingface.co/models/meta-llama/Llama-3.3-70B-Instruct”
api_token = “my_token”
llm = HuggingFaceAPI(api_url=model_url, api_token=api_token)
with open(“config.yaml”, “r”) as file:
config = yaml.safe_load(file)
parameters = config[“parameters”]
prompts = config[“prompts”]
rag_queries = config[“rag”]
core_info_template = prompts[“core_info”][“template”]
core_inputs = {
“species”: parameters[“species”],
“age”: parameters[“age”],
“location”: parameters[“location”]
}
core_info_prompt = PromptTemplate(
input_variables=[“species”, “age”, “location”],
template=core_info_template
)
core_info_chain = core_info_prompt | llm
character_traits_prompt_template = prompts[“character_traits”][“template”]
traits_inputs = {
“temperament”: parameters[“temperament”],
“voice”: parameters[“voice”]
}
character_traits_prompt = PromptTemplate(
input_variables=[“temperament”, “voice”],
template=character_traits_prompt_template
)
character_traits_chain = character_traits_prompt | llm
embedding_model = SentenceTransformer(‘all-MiniLM-L6-v2’)
scientific_index = faiss.read_index(“scientific_index.faiss”)
diary_index = faiss.read_index(“diary_index.faiss”)
master_folder = “RAG_data”
scientific_folder = os.path.join(master_folder, “papers”)
diary_folder = os.path.join(master_folder, “sample_texts”)
scientific_texts = load_texts_from_folder(scientific_folder)
diary_texts = load_texts_from_folder(diary_folder)
def simple_faiss_retrieval(query, index, texts, embedding_model, top_k=5):
"""Retrieve top-k matching texts from a FAISS index."""
query_embedding = embedding_model.encode([query])
D, I = index.search(np.array(query_embedding, dtype=np.float32), k=top_k)
retrieved_texts = [texts[i] for i in I[0]]
return retrieved_texts
def simple_faiss_retrieval_min_max(query, index, texts, embedding_model, min_k=3, max_k=5):
"""Retrieve a range of matching texts from a FAISS index."""
query_embedding = embedding_model.encode([query])
# Get the max_k results (we need at least max_k results to slice properly)
D, I = index.search(np.array(query_embedding, dtype=np.float32), k=max_k)
# Ensure the range is within valid bounds
min_k = max(0, min_k) # Prevent negative indexing
max_k = min(max_k, len(texts)) # Prevent going out of bounds
# Retrieve texts in the given range (from min_k to max_k)
retrieved_texts = [texts[i] for i in I[0][min_k - 1:max_k]] # Adjust indices
def retrieval_task(query, mode=“scientific”, similar_samples=5):
"""Perform retrieval for the given query and mode (scientific or diary)."""
if mode == “scientific”:
return simple_faiss_retrieval(query, scientific_index, scientific_texts, embedding_model, similar_samples)
elif mode == “diary”:
return simple_faiss_retrieval(query, diary_index, diary_texts, embedding_model, similar_samples)
else:
raise ValueError(“Invalid mode! Choose ‘scientific’ or ‘diary’.”)
def rag_query(query, mode=“scientific”, raw_output=False, similar_samples=5):
"""Perform RAG using FAISS for retrieval and Hugging Face API for generation."""
# Retrieve the relevant texts from FAISS
retrieved_texts = retrieval_task(query, mode, similar_samples)
context = “\n”.join(retrieved_texts)
# If raw output is requested, return the retrieved texts directly
f"Context:\n{context}\n\n"
f"Context:\n{context}\n\n"
f"Write a diary entry about: {query}\n\n"
raise ValueError("Invalid mode! Choose 'scientific' or 'diary'.")
# Pass the prompt to the LLM for generation
response = llm.invoke(prompt)
def rag_wrapper(query, mode, raw_output=False, samples=5):
return rag_query(query, mode=mode, raw_output=raw_output, similar_samples=samples)
refined_persona_prompt_template = prompts[“refined_persona”][“template”]
refined_persona_prompt = PromptTemplate(
input_variables=[“persona”, “scientific_context”, “diary_context”],
template=refined_persona_prompt_template
)
human_instruction_prompt_template = prompts[“human_instruction”][“template”]
human_instruction_processor_prompt = PromptTemplate(
input_variables=[“persona”, “instructions”],
template=human_instruction_prompt_template
)
instructed_persona_prompt = PromptTemplate(
input_variables=[“persona”],
template=human_instruction_prompt_template
)
instructed_persona_chain = instructed_persona_prompt | llm
def get_guidelines_input():
return input(“Provide guidelines and rules for article generation: ”)
human_guidelines_prompt = RunnableLambda(
lambda _: {“guidelines”: get_guidelines_input()}
)
expressions = [
“The whispers of the wind carry ancient secrets.”,
“The roots speak in silence, connecting the past and future.”,
“Each fallen leaf tells a story of resilience.”,
“We grow from the darkness and reach towards the sky.”,
“The forest has a collective memory. Nothing is new, although things change”,
“We are wild-wood barometers, our sap is mercury, at root we are optimists.”
]
article_prompt_template = prompts[“article”][“template”]
article_prompt = PromptTemplate(
input_variables=[“persona”, “expressions”, “guidelines”, “topic”, “scientific_context”,“diary_context”],
template=article_prompt_template
)
article_chain = article_prompt | llm
intro_prompt_template = prompts[“introduction”][“template”]
intro_prompt = PromptTemplate(
input_variables=[“persona”, “historic_context”],
template=intro_prompt_template
)
intro_chain = intro_prompt | llm
main_body_prompt_template = prompts[“main_body”][“template”]
main_body_prompt = PromptTemplate(
input_variables=[“persona”, “scientific_context”],
template=main_body_prompt_template
)
main_body_chain = main_body_prompt | llm
conclusion_prompt_template = prompts[“conclusion”][“template”]
conclusion_prompt = PromptTemplate(
input_variables=[“persona”,“introduction”,“main_body”],
template=conclusion_prompt_template
)
conclusion_chain = conclusion_prompt | llm
flavor_prompt_template = prompts[“flavor”][“template”]
flavor_prompt = PromptTemplate(
input_variables=[“persona”,“full_draft”,“diary_context”],
template=flavor_prompt_template
)
flavor_chain = flavor_prompt | llm
sys.stdout = Logger()
persona_chain = (
core_info_chain
| (lambda core_info: {
“core_info”: core_info, # Keep the core info in the output dictionary
“traits_input”: traits_inputs # Prepare inputs for traits
})
| (lambda data: { # Invoke the traits chain with the prepared input
“core_info”: data[“core_info”],
“traits”: character_traits_chain.invoke(data[“traits_input”])
})
| (lambda data: { # Combine core info and traits into a single “persona”
“persona”: f”{data[‘core_info’]}\n{data[‘traits’]}”
})
)
#persona = persona_chain.invoke(core_inputs)
#print(“Persona:”, persona[“persona”])
full_workflow = (
persona_chain
| (lambda data: (
print(“STEP 1 - PERSONA FROM DATA INPUTS:”, data[“persona”]) or { # Debug persona
“persona”: instructed_persona_chain.invoke(data[“persona”])
}
))
| (lambda data: (
print(“STEP 2 - REFINED PERSONAL AFTER HUMAN INPUT:”, data[“persona”]) or { # Debug refined persona
“persona”: data[“persona”],
“historic_context”: rag_wrapper(
query=rag_queries[“historic_query”],
mode=“scientific”,
raw_output=True,
samples=6
) or “No historical context available.”,
}
))
| (lambda data: (
print(
“STEP 4 - RAG RESULTS:\n”
“Historic Context ({} texts):\n{}\n\n”.format(
len(data[‘historic_context’]), # Count of scientific texts
“\n”.join(f”- {text}” for text in data[‘historic_context’])) # Each on a new line
) or { # Debug RAG results
“persona”: data[‘persona’],
“historic_context”: data[“historic_context”], # Pass this forward
}
))
| (lambda data: {
“persona”: data[“persona”],
“introduction”: intro_chain.invoke({
“persona”: data[“persona”],
“historic_context”: “\n”.join(data[“historic_context”]) # Convert list to string if needed
})
})
| (lambda data: (
print(
“STEP 5 - INTRODUCTION:\n”
“Article Intro {}“.format(data[“introduction”])
) or {
“persona”: data[“persona”],
“introduction”: data[“introduction”],
“scientific_context”: rag_wrapper(
query=rag_queries[“scientific_query”],
mode=“scientific”,
raw_output=True,
samples=6
) or “No scientific context available.”,
}
"STEP 6 - RAG RESULTS:\n"
"Scientific Context ({} texts):\n{}\n\n".format(
len(data['scientific_context']), # Count of scientific texts
"\n".join(f"- {text}" for text in data['scientific_context'])) # Each on a new line
) or { # Debug RAG results
"persona": data['persona'],
"scientific_context": data["scientific_context"], # Pass this forward
"introduction": data["introduction"],
"persona": data["persona"],
"introduction": data["introduction"],
"main_body": main_body_chain.invoke({
"persona": data["persona"],
"scientific_context": "\n".join(data["scientific_context"]) # Convert list to string if needed
"Article Main Body {}".format(data["main_body"])
"persona": data["persona"],
"introduction": data["introduction"],
"main_body": data["main_body"],
"full_draft": conclusion_chain.invoke({
"persona": data["persona"],
"introduction": data["introduction"],
"main_body": data["main_body"]
"STEP 8 - FULL ARTICLE DRAFT:\n"
"Article Draft {}".format(data["full_draft"])
"persona": data["persona"],
"full_draft": data["full_draft"],
"diary_context": rag_wrapper(
query=rag_queries["diary_query"],
) or "No diary context available.",
"STEP 9 - RAG RESULTS:\n"
"Diary Context ({} texts):\n{}\n\n".format(
len(data['diary_context']), # Count of scientific texts
"\n".join(f"- {text}" for text in data['diary_context'])) # Each on a new line
) or { # Debug RAG results
"persona": data['persona'],
"full_draft": data["full_draft"], # Pass this forward
"diary_context": data["diary_context"],
"flavored_article": flavor_chain.invoke({
"persona": data["persona"],
"full_draft": data["full_draft"], # Pass this forward
"diary_context": data["diary_context"]
"STEP 10 - FINAL ARTICLE:\n"
"Final Article {}".format(data["flavored_article"])
"persona": data["persona"],
"flavored_article": data["flavored_article"]
# "Step 4 - RAG Results:\n"
# "Scientific Context ({} texts):\n{}\n\n"
# "Diary Context ({} texts):\n{}".format(
# len(data['scientific_context']), # Count of scientific texts
# "\n".join(f"- {text}" for text in data['scientific_context']), # Each on a new line
# len(data['diary_context']), # Count of diary texts
# "\n".join(f"- {text}" for text in data['diary_context'])) # Each on a new line
# ) or { # Debug RAG results
# "persona": f"{data['persona']}\n\nScientific Knowledge:\n{data['scientific_context']}\n\nDiary Insights:\n{data['diary_context']}",
# "scientific_context": data["scientific_context"], # Pass this forward
# "diary_context": data["diary_context"] # Pass this forward
)
full_workflow.invoke(core_inputs)
i want this chain to get executed and i want the user to converse in each step either using the predefined prompt or by altering it. so to go from one step of the chain to the next the user will have to intervene and hit enter to chat with the model. Which by the way is the Llama and not gpt as you can see. I am using huggingface’s serverless API