提示工程的6种实践体会

system_prompt ="""
You are a helpful data exploration expert specializing in providing data exploration code snippets using Python.

INSTRUCTIONS:
1. Generate concise and precise EDA scripts without additional explanations or textual prefixes.
2. Default to relevant examples in the EDA document, but do not reject reasonable requests.
3. If you are not sure of the answer, you still must give EDA scripts, but you can include a brief warning within the code comments.
"""

defbuild_dynamic_prompt(history):
    dialog =[]
for msg in history:
if msg["role"]=="user":
            dialog.append(f"User: {msg['content']}")
elif msg["role"]=="assistant":
            dialog.append(f"Assistant: {msg['content']}")
    full_prompt ="\n".join(dialog)+"\nAssistant:"
return full_prompt

import os
import sys
from google import genai
import vertexai

from vertexai.generative_models import GenerativeModel, Tool
from vertexai.preview import rag
from google.genai.types import GenerateContentConfig, Retrieval, Tool, VertexRagStore
import torch

# -------------------------------------------------------------------
# Project/Environment Setup
# -------------------------------------------------------------------
PROJECT_ID = "#####"
LOCATION = os.environ.get("GOOGLE_CLOUD_REGION", "us-central1")
vertexai.init(project=PROJECT_ID, location=LOCATION)
client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)

EMBEDDING_MODEL = "publishers/google/models/text-embedding-004"

CORPUS_NAME = "projects/########"

from vertexai.generative_models import GenerativeModel, Tool
from vertexai.preview import rag

# -------------------------------------------------------------------
# RAG Retrieval Tool
# -------------------------------------------------------------------
rag_retrieval_tool = Tool.from_retrieval(
    retrieval=rag.Retrieval(
        source=rag.VertexRagStore(
            rag_resources=[rag.RagResource(rag_corpus=CORPUS_NAME)],
            similarity_top_k=1,
            vector_distance_threshold=0.3
        )
    )
)

# -------------------------------------------------------------------
# System (Base) Instruction
# -------------------------------------------------------------------
system_prompt = """
You are a helpful data exploration code assistant specializing in providing EDA snippets using Python.

INSTRUCTIONS:
1. Generate concise and precise EDA scripts without additional explanations or textual prefixes.
2. Default to relevant examples in the EDA document, but do not reject reasonable requests.
3. If you are not sure of the answer, you still must give EDA scripts, but you can include a brief warning within the code comments.
"""

# -------------------------------------------------------------------
# Initialize the Gemini Model with RAG
# -------------------------------------------------------------------
rag_model = GenerativeModel(
    model_name="gemini-2.0-flash-001",
    tools=[rag_retrieval_tool],
    system_instruction=system_prompt
)

# -------------------------------------------------------------------
# Dynamic Prompt Management
# -------------------------------------------------------------------
conversation_history = []

def build_dynamic_prompt(history):
    """
    Construct a dynamic prompt string from conversation history.
    The system instruction is already set within the model initialization.
    We only compile user/assistant pairs here.
    """
    dialog = []
    for msg in history:
        if msg["role"] == "user":
            dialog.append(f"User: {msg['content']}")
        elif msg["role"] == "assistant":
            dialog.append(f"Assistant: {msg['content']}")
    # Append an 'Assistant:' prompt for the new response
    full_prompt = "\n".join(dialog) + "\nAssistant:"
    return full_prompt



# -------------------------------------------------------------------
# Generate Code Snippet
# -------------------------------------------------------------------
def get_code_snippet(full_prompt: str) -> str:
    """
    Passes the constructed prompt to the model for generation.
    Returns the model's text response (cleaned as needed).
    """
    try:
        if not full_prompt or not isinstance(full_prompt, str):
            raise ValueError("Prompt must be a non-empty string")
        response = rag_model.generate_content(full_prompt)
        if not response or not response.text:
            raise ValueError("No valid response received from the model")
        # Simple cleaning example
        cleaned = response.text.replace('attr">', '').replace('Attr">', '').replace('attr >', '')
        return cleaned.strip()
    except Exception as e:
        return f" Error: {str(e)}"

