9.3 Practice in Hello-Agents: ContextBuilder

This section will detail the context engineering practice in the HelloAgents framework. We will gradually demonstrate how to build a production-grade context management system from design motivation, core data structures, implementation details to complete cases. The design philosophy of ContextBuilder is "simple and efficient", removing unnecessary complexity, uniformly selecting based on "relevance + recency" scores, conforming to the engineering orientation of Agent modularity and maintainability.

9.3.1 Design Motivation and Goals

Before building ContextBuilder, we first need to clarify its design goals and core value. An excellent context management system should solve the following key problems:

1. Unified Entry: Abstract "Gather-Select-Structure-Compress" as a reusable pipeline, reducing repetitive template code in Agent implementations. This unified interface design allows developers to avoid repeatedly writing context management logic in each Agent.

2. Stable Form: Output a context template with a fixed skeleton, facilitating debugging, A/B testing, and evaluation. We adopted a sectioned template structure:

   • [Role & Policies]: Clarify the Agent's role positioning and behavioral guidelines
   • [Task]: The specific task currently to be completed
   • [State]: The Agent's current state and context information
   • [Evidence]: Evidence information retrieved from external knowledge bases
   • [Context]: Historical dialogue and related memories
   • [Output]: Expected output format and requirements

3. Budget Guardian: Retain high-value information as much as possible within the token budget, providing fallback compression strategies for over-limit contexts. This ensures that even in scenarios with huge amounts of information, the system can run stably.

4. Minimum Rules: Do not introduce classification dimensions such as source/priority to avoid complexity growth. Practice shows that a simple scoring mechanism based on relevance and recency is effective enough in most scenarios.

9.3.2 Core Data Structures

The implementation of ContextBuilder relies on two core data structures that define the system's configuration and information units.

(1) ContextPacket: Candidate Information Package

from dataclasses import dataclass
from typing import Optional, Dict, Any
from datetime import datetime

@dataclass
class ContextPacket:
    """Candidate information package

    Attributes:
        content: Information content
        timestamp: Timestamp
        token_count: Token count
        relevance_score: Relevance score (0.0-1.0)
        metadata: Optional metadata
    """
    content: str
    timestamp: datetime
    token_count: int
    relevance_score: float = 0.5
    metadata: Optional[Dict[str, Any]] = None

    def __post_init__(self):
        """Post-initialization processing"""
        if self.metadata is None:
            self.metadata = {}
        # Ensure relevance score is within valid range
        self.relevance_score = max(0.0, min(1.0, self.relevance_score))

ContextPacket is the basic unit of information in the system. Each candidate information is encapsulated as a ContextPacket, containing core attributes such as content, timestamp, token count, and relevance score. This unified data structure simplifies subsequent selection and sorting logic.

(2) ContextConfig: Configuration Management

@dataclass
class ContextConfig:
    """Context building configuration

    Attributes:
        max_tokens: Maximum token count
        reserve_ratio: Ratio reserved for system instructions (0.0-1.0)
        min_relevance: Minimum relevance threshold
        enable_compression: Whether to enable compression
        recency_weight: Recency weight (0.0-1.0)
        relevance_weight: Relevance weight (0.0-1.0)
    """
    max_tokens: int = 3000
    reserve_ratio: float = 0.2
    min_relevance: float = 0.1
    enable_compression: bool = True
    recency_weight: float = 0.3
    relevance_weight: float = 0.7

    def __post_init__(self):
        """Validate configuration parameters"""
        assert 0.0 <= self.reserve_ratio <= 1.0, "reserve_ratio must be in [0, 1] range"
        assert 0.0 <= self.min_relevance <= 1.0, "min_relevance must be in [0, 1] range"
        assert abs(self.recency_weight + self.relevance_weight - 1.0) < 1e-6, \
            "recency_weight + relevance_weight must equal 1.0"

ContextConfig encapsulates all configurable parameters, making system behavior flexibly adjustable. Particularly noteworthy is the reserve_ratio parameter, which ensures that key information such as system instructions always has sufficient space and will not be squeezed out by other information.

9.3.3 GSSC Pipeline Detailed Explanation

The core of ContextBuilder is the GSSC (Gather-Select-Structure-Compress) pipeline, which decomposes the context building process into four clear stages. Let's dive into the implementation details of each stage.

(1) Gather: Multi-source Information Collection

The first stage is to collect candidate information from multiple sources. The key to this stage is fault tolerance and flexibility.

