利用ccglass观测AI Agent内部工作流：从Claude编写贪吃蛇游戏看透LLM请求链路

1. 项目缘起从“黑盒”到“透明”的探索最近在折腾AI Agent开发特别是用Claude Code来写一些自动化脚本和小应用。我让Claude帮我写了一个经典的贪吃蛇游戏整个过程很顺畅代码质量也不错。但作为一个喜欢刨根问底的开发者我总感觉少了点什么——我看到的只是Claude最终吐出来的那几十行Python代码而它背后“思考”的过程比如它调用了哪些工具、如何规划步骤、遇到了什么错误又怎么回退的对我来说完全是个“黑盒”。这就像你请了一个顶级大厨他只把做好的菜端上来却不让你看厨房里发生了什么。对于学习AI工作流和调试复杂的Agent任务来说这种不透明性是个很大的障碍。于是我想搞清楚Claude Code这个“AI员工”到底是怎么干活的。它收到我的指令后内部究竟发起了哪些请求调用了哪些函数参数是什么有没有失败重试传统的做法可能是去翻SDK文档或者看网络日志但都比较麻烦。这时我想到了ccglass这个工具。简单来说ccglass是一个专门用于观测和分析AI Agent内部工作流的工具它能让你像用X光一样“看清”Agent与底层大模型LLM之间的每一次对话、每一次工具调用。这个项目的核心就是利用ccglass把我让Claude写贪吃蛇游戏这个看似简单的任务背后所有的“真实请求”给完整地呈现出来。通过这个实践你不仅能得到一个可玩的贪吃蛇游戏更重要的是你能获得一套方法论去深入理解任何AI Agent的执行链路。这对于调试复杂任务、优化提示词、评估Agent成本乃至学习其决策逻辑都至关重要。无论你是刚接触AI Agent的新手还是想提升现有应用可观测性的资深开发者这个“解剖”过程都会给你带来新的启发。2. 环境与工具准备搭建你的观测平台工欲善其事必先利其器。要让ccglass成功捕获到Claude Code的请求我们需要一个能运行Claude Code并能被ccglass监控的环境。这里我选择了一条对大多数开发者都比较友好的路径在本地使用VSCode配合Claude Code扩展并通过一个Python中间层来“代理”和记录所有请求。2.1 Claude Code的安装与基础配置首先你需要在VSCode中安装Claude Code扩展。打开VSCode进入扩展市场搜索“Claude Code”并安装。安装完成后你通常需要在侧边栏找到Claude的图标并登录你的Claude账户通常是Anthropic的账户来获得API访问权限。这里有一个关键点Claude Code扩展默认会直接使用其集成的后端服务与Claude API通信这个通道我们很难直接拦截。因此我们的策略不是去破解这个扩展而是“模拟”或“重定向”它的请求。一个更可行的方案是我们不直接依赖VSCode扩展的完整工作流而是使用Claude提供的官方Python SDK来构建一个我们能够完全控制的脚本。这样ccglass就能清晰地捕获到我们脚本发出的每一个API调用。你可以通过pip安装Anthropic的官方库pip install anthropic安装后你需要设置你的API密钥。在项目根目录创建一个.env文件将你的Claude API密钥放进去ANTHROPIC_API_KEYyour_api_key_here然后在Python脚本中通过os.getenv或python-dotenv来加载它。这就为我们创建了一个标准、可控的与Claude对话的环境。2.2 认识ccglassAI工作流的“显微镜”ccglass并不是一个广为人知的流行框架它在社区中的定位更偏向于一个专业的可观测性工具。你可以把它理解为一个“AI Agent的调试器”或“工作流录制器”。它的核心功能是装饰Decorator或包装Wrapper你的AI调用代码自动记录下每次LLM调用的输入提示词、输出响应、消耗的Token数、使用的模型、调用的工具函数及其参数和结果、执行耗时等元数据。这些数据会被ccglass收集、结构化并通常以一个本地Web界面的形式展示出来形成一个清晰的可视化时间线。这样你就能一目了然地看到“哦Agent首先分析了我的需求然后调用了Python代码生成工具生成后可能还调用了一个代码验证工具最后把结果返回给了我。” 整个过程不再是魔法而变成了可追溯、可审计的步骤。对于本项目我们不需要部署复杂的ccglass服务端。我们可以利用其核心的日志记录理念自己构建一个轻量级的“请求记录器”。具体来说我们将创建一个自定义的Anthropic客户端类它继承或包装官方客户端在每次调用messages.create方法前后将详细的请求和响应信息以结构化的格式如JSON保存到本地文件或打印出来。这就是我们自制“ccglass”的核心。2.3 构建自制请求记录器下面是一个简单的自制记录器示例它模拟了ccglass的核心记录功能import json import time from datetime import datetime from anthropic import Anthropic import os from dotenv import load_dotenv load_dotenv() class ClaudeClientWithLogging: 一个包装了Anthropic官方客户端并添加了详细日志记录的类。 def __init__(self): self.api_key os.getenv(ANTHROPIC_API_KEY) if not self.api_key: raise ValueError(请在.env文件中设置ANTHROPIC_API_KEY) # 初始化真正的客户端 self.client Anthropic(api_keyself.api_key) # 日志文件 self.log_file fclaude_trace_{datetime.now().strftime(%Y%m%d_%H%M%S)}.jsonl self.log_entries [] def log_interaction(self, entry): 记录一次交互到内存和文件。 self.log_entries.append(entry) # 使用jsonl格式每行一个JSON对象便于后续分析 with open(self.log_file, a, encodingutf-8) as f: f.write(json.dumps(entry, ensure_asciiFalse) \n) print(f[LOG] 记录一次交互: {entry.get(type, unknown)}) def create_message(self, **kwargs): 包装messages.create方法记录请求和响应。 trace_id ftrace_{int(time.time()*1000)} # 1. 记录请求 request_entry { trace_id: trace_id, type: request, timestamp: datetime.now().isoformat(), model: kwargs.get(model), max_tokens: kwargs.get(max_tokens), messages: kwargs.get(messages), tools: kwargs.get(tools), # 记录可能使用的工具定义 metadata: kwargs.get(metadata, {}) } self.log_interaction(request_entry) # 2. 执行实际请求并计时 start_time time.time() try: response self.client.messages.create(**kwargs) elapsed_ms int((time.time() - start_time) * 1000) except Exception as e: elapsed_ms int((time.time() - start_time) * 1000) # 3. 记录错误响应 error_entry { trace_id: trace_id, type: response_error, timestamp: datetime.now().isoformat(), elapsed_ms: elapsed_ms, error: str(e) } self.log_interaction(error_entry) raise e # 4. 记录成功响应 response_entry { trace_id: trace_id, type: response_success, timestamp: datetime.now().isoformat(), elapsed_ms: elapsed_ms, response_id: response.id, model: response.model, role: response.role, content: [c.model_dump() for c in response.content], # 将内容对象转为字典 usage: { input_tokens: response.usage.input_tokens, output_tokens: response.usage.output_tokens }, stop_reason: response.stop_reason } self.log_interaction(response_entry) return response # 使用我们包装的客户端 client ClaudeClientWithLogging()这个ClaudeClientWithLogging类就是我们的“ccglass”核心。它会在每次API调用时生成一个唯一的trace_id来关联请求和响应并详细记录下所有相关信息保存为.jsonl格式的日志文件。这种格式易于用脚本分析也方便导入到其他可视化工具中。3. 任务执行与请求捕获让贪吃蛇“现形”有了可控的客户端和记录器我们就可以开始执行“编写贪吃蛇游戏”这个任务并观察全过程了。我们的目标不是简单地拿到最终代码而是记录下Claude为了完成这个任务可能进行的多轮对话和思考过程。3.1 设计一个能激发多步推理的提示词如果我们只是问“用Python写一个贪吃蛇游戏”Claude可能会直接输出一整段代码。这虽然高效但内部可能只是一次简单的代码生成调用观测到的链路太短价值不大。为了看到更丰富的Agent行为比如规划、分解任务、可能调用代码解释或验证工具我们需要设计一个更复杂、要求更高的提示词。我使用了类似下面的提示词旨在模拟一个真实开发场景system_prompt 你是一个专业的Python开发助手。请按照以下步骤为用户创建一个可运行的贪吃蛇游戏 1. 首先分析在命令行环境中实现贪吃蛇游戏需要哪些核心模块如输入处理、游戏循环、图形绘制等。 2. 然后逐步构建游戏。先实现游戏地图和蛇的初始状态表示。 3. 接着实现蛇的移动逻辑和方向控制。 4. 再实现食物的随机生成和碰撞检测吃到食物、撞墙、撞自身。 5. 最后整合所有部分并确保游戏循环逻辑正确。 请分步进行每一步完成后向我确认或直接给出该部分的代码。如果使用任何假设如使用curses库或keyboard库请先说明。这个提示词要求分步执行并且提到了具体的库curses这更有可能让Claude在回复中展示其“思考-行动”的过程甚至可能主动提出要使用某个“代码执行”工具来验证部分逻辑虽然在我们这个简单记录场景它没有实际执行环境但请求中可能会包含工具调用的意图。3.2 发起对话并记录全链路现在我们使用包装好的客户端发起这次对话# 发起初始请求 print(开始请求Claude创建贪吃蛇游戏...) try: response client.create_message( modelclaude-3-5-sonnet-20241022, # 使用一个较新的模型 max_tokens4000, systemsystem_prompt, messages[ {role: user, content: 请开始创建贪吃蛇游戏。} ] # 注意这里我们没有传递tools参数因此Claude不会进行实际工具调用。 # 但它的回复内容中可能会以文本形式描述其“计划”或“步骤”。 ) # 打印Claude的第一轮回复 print(\n--- Claude的第一轮回复 ---) for content_block in response.content: if content_block.type text: print(content_block.text) # 接下来我们可以根据Claude的回复进行多轮对话 # 例如Claude可能会说“第一步我们需要表示游戏区域。我假设我们使用一个二维列表...” # 我们可以回复“好的请给出这一步的代码。” # 这个过程会持续直到游戏完成。 # 每一轮对话都会被我们的client.create_message方法记录。 except Exception as e: print(f请求过程中发生错误: {e})在实际操作中我让这个对话进行了3-4轮。Claude首先给出了一个计划然后分步提供了地图表示、蛇的移动、食物生成等代码片段。每一轮问答都被我们的记录器完整地捕获了下来生成了一个包含多个request和response_success条目的日志文件。3.3 关键发现解读日志中的“真实请求”执行完毕后打开生成的.jsonl日志文件我们就能像用ccglass一样清晰地看到整个工作流。以下是我从日志中提炼的一些关键观察这些正是“真实请求”的细节请求的结构化每个请求体都是一个清晰的JSON对象包含了model,messages(对话历史),max_tokens,temperature(虽然我们没指定默认会有) 等字段。这揭示了Claude API的输入规范。对话历史的维护在第二轮及之后的请求中messages数组包含了之前所有的用户和助手消息。这意味着每次请求我们都需要把完整的上下文发送过去Token消耗会累积。这对于计算成本和理解Agent的“记忆”机制很重要。响应的复杂性响应体不仅仅是文本。content字段是一个列表其中可以包含type为text的文本块。如果启用了工具调用这里还可能出现type为tool_use的块。我们的简单任务没有用到工具所以全是文本块。这帮助我们理解了Agent输出结果的标准化格式。Token消耗明细每个响应都附带了usage字段明确列出了本次调用消耗的input_tokens和output_tokens。这是进行成本核算和性能优化的直接依据。例如我发现让Claude分步输出虽然请求次数多了但单次输出的Token数较少总成本可能与一次输出长篇代码相差不大但获得了更清晰的过程。元信息丰富response_id,model,stop_reason等字段对于调试和日志追踪非常有价值。stop_reason如果是“max_tokens”就说明输出被截断了需要调整max_tokens参数。通过分析这些日志我恍然大悟原来Claude Code或任何基于类似API的Agent在VSCode里看似流畅的交互背后就是这些一次次的HTTP请求堆砌起来的。它的“思考”过程就蕴含在我们提供给它的对话历史以及它根据这些历史生成的下一段文本之中。4. 深度分析从请求日志反推AI Agent的工作模式捕获到原始请求和响应只是第一步就像拿到了飞机的黑匣子数据。接下来我们需要解读这些数据从而理解AI Agent在这个例子中是Claude是如何工作的。这对于我们设计更高效的提示词、构建更稳定的Agent系统至关重要。4.1 会话状态的管理与成本从日志中可以明显看出Claude API本身是无状态的。它不会记住上一次对话的内容除非我们在本次请求的messages参数中把历史对话全部传递过去。这就是所谓的“上下文窗口”Context Window管理。我们的做法在每轮对话中我们都将之前所有的user和assistant消息作为上下文传入。这保证了对话的连贯性Claude才能基于之前的讨论继续编写代码。带来的影响Token成本输入Token数会随着对话轮次线性增长实际上是累积增长。一次长对话的总成本可能远高于多次独立短对话。在我们的贪吃蛇例子中虽然分了4步但每次请求都携带了增长的历史总输入Token数大约是最终代码Token数的好几倍。上下文长度限制所有模型都有上下文长度上限如200K Token。对于超长对话需要设计摘要或选择性遗忘的机制这本身就是Agent架构中的一个核心挑战。工程启示在构建生产级Agent时我们不能无脑地堆积全部历史。需要实现一个“对话记忆管理”模块可能采用“滑动窗口”只保留最近N条消息或者对早期历史进行“摘要”后再放入上下文。ccglass这类工具能帮你精确分析每一轮消耗的Token从而优化这个策略。4.2 工具调用Tool Use的潜在模式虽然我们这个简单任务没有显式使用工具但Claude API是支持工具调用Function Calling/Tool Use的。在更复杂的Agent场景中这是核心能力。通过分析请求日志的格式我们可以推断出其工作模式工具定义在请求中如果我们需要Claude使用工具比如执行一段生成的代码、查询数据库我们必须在请求的tools参数中以JSON Schema格式定义好工具的名称、描述和参数。这相当于给了Agent一个“技能列表”。工具调用在响应中Claude的响应content里可能会出现type为tool_use的块其中包含它决定调用的工具名称和参数。然后我们需要在本地执行这个工具如运行一段Python代码并将执行结果作为下一条user消息的一部分以type为tool_result的块形式发送给Claude。循环往复这样就形成了一个“模型思考 - 决定调用工具 - 用户执行工具并返回结果 - 模型基于结果继续思考”的循环。一个强大的AI Agent正是通过多次这样的循环来完成复杂任务的。我们的日志记录器可以清晰地记录下每一次“模型请求”、“工具调用请求”、“工具返回结果”的节点从而绘制出完整的Agent工作流图谱。实操心得在定义工具时描述description字段极其重要。清晰、具体的描述能极大提高模型选择正确工具和生成正确参数的几率。例如与其定义一个叫run_code的工具不如定义为run_python_code_in_isolated_sandbox并在描述中写明“在安全的沙箱环境中执行一段Python代码字符串返回标准输出和标准错误。无法访问网络和文件系统。”4.3 停止原因Stop Reason与输出控制响应中的stop_reason字段是一个重要的诊断信息。它告诉我们模型为什么停止生成文本。“end_turn”模型自然地结束了它的回复。这是最理想的情况。“max_tokens”达到了max_tokens参数设置的限制输出被截断。这说明我们设置的max_tokens可能不够需要根据任务调大。但也要注意盲目调大会增加不必要的成本和等待时间。“stop_sequence”遇到了我们预设的停止序列如“\n\nHuman:”。这在设计特定的交互格式时有用。在我们的贪吃蛇任务日志中所有回复的stop_reason都是“end_turn”说明我们的max_tokens设置是充足的对话进行得很顺畅。如果看到“max_tokens”我们就需要检查输出的代码是否完整很可能游戏循环的最后几行被截掉了导致代码无法运行。5. 可视化与进阶应用让日志“活”起来原始的JSONL日志文件虽然信息全面但可读性不强。ccglass这类工具的核心价值之一就是提供可视化界面。我们可以用一些简单的方法自己实现基础的可视化让分析更直观。5.1 使用Streamlit快速构建日志查看器Streamlit是一个快速创建数据应用的工具非常适合用来将我们的日志可视化。下面是一个简单的脚本可以展示对话流和Token消耗# trace_viewer.py import streamlit as st import json import pandas as pd from datetime import datetime st.set_page_config(page_titleClaude 请求追踪分析器, layoutwide) st.title( 贪吃蛇游戏开发 - Claude 请求链路追踪) uploaded_file st.file_uploader(上传你的 .jsonl 追踪日志文件, type[jsonl]) if uploaded_file is not None: logs [] for line in uploaded_file: logs.append(json.loads(line.decode(utf-8))) df pd.DataFrame(logs) # 1. 显示总体统计 st.header( 总体统计) col1, col2, col3 st.columns(3) total_requests len(df[df[type].str.contains(request)]) total_input_tokens df[df[type]response_success][usage].apply(lambda x: x.get(input_tokens, 0)).sum() total_output_tokens df[df[type]response_success][usage].apply(lambda x: x.get(output_tokens, 0)).sum() col1.metric(总请求数, total_requests) col2.metric(总输入Token, f{total_input_tokens:,}) col3.metric(总输出Token, f{total_output_tokens:,}) # 2. 显示对话时间线 st.header( 对话时间线) df[timestamp] pd.to_datetime(df[timestamp]) df_sorted df.sort_values(timestamp) for idx, row in df_sorted.iterrows(): with st.expander(f{row[timestamp].strftime(%H:%M:%S)} - {row[type]} (Trace: {row.get(trace_id, N/A)})): if row[type] request: st.markdown(**请求详情**) st.json(row.get(messages, []), expandedFalse) if row.get(tools): st.markdown(**定义的工具**) st.json(row[tools], expandedFalse) elif row[type] response_success: st.markdown(**响应详情**) st.text(f模型: {row.get(model)}, 耗时: {row.get(elapsed_ms)}ms) st.text(fToken消耗: 输入 {row.get(usage, {}).get(input_tokens)}, 输出 {row.get(usage, {}).get(output_tokens)}) st.markdown(**回复内容**) for content in row.get(content, []): if content[type] text: st.code(content[text], languagepython if python in content[text].lower() else text) # 3. Token消耗折线图 st.header( Token消耗趋势) success_df df[df[type]response_success].copy() if not success_df.empty: success_df[request_seq] range(1, len(success_df)1) chart_data success_df[[request_seq, usage]].copy() chart_data[input_tokens] chart_data[usage].apply(lambda x: x.get(input_tokens)) chart_data[output_tokens] chart_data[usage].apply(lambda x: x.get(output_tokens)) st.line_chart(chart_data.set_index(request_seq)[[input_tokens, output_tokens]])运行streamlit run trace_viewer.py上传你的日志文件你就能看到一个交互式的仪表盘清晰地展示了整个对话流程、每轮的具体内容以及Token消耗的变化趋势。这比看纯文本日志要直观得多。5.2 基于观测的Agent优化实践观测的最终目的是优化。通过分析这次贪吃蛇任务的日志我们可以得出一些优化Agent设计的经验提示词工程我发现在第一轮请求中由于我的系统提示词较长消耗了约500个输入Token。如果这是一个会被频繁调用的Agent可以考虑精简系统提示词或者将固定的上下文如公司规范、代码风格通过其他方式如RAG检索动态注入以减少每次请求的固定开销。分步vs整体分步请求让过程更透明也便于在中间步骤进行人工校正或干预。但代价是总输入Token数增加因为要携带历史。对于逻辑复杂、容错率低的任务如部署脚本分步更安全。对于简单、成熟的任务如生成工具函数一次性请求更经济。ccglass的数据可以帮助你做出量化决策。错误处理与重试在我们的记录器中我们捕获了请求异常。在生产环境中你需要根据不同的错误类型如网络超时、速率限制、内容过滤设计重试、回退或上报逻辑。观测到的错误日志是设计这些策略的基础。成本监控与预警通过实时分析usage数据可以设置成本预警。例如如果单次对话的输入Token超过某个阈值可能意味着上下文积累过多需要触发摘要或清理操作。5.3 将模式推广到复杂Agent系统贪吃蛇游戏只是一个简单的起点。这种“请求捕获-分析-可视化”的模式可以无缝推广到任何复杂的AI Agent系统中。多工具编排当一个Agent需要顺序或并行调用搜索工具、代码解释器、数据库查询等多个工具时ccglass这样的观测平台能清晰地展示出整个编排流程帮你发现瓶颈比如某个工具调用特别慢或逻辑错误比如工具调用顺序不对。评估与测试你可以录制Agent处理一系列标准测试用例的完整流程。通过对比不同提示词版本或模型版本下的工作流日志可以定量评估哪个版本更高效、更准确、成本更低。调试与溯源当用户报告Agent输出了一个错误结果时你可以通过trace_id快速定位到当时的完整会话上下文、工具调用参数和结果精准复现问题而不是靠猜测。让Claude写贪吃蛇游戏只是一个引子真正有价值的是通过ccglass或自制的记录器打开的那个“观察窗”。它让你从被动的API调用者变成了主动的AI工作流架构师和调试员。当你能够看清每一个请求、每一次思考、每一次工具调用的细节时你构建和驾驭AI Agent的能力才会发生质的飞跃。