如何通过LangChain Callbacks记录LLM调用与结果至变量及问题排查

一、「Prompt after formatting」和ANSI编码的来源 #

• 「Prompt after formatting」是LangChain内置的StdOutCallbackHandler输出内容的一部分，大概率是你初始化LLMChain时，除了自定义Handler，还默认加载了这个内置Handler（LangChain默认会自动添加StdOutCallbackHandler，除非显式指定callbacks参数）。
• ANSI绿色编码是StdOutCallbackHandler为了在终端高亮显示内容添加的控制台颜色控制符，多个Handler会共享回调事件，所以这些编码会被你的自定义Handler捕获到日志里。

二、去除方法 #

1. 禁用默认的StdOutCallbackHandler #

初始化LLMChain时，显式指定callbacks参数为你的自定义Handler列表，完全替代默认Handler：

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.callbacks.base import BaseCallbackHandler

class CustomLogHandler(BaseCallbackHandler):
    def __init__(self):
        self.log = []
    
    def on_chain_start(self, serialized, inputs, **kwargs):
        self.log.append(f"Chain started with inputs: {inputs}")
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.log.append(f"LLM prompts: {prompts}")
    
    def on_llm_end(self, response, **kwargs):
        self.log.append(f"LLM response: {response}")

# 仅传入自定义Handler，避免默认StdOutHandler加载
llm = OpenAI(temperature=0)
prompt = PromptTemplate(input_variables=["question"], template="Answer the question: {question}")
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[CustomLogHandler()])

2. 清理已捕获的ANSI编码 #

如果日志已经混入ANSI编码，用正则表达式批量清理：

import re

ansi_pattern = re.compile(r'\x1b\[[0-9;]*m')
clean_log_entries = [ansi_pattern.sub('', entry) for entry in handler.log]

三、更优的Callback日志实现方案 #

1. 用CallbackManager统一管理Handler #

通过CallbackManager可以灵活添加、移除多个Handler，同时避免默认Handler的干扰：

from langchain.callbacks import CallbackManager

# 初始化管理器并添加自定义Handler
callback_manager = CallbackManager([CustomLogHandler()])
chain = LLMChain(llm=llm, prompt=prompt, callback_manager=callback_manager)

2. 实现结构化日志记录 #

将日志存储为字典而非纯字符串，方便后续的解析、过滤或导出到日志系统：

import datetime

class StructuredLogHandler(BaseCallbackHandler):
    def __init__(self):
        self.logs = []
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.logs.append({
            "event_type": "llm_start",
            "timestamp": datetime.datetime.now().isoformat(),
            "prompts": prompts,
            "llm_name": serialized.get("name")
        })
    
    def on_llm_end(self, response, **kwargs):
        self.logs.append({
            "event_type": "llm_end",
            "timestamp": datetime.datetime.now().isoformat(),
            "generations": [gen[0].text for gen in response.generations]
        })

3. 集成Python标准logging模块 #

把自定义Handler和Python内置的logging模块结合，实现分级日志（DEBUG/INFO/WARNING），适配成熟的日志生态：

import logging

# 配置全局日志
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("langchain_app")

class LoggingCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        logger.debug(f"LLM [{serialized.get('name')}] started with prompts: {prompts}")
    
    def on_llm_end(self, response, **kwargs):
        logger.info(f"LLM completed, first response: {response.generations[0][0].text[:100]}...")

内容的提问来源于stack exchange，提问作者Pythonist