def _gather(
    self,
    user_query: str,
    conversation_history: Optional[List[Message]] = None,
    system_instructions: Optional[str] = None,
    custom_packets: Optional[List[ContextPacket]] = None
) -> List[ContextPacket]:
    """Collect all candidate information

    Args:
        user_query: User query
        conversation_history: Conversation history
        system_instructions: System instructions
        custom_packets: Custom information packages

    Returns:
        List[ContextPacket]: Candidate information list
    """
    packets = []

    # 1. Add system instructions (highest priority, not scored)
    if system_instructions:
        packets.append(ContextPacket(
            content=system_instructions,
            timestamp=datetime.now(),
            token_count=self._count_tokens(system_instructions),
            relevance_score=1.0,  # System instructions always retained
            metadata={"type": "system_instruction", "priority": "high"}
        ))

    # 2. Retrieve relevant memories from memory system
    if self.memory_tool:
        try:
            memory_results = self.memory_tool.run({
                "action": "search",
                "query": user_query,
                "limit": 10,
                "min_importance": 0.3
            })
            # Parse memory results and convert to ContextPacket
            memory_packets = self._parse_memory_results(memory_results, user_query)
            packets.extend(memory_packets)
        except Exception as e:
            print(f"[WARNING] Memory retrieval failed: {e}")

    # 3. Retrieve relevant knowledge from RAG system
    if self.rag_tool:
        try:
            rag_results = self.rag_tool.run({
                "action": "search",
                "query": user_query,
                "limit": 5,
                "min_score": 0.3
            })
            # Parse RAG results and convert to ContextPacket
            rag_packets = self._parse_rag_results(rag_results, user_query)
            packets.extend(rag_packets)
        except Exception as e:
            print(f"[WARNING] RAG retrieval failed: {e}")

    # 4. Add conversation history (only keep recent N entries)
    if conversation_history:
        recent_history = conversation_history[-5:]  # Default keep recent 5 entries
        for msg in recent_history:
            packets.append(ContextPacket(
                content=f"{msg.role}: {msg.content}",
                timestamp=msg.timestamp if hasattr(msg, 'timestamp') else datetime.now(),
                token_count=self._count_tokens(msg.content),
                relevance_score=0.6,  # Base relevance of historical messages
                metadata={"type": "conversation_history", "role": msg.role}
            ))

    # 5. Add custom information packages
    if custom_packets:
        packets.extend(custom_packets)

    print(f"[ContextBuilder] Collected {len(packets)} candidate information packages")
    return packets

This implementation demonstrates several important design considerations:

• Fault Tolerance Mechanism: Each external data source call is wrapped in try-except, ensuring that failure of a single source does not affect the overall process
• Priority Handling: System instructions are marked as high priority, ensuring they are always retained
• History Limitation: Conversation history only keeps the most recent entries, avoiding the context window being occupied by historical information

(2) Select: Intelligent Information Selection

The second stage is to score and select candidate information based on relevance and recency. This is the core of the entire pipeline and directly determines the quality of the final context.

def _select(
    self,
    packets: List[ContextPacket],
    user_query: str,
    available_tokens: int
) -> List[ContextPacket]:
    """Select the most relevant information packages

    Args:
        packets: Candidate information package list
        user_query: User query (for calculating relevance)
        available_tokens: Available token count

    Returns:
        List[ContextPacket]: Selected information package list
    """
    # 1. Separate system instructions and other information
    system_packets = [p for p in packets if p.metadata.get("type") == "system_instruction"]
    other_packets = [p for p in packets if p.metadata.get("type") != "system_instruction"]

    # 2. Calculate tokens occupied by system instructions
    system_tokens = sum(p.token_count for p in system_packets)
    remaining_tokens = available_tokens - system_tokens

    if remaining_tokens <= 0:
        print("[WARNING] System instructions have occupied all token budget")
        return system_packets

    # 3. Calculate comprehensive scores for other information
    scored_packets = []
    for packet in other_packets:
        # Calculate relevance score (if not yet calculated)
        if packet.relevance_score == 0.5:  # Default value, needs recalculation
            relevance = self._calculate_relevance(packet.content, user_query)
            packet.relevance_score = relevance

        # Calculate recency score
        recency = self._calculate_recency(packet.timestamp)

        # Combined score = relevance weight × relevance + recency weight × recency
        combined_score = (
            self.config.relevance_weight * packet.relevance_score +
            self.config.recency_weight * recency
        )

        # Filter information below minimum relevance threshold
        if packet.relevance_score >= self.config.min_relevance:
            scored_packets.append((combined_score, packet))

    # 4. Sort by score in descending order
    scored_packets.sort(key=lambda x: x[0], reverse=True)

    # 5. Greedy selection: fill from high to low score until token limit is reached
    selected = system_packets.copy()
    current_tokens = system_tokens

    for score, packet in scored_packets:
        if current_tokens + packet.token_count <= available_tokens:
            selected.append(packet)
            current_tokens += packet.token_count
        else:
            # Token budget is full, stop selection
            break

    print(f"[ContextBuilder] Selected {len(selected)} information packages, total {current_tokens} tokens")
    return selected

