如何提取并查看发送至OpenAI API的Prompt?多框架适配方案
解决方案:提取发送至OpenAI API的Prompt
一、针对Langchain框架的专属方法
利用Langchain原生的回调机制和属性,无需修改底层库或日志配置:
自定义回调Handler捕获Prompt
通过实现BaseCallbackHandler,在LLM启动阶段直接获取原始Prompt:from langchain.callbacks.base import BaseCallbackHandler class PromptLogger(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): print("[LANGCHAIN PROMPT]") for idx, prompt in enumerate(prompts): print(f"Prompt {idx+1}:\n{prompt}\n") # 初始化LLM时绑定回调 from langchain.llms import OpenAI llm = OpenAI(temperature=0.7, callbacks=[PromptLogger()])直接访问Chain的Prompt内容
对于复杂Chain,可通过属性或格式化方法查看填充后的完整Prompt:# 查看Prompt模板 print("Prompt Template:", chain.prompt.template) # 查看填充后的最终Prompt print("Formatted Prompt:", chain.format(inputs={"question": "你的问题"}))
二、通用方法(适配所有基于OpenAI Python库的应用)
这些方法不依赖上层框架,直接针对OpenAI库或HTTP请求层拦截:
1. 猴子补丁(Monkey Patch)OpenAI请求方法
创建独立补丁文件patch_openai.py,无需修改原应用代码:
import openai from openai.api_requestor import APIRequestor original_request = APIRequestor.request def patched_request(self, method, url, params=None, **kwargs): # 捕获并打印Prompt if params and ("prompt" in params or "messages" in params): print("[OPENAI PROMPT]") if "prompt" in params: prompt = params["prompt"] if isinstance(prompt, list): for idx, p in enumerate(prompt): print(f"Prompt {idx+1}:\n{p}\n") else: print(prompt) elif "messages" in params: print("\n".join([f"{m['role']}: {m['content']}" for m in params["messages"]])) # 调用原始请求方法 return original_request(self, method, url, params, **kwargs) # 应用补丁 APIRequestor.request = patched_request
启动应用时加载补丁:
python -c "import patch_openai; exec(open('my_app.py').read())"
2. 用mitmproxy拦截HTTP请求
无需修改任何代码,通过代理捕获所有OpenAI API请求:
- 安装mitmproxy:
pip install mitmproxy - 启动带Web界面的代理:
mitmweb - 设置应用代理并启动:
在mitmweb界面筛选export HTTP_PROXY=http://localhost:8080 export HTTPS_PROXY=http://localhost:8080 python my_app.pyapi.openai.com的请求,查看请求体中的prompt或messages字段即可。
3. 外部控制OpenAI日志级别
避免修改原应用代码,通过外部配置单独开启OpenAI的DEBUG日志:
创建logging_config.py:
import logging # 仅设置openai库的日志级别为DEBUG openai_logger = logging.getLogger("openai") openai_logger.setLevel(logging.DEBUG) # 配置日志格式,只保留关键数据 handler = logging.StreamHandler() handler.setFormatter(logging.Formatter("%(message)s")) openai_logger.addHandler(handler) openai_logger.propagate = False # 避免日志扩散到其他库
启动应用时加载配置并解析日志:
python -c "import logging_config; exec(open('my_app.py').read())" | python extract.py
三、对现有日志方法的改进建议
针对你当前方案的不足,可做以下优化:
移除应用内日志修改:
删掉原应用中的logging.basicConfig()代码,改用上述logging_config.py外部控制日志,避免影响其他库的日志输出。简化启动流程:
编写shell脚本run_app.sh,统一启动命令:#!/bin/bash python -c "import logging_config; exec(open('my_app.py').read())" | python extract.py直接运行
./run_app.sh即可。优化日志解析脚本:
支持ChatCompletion的messages字段,处理转义字符问题:import sys import re import json def extract_prompt(line: str): match = re.search(r"data='(.*?)'", line) if not match: return None # 处理日志中转义的单引号 data_string = match.group(1).replace("\\'", "'") try: data = json.loads(data_string) if "prompt" in data: return data["prompt"] elif "messages" in data: return "\n".join([f"{m['role']}: {m['content']}" for m in data["messages"]]) except json.JSONDecodeError: return None return None for line in sys.stdin: prompt = extract_prompt(line) if prompt: print(f'[PROMPT] {prompt}\n')
内容的提问来源于stack exchange,提问作者washingweb
相关产品推荐
相关产品推荐