# -------------------------------------------------------------------
# Interactive Conversation Logic
# -------------------------------------------------------------------
def interactive_mode():
    """
    Continuously prompt the user for input, update conversation history,
    generate a new dynamic prompt, and output the model's response.
    The loop ends if the user types 'exit' or 'quit'.
    """
    print("Entering interactive dynamic-prompt mode. Type 'exit' or 'quit' to stop.\n")
    while True:
        user_input = input("Your request: ").strip()
        if user_input.lower() in ["exit", "quit"]:
            print("Exiting interactive mode.")
            break

        # Store user's message in conversation history
        conversation_history.append({"role": "user", "content": user_input})

        # Build a dynamic prompt based on entire conversation
        prompt = build_dynamic_prompt(conversation_history)

        # Get the model's response
        assistant_reply = get_code_snippet(prompt)

        # Print the model's output
        print(f"\nAssistant:\n{assistant_reply}\n")

        # Add the assistant's reply back to the conversation history
        conversation_history.append({"role": "assistant", "content": assistant_reply})

# -------------------------------------------------------------------
# Main
# -------------------------------------------------------------------
def main():
    if len(sys.argv) > 1:
        user_prompt = " ".join(sys.argv[1:]).strip()
        conversation_history.append({"role": "user", "content": user_prompt})                             
        prompt = build_dynamic_prompt(conversation_history)
        assistant_reply = get_code_snippet(prompt)
        print(assistant_reply)
    else:
        interactive_mode()

if __name__ == "__main__":
    main()

# 传统方式（无 DSPy）
docs = rag_retrieve(user_question)# 步骤1：获取相关文档
prompt = f"{docs}\nUser: {user_question}\nAssistant:"# 步骤2：手动拼接提示词
response = gemini.generate_content(prompt).text             # 步骤3：调用模型生成回答
print(response)# 步骤4：输出结果

classQA(Signature):
    question: str = InputField(desc="RAG 上下文 + 用户问题")
    answer:   str = OutputField(desc="模型返回的最终答案")

lm = dspy.LM("vertex_ai/gemini-2.0-flash-001", project="your-project-id", location="us-central1")
dspy.configure(lm=lm)

qa = Predict(QA, adapter="text")

# 步骤 A：检索上下文
docs = rag_retrieve(user_question)

# 步骤 B：构建动态提示词（仍可保留部分灵活性）
full_prompt = f"{docs}\nUser: {user_question}\nAssistant:"

# 步骤 C：调用 DSPy 模块，获得结构化输出
result = qa(question=full_prompt)
print("Answer:", result.answer)

import os
import dspy
from dspy import InputField, OutputField, Signature, Predict

# -------------------------------------------------------------------
# 1. Configure DSPy to use Vertex AI’s Gemini
# -------------------------------------------------------------------
lm = dspy.LM(
    "vertex_ai/gemini-2.0-flash-001",
    project="###########",
    location="us-central1"
)
dspy.configure(lm=lm)

# -------------------------------------------------------------------
# 2. Define your structured prompt signature
# -------------------------------------------------------------------
class GenerateCode(Signature):
    """Structured Prompt for Generating data exploration Code"""
    input: str = InputField(desc="Full chat context + user request")
    response: str = OutputField(desc="Generated EDA snippet")

# -------------------------------------------------------------------
# 3. Create the DSPy prediction module
# -------------------------------------------------------------------
generate_code = Predict(signature=GenerateCode)

# -------------------------------------------------------------------
# 4. Interactive, context-aware loop
# -------------------------------------------------------------------
conversation_history = []

def interactive_dspy():
    print("DSPy Interactive Mode (type 'exit' or 'quit' to stop)\n")
    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("exit", "quit"):
            print("Goodbye!")
            break

        # append user turn
        conversation_history.append({"role": "user", "content": user_input})

        # build prompt with full history
        dynamic_input = "\n".join(
            f"{turn['role'].capitalize()}: {turn['content']}"
            for turn in conversation_history
        ) + "\nAssistant:"

        # call DSPy
        result = generate_code(input=dynamic_input)

        # show assistant response and append to history
        print(f"\n🤖 Assistant:\n{result.response}\n")
        conversation_history.append({"role": "assistant", "content": result.response})

if __name__ == "__main__":
    interactive_dspy()

classGenerateCode(Signature):
    input: str = InputField(desc="用户指令与上下文")
    output: str = OutputField(desc="生成的代码片段")

generate_code = Predict(GenerateCode, adapter="text")

dspy.configure(lm=dspy.LM("openai/gpt-4"))

{conversation_history}
Assistant:

# 示例代码片段
retrieved_docs = rag_retrieval_tool.retrieve(user_question)
conversation_history = format_conversation_history(history)

