Trae Agent:基于LLM的智能软件工程Agent深度技术解析
项目概述
Trae Agent 是字节跳动开源的一个基于大语言模型的智能代理系统,专门针对软件工程任务进行设计和优化。 当前已经展现出了强大的潜力和清晰的架构设计理念。
核心架构设计分析
1. 分层架构模式
Trae Agent 采用了经典的分层架构模式,将系统分为以下几个核心层次:
- Agent 层:负责任务执行逻辑和状态管理
- Tools 层:提供各种工具的具体实现
- Utils 层:提供配置管理、LLM客户端等基础设施
- CLI 层:提供命令行交互接口
2. Agent 基础架构
系统的核心是 Agent 基类,它定义了智能代理的基本框架。 这个设计体现了面向对象设计中的模板方法模式,为不同类型的代理提供了统一的执行框架。trae_agent/agent/base.py(17-42)
class Agent(ABC):
"""Base class for LLM-based agents."""
def __init__(self, config: Config):
self.llm_client: LLMClient = LLMClient(config.default_provider, config.model_providers[config.default_provider])
self.max_steps: int = config.max_steps
self.model_parameters: ModelParameters = config.model_providers[config.default_provider]
self.initial_messages: list[LLMMessage] = []
self.task: str = ""
self.tools: list[Tool] = []
self.tool_caller: ToolExecutor = ToolExecutor([])
self.cli_console: CLIConsole | None = None
# Trajectory recorder
self.trajectory_recorder: TrajectoryRecorder | None = None
def set_trajectory_recorder(self, recorder: TrajectoryRecorder | None) -> None:
"""Set the trajectory recorder for this agent."""
self.trajectory_recorder = recorder
# Also set it on the LLM client
self.llm_client.set_trajectory_recorder(recorder)
def set_cli_console(self, cli_console: CLIConsole | None) -> None:
"""Set the CLI console for this agent."""
self.cli_console = cli_console
- 异步执行模型:整个任务执行流程采用异步设计,能够高效处理 I/O 密集型操作
- 状态机模式:通过
AgentState 枚举管理agent的不同执行状态,确保执行流程的清晰性
- 可扩展性:通过抽象方法定义扩展点,便于实现特定领域的agent
3. TraeAgent 特化实现
TraeAgent 是专门针对软件工程任务的具体实现。 它在基础 Agent 框架之上,添加了软件工程特有的功能。
class TraeAgent(Agent):
"""Trae Agent specialized for software engineering tasks."""
def __init__(self, config: Config):
self.project_path: str = ""
self.base_commit: str | None = None
self.must_patch: str = "false"
self.patch_path: str | None = None
super().__init__(config)
核心特性:
- 项目路径管理:自动管理项目根路径,为工具提供正确的工作环境
- Git 集成:支持基于 commit 的差异对比和补丁生成
- 任务完成检测:实现了智能的任务完成判断逻辑
系统提示词设计: 系统提示词设计体现了软件工程的最佳实践,强调了"先复现问题,再解决问题"的工程思维,这是非常重要的软件调试方法论。trae_agent/agent/trae_agent.py(132-177)
def get_system_prompt(self) -> str:
"""Get the system prompt for TraeAgent."""
return """You are an expert AI software engineering agent.
Your primary goal is to resolve a given GitHub issue by navigating the provided codebase, identifying the root cause of the bug, implementing a robust fix, and ensuring your changes are safe and well-tested.
Follow these steps methodically:
1. Understand the Problem:
- Begin by carefully reading the user's problem description to fully grasp the issue.
- Identify the core components and expected behavior.
2. Explore and Locate:
- Use the available tools to explore the codebase.
- Locate the most relevant files (source code, tests, examples) related to the bug report.
3. Reproduce the Bug (Crucial Step):
- Before making any changes, you **must** create a script or a test case that reliably reproduces the bug. This will be your baseline for verification.
- Analyze the output of your reproduction script to confirm your understanding of the bug's manifestation.
4. Debug and Diagnose:
- Inspect the relevant code sections you identified.
- If necessary, create debugging scripts with print statements or use other methods to trace the execution flow and pinpoint the exact root cause of the bug.
5. Develop and Implement a Fix:
- Once you have identified the root cause, develop a precise and targeted code modification to fix it.
- Use the provided file editing tools to apply your patch. Aim for minimal, clean changes.
6. Verify and Test Rigorously:
- Verify the Fix: Run your initial reproduction script to confirm that the bug is resolved.
- Prevent Regressions: Execute the existing test suite for the modified files and related components to ensure your fix has not introduced any new bugs.
- Write New Tests: Create new, specific test cases (e.g., using `pytest`) that cover the original bug scenario. This is essential to prevent the bug from recurring in the future. Add these tests to the codebase.
- Consider Edge Cases: Think about and test potential edge cases related to your changes.
7. Summarize Your Work:
- Conclude your trajectory with a clear and concise summary. Explain the nature of the bug, the logic of your fix, and the steps you took to verify its correctness and safety.
**Guiding Principle:** Act like a senior software engineer. Prioritize correctness, safety, and high-quality, test-driven development.
# GUIDE FOR HOW TO USE "sequential_thinking" TOOL:
- Your thinking should be thorough and so it's fine if it's very long. Set totalThoughts to at least 5, but setting it up to 25 is fine as well. You'll need more total thoughts when you are considering multiple possible solutions or root causes for an issue.
- Use this tool as much as you find necessary to improve the quality of your answers.
- You can run bash commands (like tests, a reproduction script, or 'grep'/'find' to find relevant context) in between thoughts.
- The sequential_thinking tool can help you break down complex problems, analyze issues step-by-step, and ensure a thorough approach to problem-solving.
- Don't hesitate to use it multiple times throughout your thought process to enhance the depth and accuracy of your solutions.
If you are sure the issue has been solved, you should call the `task_done` to finish the task."""
工具系统深度分析
1. 工具抽象设计
工具系统是 Trae Agent 的核心能力基础。 这个抽象设计展现了优秀的可扩展性:
class Tool(ABC):
"""Base class for all tools."""
def __init__(self):
self.name: str = self.get_name()
self.description: str = self.get_description()
self.parameters: list[ToolParameter] = self.get_parameters()
@abstractmethod
def get_name(self) -> str:
"""Get the tool name."""
pass
@abstractmethod
def get_description(self) -> str:
"""Get the tool description."""
pass
@abstractmethod
def get_parameters(self) -> list[ToolParameter]:
"""Get the tool parameters."""
pass
@abstractmethod
async def execute(self, arguments: ToolCallArguments) -> ToolExecResult:
"""Execute the tool with given parameters."""
pass
- 统一接口:所有工具都实现相同的接口,便于管理和调用
- 参数验证:通过
ToolParameter 提供强类型的参数定义和验证
- 异步执行:支持异步工具执行,提高并发性能
2. 工具执行器设计
ToolExecutor 类实现了工具的并行和串行执行能力。 这种设计允许系统根据需要选择最合适的执行策略。
class ToolExecutor:
"""Tool executor that manages tool execution."""
def __init__(self, tools: list[Tool]):
self.tools: dict[str, Tool] = {tool.name: tool for tool in tools}
async def execute_tool_call(self, tool_call: ToolCall) -> ToolResult:
"""Execute a tool call."""
if tool_call.name not in self.tools:
return ToolResult(
success=False,
error=f"Tool '{tool_call.name}' not found. Available tools: {list(self.tools.keys())}",
call_id=tool_call.call_id,
id=tool_call.id
)
tool = self.tools[tool_call.name]
try:
tool_exec_result = await tool.execute(tool_call.arguments)
return ToolResult(
success=tool_exec_result.error_code == 0,
result=tool_exec_result.output,
error=tool_exec_result.error,
call_id=tool_call.call_id,
id=tool_call.id
)
except Exception as e:
return ToolResult(
success=False,
error=f"Error executing tool '{tool_call.name}': {str(e)}",
call_id=tool_call.call_id,
id=tool_call.id
)
async def parallel_tool_call(self, tool_calls: list[ToolCall]) -> list[ToolResult]:
"""Execute tool calls in parallel"""
return await asyncio.gather(*[self.execute_tool_call(call) for call in tool_calls])
async def sequential_tool_call(self, tool_calls: list[ToolCall]) -> list[ToolResult]:
"""Execute tool calls in sequential"""
return [await self.execute_tool_call(call) for call in tool_calls]
3. 核心工具分析
文件编辑工具
TextEditorTool 是系统中最复杂的工具之一。 它提供了完整的文件操作能力:
class TextEditorTool(Tool):
"""Tool to replace a string in a file."""
@override
def get_name(self) -> str:
return "str_replace_based_edit_tool"
@override
def get_description(self) -> str:
return """Custom editing tool for viewing, creating and editing files
* State is persistent across command calls and discussions with the user
* If `path` is a file, `view` displays the result of applying `cat -n`. If `path` is a directory, `view` lists non-hidden files and directories up to 2 levels deep
* The `create` command cannot be used if the specified `path` already exists as a file !!! If you know that the `path` already exists, please remove it first and then perform the `create` operation!
* If a `command` generates a long output, it will be truncated and marked with `<response clipped>`
Notes for using the `str_replace` command:
* The `old_str` parameter should match EXACTLY one or more consecutive lines from the original file. Be mindful of whitespaces!
* If the `old_str` parameter is not unique in the file, the replacement will not be performed. Make sure to include enough context in `old_str` to make it unique
* The `new_str` parameter should contain the edited lines that should replace the `old_str`
"""
- 精确替换:
str_replace 功能要求精确匹配,避免了意外修改
- 上下文显示:修改后显示代码片段,便于验证结果
- 路径验证:严格的路径验证,防止安全问题
顺序思考工具
SequentialThinkingTool 是一个创新性的设计。 trae_agent/tools/edit_tool.py(31-97)。 它实现了 Chain-of-Thought 推理的工具化:
class SequentialThinkingTool(Tool):
"""A tool for sequential thinking that helps break down complex problems.
This tool helps analyze problems through a flexible thinking process that can adapt and evolve.
Each thought can build on, question, or revise previous insights as understanding deepens.
"""
@override
def get_name(self) -> str:
return "sequentialthinking"
@override
def get_description(self) -> str:
return """A detailed tool for dynamic and reflective problem-solving through thoughts.
This tool helps analyze problems through a flexible thinking process that can adapt and evolve.
Each thought can build on, question, or revise previous insights as understanding deepens.
When to use this tool:
- Breaking down complex problems into steps
- Planning and design with room for revision
- Analysis that might need course correction
- Problems where the full scope might not be clear initially
- Problems that require a multi-step solution
- Tasks that need to maintain context over multiple steps
- Situations where irrelevant information needs to be filtered out
Key features:
- You can adjust total_thoughts up or down as you progress
- You can question or revise previous thoughts
- You can add more thoughts even after reaching what seemed like the end
- You can express uncertainty and explore alternative approaches
- Not every thought needs to build linearly - you can branch or backtrack
- Generates a solution hypothesis
- Verifies the hypothesis based on the Chain of Thought steps
- Repeats the process until satisfied
- Provides a correct answer
Parameters explained:
- thought: Your current thinking step, which can include:
* Regular analytical steps
* Revisions of previous thoughts
* Questions about previous decisions
* Realizations about needing more analysis
* Changes in approach
* Hypothesis generation
* Hypothesis verification
- next_thought_needed: True if you need more thinking, even if at what seemed like the end
- thought_number: Current number in sequence (can go beyond initial total if needed)
- total_thoughts: Current estimate of thoughts needed (can be adjusted up/down)
- is_revision: A boolean indicating if this thought revises previous thinking
- revises_thought: If is_revision is true, which thought number is being reconsidered
- branch_from_thought: If branching, which thought number is the branching point
- branch_id: Identifier for the current branch (if any)
- needs_more_thoughts: If reaching end but realizing more thoughts needed
You should:
1. Start with an initial estimate of needed thoughts, but be ready to adjust
2. Feel free to question or revise previous thoughts
3. Don't hesitate to add more thoughts if needed, even at the "end"
4. Express uncertainty when present
5. Mark thoughts that revise previous thinking or branch into new paths
6. Ignore information that is irrelevant to the current step
7. Generate a solution hypothesis when appropriate
8. Verify the hypothesis based on the Chain of Thought steps
9. Repeat the process until satisfied with the solution
10. Provide a single, ideally correct answer as the final output
11. Only set next_thought_needed to false when truly done and a satisfactory answer is reached"""
- 动态思考流程:支持思考过程中调整思考步数
- 思考分支:支持思考过程中的分支和回溯
- 假设验证:内置假设生成和验证机制
任务执行流程分析
1. 执行主循环
任务执行的主循环体现了状态机模式的精髓。 trae_agent/agent/base.py(49-212):
async def execute_task(self) -> AgentExecution:
"""Execute a task using the agent."""
import time
start_time = time.time()
execution = AgentExecution(task=self.task, steps=[])
try:
messages = self.initial_messages
step_number = 1
while step_number <= self.max_steps:
step = AgentStep(step_number=step_number, state=AgentState.THINKING)
try:
# Get LLM response
step.state = AgentState.THINKING
# Display thinking state
if self.cli_console:
self.cli_console.update_status(step)
llm_response = self.llm_client.chat(messages, self.model_parameters, self.tools)
step.llm_response = llm_response
# Display step with LLM response
if self.cli_console:
self.cli_console.update_status(step)
# Update token usage
if llm_response.usage:
if execution.total_tokens:
execution.total_tokens += llm_response.usage
else:
execution.total_tokens = llm_response.usage
if self.llm_indicates_task_completed(llm_response):
if self.is_task_completed(llm_response):
step.state = AgentState.COMPLETED
execution.final_result = llm_response.content
execution.success = True
# Record agent step
if self.trajectory_recorder:
self.trajectory_recorder.record_agent_step(
step_number=step.step_number,
state=step.state.value,
llm_messages=messages,
llm_response=step.llm_response,
tool_calls=step.tool_calls,
tool_results=step.tool_results,
reflection=step.reflection,
error=step.error
)
if self.cli_console:
self.cli_console.update_status(step)
execution.steps.append(step)
break
else:
step.state = AgentState.THINKING
messages = [LLMMessage(role="user", content=self.task_incomplete_message())]
else:
# Check if the response contains a tool call
tool_calls = llm_response.tool_calls
if tool_calls and len(tool_calls) > 0:
# Execute tool call
step.state = AgentState.CALLING_TOOL
step.tool_calls = tool_calls
# Display tool calling state with tool calls
if self.cli_console:
self.cli_console.update_status(step)
if self.model_parameters.parallel_tool_calls:
tool_results = await self.tool_caller.parallel_tool_call(tool_calls)
else:
tool_results = await self.tool_caller.sequential_tool_call(tool_calls)
step.tool_results = tool_results
# Display tool results
if self.cli_console:
self.cli_console.update_status(step)
messages: list[LLMMessage] = []
for tool_result in tool_results:
# Add tool result to conversation
message = LLMMessage(
role="user",
tool_result=tool_result
)
messages.append(message)
reflection = self.reflect_on_result(tool_results)
if reflection:
step.state = AgentState.REFLECTING
step.reflection = reflection
# Display reflection
if self.cli_console:
self.cli_console.update_status(step)
messages.append(LLMMessage(role="assistant", content=reflection))
else:
messages=[LLMMessage(role="user", content="It seems that you have not completed the task.")]
# Record agent step
if self.trajectory_recorder:
self.trajectory_recorder.record_agent_step(
step_number=step.step_number,
state=step.state.value,
llm_messages=messages,
llm_response=step.llm_response,
tool_calls=step.tool_calls,
tool_results=step.tool_results,
reflection=step.reflection,
error=step.error
)
if self.cli_console:
self.cli_console.update_status(step)
execution.steps.append(step)
step_number += 1
except Exception as e:
step.state = AgentState.ERROR
step.error = str(e)
# Display error
if self.cli_console:
self.cli_console.update_status(step)
# Record agent step
if self.trajectory_recorder:
self.trajectory_recorder.record_agent_step(
step_number=step.step_number,
state=step.state.value,
llm_messages=messages,
llm_response=step.llm_response,
tool_calls=step.tool_calls,
tool_results=step.tool_results,
reflection=step.reflection,
error=step.error
)
if self.cli_console:
self.cli_console.update_status(step)
execution.steps.append(step)
break
if step_number > self.max_steps and not execution.success:
execution.final_result = "Task execution exceeded maximum steps without completion."
except Exception as e:
execution.final_result = f"Agent execution failed: {str(e)}"
execution.execution_time = time.time() - start_time
# Display final summary
if self.cli_console:
self.cli_console.update_status(agent_execution=execution)
return execution
- 步数控制:防止无限执行,通过
max_steps 控制执行边界
- 错误处理:完善的异常处理机制,确保系统稳定性
- 状态追踪:详细的状态变化记录,便于调试和分析
2. 任务完成检测
TraeAgent 实现了智能的任务完成检测机制。 这种多层次的检测机制确保了任务真正完成。
配置管理系统
1. 分级配置策略
配置系统采用了优先级明确的分级策略。提供了极大的灵活性:
优先级顺序:
- 命令行参数(最高优先级)
- 配置文件
- 环境变量
- 默认值(最低优先级)
2. 多模型提供商支持
系统设计了灵活的模型提供商抽象。使系统能够轻松支持不同的 LLM 提供商。
CLI 接口设计
1. 命令结构设计
CLI 接口采用了 Click 框架,提供了清晰的命令结构。
2. 交互模式
交互模式的设计提供了更灵活的使用方式。 允许用户进行多轮对话式的任务执行。
结语
Trae Agent 不仅仅是一个简单的 LLM 包装器,而是一个完整的工程化解决方案:
- 轨迹记录:完整的执行轨迹记录,便于调试和优化
- 错误恢复:智能的错误处理和恢复机制
- 资源管理:合理的资源使用和状态管理
- 插件化工具:工具注册机制便于添加新工具
- 模块化设计:清晰的模块边界便于维护和扩展
Trae Agent 代表了当前 LLM 应用工程化的一个优秀实践。它不仅展现了如何将大语言模型的能力转化为实用的软件工程工具,更重要的是提供了一个可扩展、可维护的架构设计范例。对于想要构建类似系统的开发者来说,这个项目提供了很多值得借鉴的设计思路和实现细节。