def _calculate_relevance(self, content: str, query: str) -> float:
    """Calculate relevance between content and query

    Uses simple keyword overlap algorithm. In production, can be replaced with vector similarity calculation.

    Args:
        content: Content text
        query: Query text

    Returns:
        float: Relevance score (0.0-1.0)
    """
    # Tokenization (simple implementation, can use more complex tokenizers)
    content_words = set(content.lower().split())
    query_words = set(query.lower().split())

    if not query_words:
        return 0.0

    # Jaccard similarity
    intersection = content_words & query_words
    union = content_words | query_words

    return len(intersection) / len(union) if union else 0.0

def _calculate_recency(self, timestamp: datetime) -> float:
    """Calculate temporal recency score

    Uses exponential decay model, maintains high score within 24 hours, then gradually decays.

    Args:
        timestamp: Information timestamp

    Returns:
        float: Recency score (0.0-1.0)
    """
    import math

    age_hours = (datetime.now() - timestamp).total_seconds() / 3600

    # Exponential decay: maintain high score within 24 hours, then gradually decay
    decay_factor = 0.1  # Decay coefficient
    recency_score = math.exp(-decay_factor * age_hours / 24)

    return max(0.1, min(1.0, recency_score))  # Limit to [0.1, 1.0] range

The core algorithm of the selection stage embodies several important engineering considerations:

• Scoring Mechanism: Uses weighted combination of relevance and recency, with configurable weights
• Greedy Algorithm: Fills from high to low score, ensuring selection of the most valuable information within limited budget
• Filtering Mechanism: Filters low-quality information through the min_relevance parameter

(3) Structure: Structured Output

The third stage is to organize selected information into a structured context template.

def _structure(self, selected_packets: List[ContextPacket], user_query: str) -> str:
    """Organize selected information packages into structured context template

    Args:
        selected_packets: Selected information package list
        user_query: User query

    Returns:
        str: Structured context string
    """
    # Group by type
    system_instructions = []
    evidence = []
    context = []

    for packet in selected_packets:
        packet_type = packet.metadata.get("type", "general")

        if packet_type == "system_instruction":
            system_instructions.append(packet.content)
        elif packet_type in ["rag_result", "knowledge"]:
            evidence.append(packet.content)
        else:
            context.append(packet.content)

    # Build structured template
    sections = []

    # [Role & Policies]
    if system_instructions:
        sections.append("[Role & Policies]\n" + "\n".join(system_instructions))

    # [Task]
    sections.append(f"[Task]\n{user_query}")

    # [Evidence]
    if evidence:
        sections.append("[Evidence]\n" + "\n---\n".join(evidence))

    # [Context]
    if context:
        sections.append("[Context]\n" + "\n".join(context))

    # [Output]
    sections.append("[Output]\nPlease provide accurate, evidence-based answers based on the above information.")

    return "\n\n".join(sections)

The structuring stage organizes scattered information packages into clear sections. This design has several advantages:

• Readability: Clear sections make it easier for both humans and models to understand the context structure
• Debuggability: Problem localization is easier, can quickly identify which area has problematic information
• Extensibility: Adding new information sources only requires creating new sections

(4) Compress: Fallback Compression

The fourth stage is to compress over-limit contexts.