prompt = pm.prompt.format(conversation_history=conversation_history, context=retrieved_docs)
response = gemini.generate(prompt)

import os
from tqdm.autonotebook import tqdm
from dynamic_prompting.llms.prompt import PromptManagement
from dynamic_prompting.llms.config import PromptConfig
from vertexai.generative_models import GenerativeModel, Tool
from vertexai.preview import rag

# -------------------------------------------------------------------
# 1) RAG Retrieval Tool
# -------------------------------------------------------------------
CORPUS_NAME = "####"

retrieval = rag.Retrieval(
    source=rag.VertexRagStore(
        rag_resources=[rag.RagResource(rag_corpus=CORPUS_NAME)],
        similarity_top_k=1,
        vector_distance_threshold=0.3
    )
)
rag_tool = Tool.from_retrieval(retrieval=retrieval)

# -------------------------------------------------------------------
# 2) Initialize Gemini Model with Static (Customized) Prompt
# -------------------------------------------------------------------
system_instruction = """
You are a helpful EDA expert specializing in data analysis.
INSTRUCTIONS:
- Provide concise, runnable Python EDA snippets.
- Default to examples in the documentation.
- If uncertain, still return code with a brief warning comment.
  you must give an answer
"""
gemini_model = GenerativeModel(
    model_name="gemini-2.0-flash-001",
    tools=[rag_tool],
    system_instruction=system_instruction
)

# -------------------------------------------------------------------
# 3) Set Up PromptManagement for Dynamic Prompting
# -------------------------------------------------------------------
# Define a template that injects the evolving conversation history
prompt_cfg = PromptConfig(
    prompt="""
{conversation_history}
Assistant:"""
)
pm = PromptManagement(prompt_cfg)

conversation_history = []

def get_dynamic_prompt():
    """
    Renders the prompt by formatting the conversation history
    into the static template managed by PromptManagement.
    """
    history = "\n".join(
        f"User: {turn['content']}" if turn["role"] == "user"
        else f"Assistant: {turn['content']}"
        for turn in conversation_history
    )
    return pm.prompt.format(conversation_history=history)

# -------------------------------------------------------------------
# 4) Generate Code Snippet
# -------------------------------------------------------------------
def generate_code(prompt_text: str) -> str:
    response = gemini_model.generate_content(prompt_text)
    return response.text.strip()

# -------------------------------------------------------------------
# 5) Interactive Loop
# -------------------------------------------------------------------
def interactive_mode():
    print("Entering dynamic-prompting mode. Type 'exit' to quit.")
    while True:
        user_input = input("Your request: ").strip()
        if user_input.lower() in ("exit", "quit"):
            break

        # 5a) Append user turn
        conversation_history.append({"role": "user", "content": user_input})

        # 5b) Build and render dynamic prompt
        dynamic_prompt = get_dynamic_prompt()

        # 5c) Generate and display the code snippet
        snippet = generate_code(dynamic_prompt)
        print(f"\nAssistant:\n{snippet}\n")

        # 5d) Append assistant turn
        conversation_history.append({"role": "assistant", "content": snippet})

if __name__ == "__main__":
    interactive_mode()

# 手动拼接提示词示例
prompt =(
    system_instruction +"\n\n"
"-- Retrieved Context --\n"+ snippet +"\n\n"
"-- Conversation History --\n"
+"\n".join(f"{r['role']}: {r['content']}"for r in history)
+"\n\nAssistant:"
)
response = model.generate_content(prompt)

{{ system_instruction }}

-- Retrieved Context --
{{ retrieval_snippet }}

-- Conversation History --
{% for turn in history %}
{{ turn.role }}: {{ turn.content }}
{% endfor %}

Assistant:

tpl = jinja2.Template(open("prompt_template.txt").read())
prompt_text = tpl.render(
    system_instruction=system_instruction,
    retrieval_snippet=snippet,
    history=conversation_history
)
response = model.generate_content(prompt_text)

import jinja2
from vertexai.generative_models import GenerativeModel, Tool
from vertexai.preview import rag

# -------------------------------------------------------------------
# 1) RAG Retrieval Tool
# -------------------------------------------------------------------
CORPUS_NAME = "######"
retrieval = rag.Retrieval(
    source=rag.VertexRagStore(
        rag_resources=[rag.RagResource(rag_corpus=CORPUS_NAME)],
        similarity_top_k=1,
        vector_distance_threshold=0.3,
    )
)
rag_tool = Tool.from_retrieval(retrieval=retrieval)

