在LangChain中创建create_sql_agent时，如何获取prompt、thought与observation？

默认情况下agent_executor.run()仅返回最终答案，要捕获中间的Prompt、思考过程（Thought）和观察结果（Observation），可以通过以下两种方式实现：

方法1：自定义回调函数捕获全流程信息 #

通过继承BaseCallbackHandler实现自定义回调器，在Agent运行的各个阶段拦截并记录关键信息：

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish
from langchain.agents import create_sql_agent
from langchain.sql_database import SQLDatabase
from langchain.chat_models import OpenAI
from langchain.agents.agent_toolkits import SQLDatabaseToolkit

# 初始化数据库和LLM
dbsql = SQLDatabase.from_uri("your_database_uri")
llm = OpenAI(temperature=0, verbose=True)
toolkit = SQLDatabaseToolkit(llm=llm, db=dbsql)

# 自定义回调处理器
class AgentStepLogger(BaseCallbackHandler):
    def __init__(self):
        self.execution_steps = []

    def on_prompt(self, prompt: str, **kwargs) -> None:
        # 记录初始Prompt
        self.execution_steps.append({"type": "prompt", "content": prompt})

    def on_agent_action(self, action: AgentAction, **kwargs) -> None:
        # 拆分并记录Thought
        thought_content = action.log.split("Action:")[0].strip()
        self.execution_steps.append({"type": "thought", "content": thought_content})
        # 记录工具调用信息
        self.execution_steps.append({
            "type": "tool_call",
            "tool": action.tool,
            "input": action.tool_input
        })

    def on_tool_end(self, output: str, **kwargs) -> None:
        # 记录工具返回的Observation
        self.execution_steps.append({"type": "observation", "content": output})

    def on_agent_finish(self, finish: AgentFinish, **kwargs) -> None:
        # 记录最终结论
        self.execution_steps.append({"type": "finish", "content": finish.log})

# 实例化回调器并创建Agent
step_logger = AgentStepLogger()
agent_executor = create_sql_agent(
    llm=OpenAI(temperature=0),
    toolkit=toolkit,
    verbose=True,
    callbacks=[step_logger]
)

# 执行查询
output = agent_executor.run("MY QUESTION")

# 输出捕获的所有步骤
for step in step_logger.execution_steps:
    print(f"[{step['type'].upper()}] {step['content']}")

方法2：使用invoke方法获取完整运行轨迹 #

agent_executor.invoke()会返回包含中间步骤的完整结果字典，其中intermediate_steps字段存储了所有(AgentAction, Observation)元组：

# 用invoke替代run执行查询
result = agent_executor.invoke({"input": "MY QUESTION"})

# 提取最终答案
print(f"最终答案: {result['output']}")

# 解析并输出中间步骤
print("\n=== 中间执行详情 ===")
for action, observation in result["intermediate_steps"]:
    # 从action.log中提取Thought内容
    thought = action.log.split("Action:")[0].strip()
    print(f"思考: {thought}")
    print(f"调用工具: {action.tool}，输入参数: {action.tool_input}")
    print(f"观察结果: {observation}\n")

步骤可视化示例 #

基于捕获的步骤数据，可以快速实现文本形式的可视化：

def visualize_execution(steps):
    print("=== Agent执行流程可视化 ===")
    step_count = 1
    for step in steps:
        match step["type"]:
            case "prompt":
                print(f"📝 初始Prompt:\n{step['content']}\n")
            case "thought":
                print(f"💡 第{step_count}轮思考: {step['content']}")
            case "tool_call":
                print(f"🔧 调用工具: {step['tool']}，输入: {step['input']}")
            case "observation":
                print(f"👀 观察结果: {step['content']}\n")
                step_count +=1
            case "finish":
                print(f"🏁 最终结论: {step['content']}")

# 使用回调器捕获的步骤进行可视化
visualize_execution(step_logger.execution_steps)

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