def _compress(self, context: str, max_tokens: int) -> str:
    """Compress over-limit context

    Args:
        context: Original context
        max_tokens: Maximum token limit

    Returns:
        str: Compressed context
    """
    current_tokens = self._count_tokens(context)

    if current_tokens <= max_tokens:
        return context  # No compression needed

    print(f"[ContextBuilder] Context over limit ({current_tokens} > {max_tokens}), executing compression")

    # Section compression: maintain structural integrity
    sections = context.split("\n\n")
    compressed_sections = []
    current_total = 0

    for section in sections:
        section_tokens = self._count_tokens(section)

        if current_total + section_tokens <= max_tokens:
            # Fully retain
            compressed_sections.append(section)
            current_total += section_tokens
        else:
            # Partially retain
            remaining_tokens = max_tokens - current_total
            if remaining_tokens > 50:  # Retain at least 50 tokens
                # Simple truncation (can use LLM summarization in production)
                truncated = self._truncate_text(section, remaining_tokens)
                compressed_sections.append(truncated + "\n[... Content compressed ...]")
            break

    compressed_context = "\n\n".join(compressed_sections)
    final_tokens = self._count_tokens(compressed_context)
    print(f"[ContextBuilder] Compression complete: {current_tokens} -> {final_tokens} tokens")

    return compressed_context

def _truncate_text(self, text: str, max_tokens: int) -> str:
    """Truncate text to specified token count

    Args:
        text: Original text
        max_tokens: Maximum token count

    Returns:
        str: Truncated text
    """
    # Simple implementation: estimate by character ratio
    # Should use precise tokenizer in production
    char_per_token = len(text) / self._count_tokens(text) if self._count_tokens(text) > 0 else 4
    max_chars = int(max_tokens * char_per_token)

    return text[:max_chars]

def _count_tokens(self, text: str) -> int:
    """Estimate token count of text

    Args:
        text: Text content

    Returns:
        int: Token count
    """
    # Simple estimation: Chinese 1 char ≈ 1 token, English 1 word ≈ 1.3 tokens
    # Should use actual tokenizer in production
    chinese_chars = sum(1 for ch in text if '\u4e00' <= ch <= '\u9fff')
    english_words = len([w for w in text.split() if w])

    return int(chinese_chars + english_words * 1.3)

The design of the compression stage embodies the principle of "maintaining structural integrity". Even when the token budget is tight, it tries to retain key information from each section.

9.3.4 Complete Usage Example

Now let's demonstrate how to use ContextBuilder in actual projects through a complete example.

(1) Basic Usage

from hello_agents.context import ContextBuilder, ContextConfig
from hello_agents.tools import MemoryTool, RAGTool
from hello_agents.core.message import Message
from datetime import datetime

# 1. Initialize tools
memory_tool = MemoryTool(user_id="user123")
rag_tool = RAGTool(knowledge_base_path="./knowledge_base")

# 2. Create ContextBuilder
config = ContextConfig(
    max_tokens=3000,
    reserve_ratio=0.2,
    min_relevance=0.2,
    enable_compression=True
)

builder = ContextBuilder(
    memory_tool=memory_tool,
    rag_tool=rag_tool,
    config=config
)

# 3. Prepare conversation history
conversation_history = [
    Message(content="I'm developing a data analysis tool", role="user", timestamp=datetime.now()),
    Message(content="Great! Data analysis tools usually need to handle large amounts of data. What tech stack do you plan to use?", role="assistant", timestamp=datetime.now()),
    Message(content="I plan to use Python and Pandas, and have completed the CSV reading module", role="user", timestamp=datetime.now()),
    Message(content="Good choice! Pandas is very powerful for data processing. Next you may need to consider data cleaning and transformation.", role="assistant", timestamp=datetime.now()),
]

# 4. Add some memories
memory_tool.run({
    "action": "add",
    "content": "User is developing a data analysis tool using Python and Pandas",
    "memory_type": "semantic",
    "importance": 0.8
})

memory_tool.run({
    "action": "add",
    "content": "Completed development of CSV reading module",
    "memory_type": "episodic",
    "importance": 0.7
})

# 5. Build context
context = builder.build(
    user_query="How to optimize Pandas memory usage?",
    conversation_history=conversation_history,
    system_instructions="You are a senior Python data engineering consultant. Your answers need to: 1) Provide specific actionable advice 2) Explain technical principles 3) Provide code examples"
)

print("=" * 80)
print("Built context:")
print("=" * 80)
print(context)
print("=" * 80)

(2) Running Effect Demonstration

After running the above code, you will see the following structured context output:

================================================================================
Built context:
================================================================================
[Role & Policies]
You are a senior Python data engineering consultant. Your answers need to: 1) Provide specific actionable advice 2) Explain technical principles 3) Provide code examples

[Task]
How to optimize Pandas memory usage?