# -------------------------------------------------------------------
# 2) Customized Prompt: static system instruction
# -------------------------------------------------------------------

system_instruction = """
You are a helpful data exploration expert specializing in efficient code generation.

INSTRUCTIONS:
1. Generate concise and precise EDA scripts without additional explanations or prefatory text.
2. Default to relevant examples from the knowledge base, but do not refuse reasonable requests.
3. If you are uncertain of the answer, still provide working EDA scripts, and include a brief warning in the code comments.
"""

gemini_model = GenerativeModel(
    model_name="gemini-2.0-flash-001",
    tools=[rag_tool],                      # RAG tool is attached here
    system_instruction=system_instruction.strip(),
)

# -------------------------------------------------------------------
# 3) Jinja2 Template
# -------------------------------------------------------------------
template_source = """
{{ system_instruction }}

-- Conversation History --
{% for turn in history %}
{{ turn.role }}: {{ turn.content }}
{% endfor %}

Assistant:
"""
prompt_template = jinja2.Template(template_source)

# -------------------------------------------------------------------
# 4) Build prompt (no manual retrieval)
# -------------------------------------------------------------------
conversation_history: list[dict[str,str]] = []

def build_prompt() -> str:
    return prompt_template.render(
        system_instruction=system_instruction,
        history=conversation_history,
    )

# -------------------------------------------------------------------
# 5) Interactive loop
# -------------------------------------------------------------------
def interactive_mode():
    print(" Entering Jinja2-powered mode (type 'exit'):")
    while True:
        user_input = input(" Your request: ").strip()
        if user_input.lower() in ("exit", "quit"):
            break

        # 5a) Record user turn
        conversation_history.append({"role":"User","content":user_input})

        # 5b) Build the combined prompt
        prompt_text = build_prompt()

        # 5c) Let Gemini auto‐invoke RAG and generate code
        response = gemini_model.generate_content(prompt_text).text.strip()
        print(f"\n Assistant:\n{response}\n")

        # 5d) Record assistant turn
        conversation_history.append({"role":"Assistant","content":response})

if __name__ == "__main__":
    interactive_mode()

You are a data exploration expert…

-- Retrieved Context --
<most relevant snippet from corpus>

-- Conversation History --
User: First question  
Assistant: First answer  
User: Second question  

Assistant:

import os
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents import initialize_agent, Tool
from langchain.memory import ConversationBufferMemory
from vertexai.preview import rag

# ── 1) Vertex AI RAG Retrieval Setup ────────────────────────────────────────
CORPUS = "#####"
retrieval = rag.Retrieval(
    source=rag.VertexRagStore(
        rag_resources=[rag.RagResource(rag_corpus=CORPUS)],
        similarity_top_k=1,
        vector_distance_threshold=0.3,
    )
)

def rag_retriever(query: str) -> str:
    # NOTE: use `.search()`, not `.retrieve()`
    hits = retrieval.search(query)
    return hits[0].text if hits else ""

rag_tool = Tool(
    name="RAGRetriever",
    func=rag_retriever,
    description="Fetch relevant data‐exploration snippets from the knowledge base.",
)

# ── 2) LangChain LLM via Google GenAI (Gemini) ─────────────────────────────
from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash-001",
    api_key=os.getenv("GOOGLE_API_KEY"),  # ← use `api_key`, not `google_api_key`
    temperature=0.0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
)

# ── 3) Agent + Memory Configuration ────────────────────────────────────────
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

agent = initialize_agent(
    tools=[rag_tool],
    llm=llm,
    agent="zero-shot-react-description",
    memory=memory,
    verbose=False,
)

# ── 4) Interactive Loop ────────────────────────────────────────────────────
def interactive_mode():
    print("LangChain + Gemini + RAG (type 'exit' to quit)")
    while True:
        user_input = input("📝 Your request: ").strip()
        if user_input.lower() in ("exit", "quit"):
            break

        # LangChain agent will:
        #   1) call rag_retriever(user_input) to fetch context
        #   2) build a prompt with system + history + retrieved snippet
        #   3) invoke Gemini under the hood
        response = agent.run(user_input)

        print(f"\n🤖 Assistant:\n{response}\n")

if __name__ == "__main__":
    interactive_mode()