[Evidence]
Core strategies for Pandas memory optimization include:
1. Use appropriate data types (such as category instead of object)
2. Read large files in chunks
3. Use chunksize parameter
---
Data type optimization can significantly reduce memory usage. For example, downgrading int64 to int32 can save 50% memory.

[Context]
user: I'm developing a data analysis tool
assistant: Great! Data analysis tools usually need to handle large amounts of data. What tech stack do you plan to use?
user: I plan to use Python and Pandas, and have completed the CSV reading module
assistant: Good choice! Pandas is very powerful for data processing. Next you may need to consider data cleaning and transformation.
Memory: User is developing a data analysis tool using Python and Pandas
Memory: Completed development of CSV reading module

[Output]
Please provide accurate, evidence-based answers based on the above information.
================================================================================

This structured context contains all necessary information:

• [Role & Policies]: Clarifies the AI's role and answer requirements
• [Task]: Clearly expresses the user's question
• [Evidence]: Relevant knowledge retrieved from the RAG system
• [Context]: Conversation history and related memories, providing sufficient background information
• [Output]: Guides the LLM on how to organize the answer

(3) Integration with Agent

Finally, let's demonstrate how to integrate ContextBuilder into an Agent:

from hello_agents import SimpleAgent, HelloAgentsLLM, ToolRegistry
from hello_agents.context import ContextBuilder, ContextConfig
from hello_agents.tools import MemoryTool, RAGTool

class ContextAwareAgent(SimpleAgent):
    """Agent with context awareness capability"""

    def __init__(self, name: str, llm: HelloAgentsLLM, **kwargs):
        super().__init__(name=name, llm=llm, system_prompt=kwargs.get("system_prompt", ""))

        # Initialize context builder
        self.memory_tool = MemoryTool(user_id=kwargs.get("user_id", "default"))
        self.rag_tool = RAGTool(knowledge_base_path=kwargs.get("knowledge_base_path", "./kb"))

        self.context_builder = ContextBuilder(
            memory_tool=self.memory_tool,
            rag_tool=self.rag_tool,
            config=ContextConfig(max_tokens=4000)
        )

        self.conversation_history = []

    def run(self, user_input: str) -> str:
        """Run Agent, automatically build optimized context"""

        # 1. Use ContextBuilder to build optimized context
        optimized_context = self.context_builder.build(
            user_query=user_input,
            conversation_history=self.conversation_history,
            system_instructions=self.system_prompt
        )

        # 2. Call LLM with optimized context
        messages = [
            {"role": "system", "content": optimized_context},
            {"role": "user", "content": user_input}
        ]
        response = self.llm.invoke(messages)

        # 3. Update conversation history
        from hello_agents.core.message import Message
        from datetime import datetime

        self.conversation_history.append(
            Message(content=user_input, role="user", timestamp=datetime.now())
        )
        self.conversation_history.append(
            Message(content=response, role="assistant", timestamp=datetime.now())
        )

        # 4. Record important interactions to memory system
        self.memory_tool.run({
            "action": "add",
            "content": f"Q: {user_input}\nA: {response[:200]}...",  # Summary
            "memory_type": "episodic",
            "importance": 0.6
        })

        return response

# Usage example
agent = ContextAwareAgent(
    name="Data Analysis Consultant",
    llm=HelloAgentsLLM(),
    system_prompt="You are a senior Python data engineering consultant.",
    user_id="user123",
    knowledge_base_path="./data_science_kb"
)

response = agent.run("How to optimize Pandas memory usage?")
print(response)

Through this approach, ContextBuilder becomes the "context management brain" of the Agent, automatically handling information collection, filtering, and organization, allowing the Agent to always reason and generate under optimal context.

9.3.5 Best Practices and Optimization Recommendations

When actually applying ContextBuilder, the following best practices are worth noting:

1. Dynamically adjust token budget: Dynamically adjust max_tokens based on task complexity, use smaller budgets for simple tasks, increase budgets for complex tasks.

2. Relevance calculation optimization: In production environments, replace simple keyword overlap with vector similarity calculation to improve retrieval quality.

3. Caching mechanism: For unchanging system instructions and knowledge base content, implement caching mechanisms to avoid repeated calculations.

4. Monitoring and logging: Record statistical information for each context build (number of selected information, token usage rate, etc.) for subsequent optimization.

5. A/B testing: For key parameters (such as relevance weight, recency weight), find optimal configuration through A/B testing.