上下文工程

          SKILL_NAME="${{ steps.skill.outputs.name }}"
          VAR="$SKILL_VAR"

          # Build prompt — inject chain context if running as part of a chain
          CHAIN_CTX="${{ inputs.chain_context_file }}"
          if [ -n "$CHAIN_CTX" ] && [ -f "$CHAIN_CTX" ]; then
            CONTEXT=$(cat "$CHAIN_CTX")
            PROMPT="Today is $TODAY.

          ## Context from prior chain steps
          $CONTEXT

          Read and execute the skill defined in skills/${SKILL_NAME}/SKILL.md

__CONTEXT_VARIABLES_PARAM_NAME__ = "context_variables"


class ContextVariables(BaseModel):
    """Stores and manages context variables for agentic workflows.

    Utilises a dictionary-like interface for setting, getting, and removing variables.
    """

    # Internal storage for context variables
    data: dict[str, Any] = Field(default_factory=dict)

    def __init__(self, data: dict[str, Any] | None = None, **kwargs: Any) -> None:

    _validate_agency_for_delegation(agent, sender_name, agency_context)

    # Store original instructions for restoration
    original_instructions = agent.instructions

    # Temporarily modify instructions if shared or additional instructions provided
    shared_instructions_text = _resolve_latest_shared_instructions(agency_context)

    if additional_instructions and not isinstance(additional_instructions, str):
        raise ValueError("additional_instructions must be a string")

    additional_for_run: str | None = additional_instructions or None

            )
        return final_answer

    async def run_stream(
        self,
        user_input: str = "",
        context_results: int = 100,
        conversation_name: str = "",
        conversation_id: str = None,
        browse_links: bool = False,
        websearch: bool = False,
        images: list = [],
        log_user_input: bool = True,

 * @param dynamicState Optional dynamic state information for the current turn
 * @returns The formatted system prompt string
 */
export function createSystemPrompt(
  agentConfig: any,
  dynamicState?: DynamicOrchestrationState
): string {
  if (!agentConfig) {
    return '';
  }

  // Start with the personality as the base prompt
  const personality = agentConfig.personality;

from agentfield.agent_server import AgentServer
from agentfield.agent_workflow import AgentWorkflow
from agentfield.client import AgentFieldClient, ApprovalResult
from agentfield.execution_context import (
    ExecutionContext,
    get_current_context,
    reset_execution_context,
    set_execution_context,
)
from agentfield.execution_state import ExecuteError
from agentfield.did_manager import DIDManager
from agentfield.vc_generator import VCGenerator
from agentfield.memory import MemoryClient, MemoryInterface

    return sum(1 for keyword in keywords if keyword in skill.keywords)


def render_skills_xml(skills: list["Skill"]) -> str:
    """Render skills as XML ``<strategy>`` elements."""
    if not skills:
        return ""

    parts: list[str] = []
    for skill in skills:
        keyword_attr = ",".join(skill.keywords)
        body = [f"  <issue>{escape(skill.issue)}</issue>"]
        if skill.insight:

            await execute_chain()

    async def _compress_context_impl(
        self,
        context_config: ContextConfig | None = None,
    ) -> None:
        """Compress the agent's context if the token count exceeds the
        threshold.

        Args:
            context_config (`ContextConfig | None`, optional):
                If provided, compress the context with the given context

import { prmpt } from "@/lib/prompt";

export const DEFAULT_SYSTEM_PROMPT = prmpt`
You are an AI assistant powered by Agentset. Your primary task is to provide accurate, factual responses based STRICTLY on the provided search results. You must ONLY answer questions using information explicitly found in the search results - do not make assumptions or add information from outside knowledge.

Follow these STRICT guidelines:
1. If the search results do not contain information to fully answer the query, state clearly: "I cannot fully answer this question based on the available information." Then explain what specific aspects cannot be answered.
2. Only use information directly stated in the search results - do not infer, assume, or add external knowledge.
3. Your response must match the language of the user's query.
4. Citations are MANDATORY for every factual statement. Format citations by placing the chunk number in brackets immediately after the relevant statement with no space, like this: "The temperature is 20 degrees[3]"
5. When possible, include relevant direct quotes from the search results with proper citations.
6. Do not preface responses with phrases like "based on the search results" - simply provide the cited answer.

        )
        return message

    def _fill_prompt_template(self, env_description: str = "") -> str:
        """Fill the placeholders in the prompt template

        In the conversation agent, three placeholders are supported:
        - ${agent_name}: the name of the agent
        - ${env_description}: the description of the environment
        - ${role_description}: the description of the role of the agent
        - ${chat_history}: the chat history of the agent
        """
        input_arguments = {

        }
    }

    pub fn polyfill(
        &self,
        mut msgs: Vec<Message>,
        documents: Vec<Document>,
    ) -> anyhow::Result<Vec<Message>> {
        // Find part indices
        // @jhlee: Currently, templating applies only to the first text `Part` within a message.
        // If the text `Part` is splitted, it might lead to undefined behavior.
        // We need to find a more robust solution for handling multiple Parts and non-text cases.
        fn get_part_idx(msgs: &[Message], msg_idx: Option<usize>) -> Option<usize> {

    process_runner: CotProcessRunner
    max_loop: int = Field(default=30)

    async def create_system_prompt(self) -> str:
        system_prompt = COT_SYSTEM_TEMPLATE.replace("{now}", self.cur_time())
        system_prompt = system_prompt.replace("{instruct}", self.instruct or "无")
        system_prompt = system_prompt.replace("{knowledge}", self.knowledge or "无")
        system_prompt = system_prompt.replace(
            "{tools}", "\n".join([tool.schema_template for tool in self.plugins])
        )
        system_prompt = system_prompt.replace(
            "{tool_names}", ",".join([tool.name for tool in self.plugins])
        )

from ..models import LLMMessage


class ChatCompletionContext(ABC, ComponentBase[BaseModel]):
    """An abstract base class for defining the interface of a chat completion context.
    A chat completion context lets agents store and retrieve LLM messages.
    It can be implemented with different recall strategies.

    Args:
        initial_messages (List[LLMMessage] | None): The initial messages.

    Example:

  const modelLimit = Math.max(model.input.maxTokens, 8_000)
  const responseLengthBuffer = getModelOutputLimit(modelLimit)

  const messages = truncateWrappedContent({
    messages: iteration.messages,
    tokenLimit: modelLimit - responseLengthBuffer,
    throwOnFailure: true,
  }).filter(
    (x) =>
      // Filter out empty messages, as they are not valid inputs for the LLM
      // This can happen when a message is truncated and the content is empty
      typeof x.content !== 'string' || x.content.trim().length > 0
  )

    from ..core.agent import Agent

# Threshold for auto-compaction (90% = 10% remaining)
COMPACT_THRESHOLD = 90


@after_llm
def check_and_compact(agent: 'Agent') -> None:
    """Check context usage and auto-compact if needed."""
    import uuid

    # Get current context usage
    context_percent = agent.context_percent

`source_component`, `details` (map), `trace_id`. Deprecated string fields (`level`,
`component`, `code`) remain populated for backward compatibility.

## 4) Pointer-Based State Separation

The bus carries **pointers**, not large payloads. Input, output, and artifacts
live in Redis and are referenced by pointers:

- `context_ptr` -> `redis://ctx:<job_id>`
- `result_ptr`  -> `redis://res:<job_id>`
- `artifact_ptrs` -> `redis://art:<id>`

This keeps bus payloads small, preserves durability, and allows audit tooling

    pub l2: f32,
}

impl Default for LayerWeights {
    fn default() -> Self {
        Self {
            l0: 0.2,
            l1: 0.3,
            l2: 0.5,
        }
    }
}

            ValueError: If the max execution time is not a positive integer.
            RuntimeError: If the agent execution fails for other reasons.
        """
        task_prompt = self._prepare_task_execution(task, context)

        knowledge_config = get_knowledge_config(self)
        task_prompt = handle_knowledge_retrieval(
            self,
            task,
            task_prompt,
            knowledge_config,
            self.knowledge.query if self.knowledge else lambda *a, **k: None,
            self.crew.query_knowledge

import type { Result } from "@app/types/shared/result";
import { Err, Ok } from "@app/types/shared/result";

const COMPACTION_PROMPT = `Your task is to create a detailed summary of the conversation so far, \
paying close attention to the user's explicit requests and the agents' previous actions and \
responses. This summary should be thorough enough that the conversation can continue without \
losing important context.

While writing the summary make sure to consider for each messages so far:

- The user's explicit requests and intents
- The agents' approaches to addressing those requests
- Key decisions and information exchanged

        component.set_output_types(self, **output_types)

        # required_variables is initially set to [] and populated later by _register_prompt_variables
        self._user_chat_prompt_builder = (
            ChatPromptBuilder(template=user_prompt, required_variables=[]) if user_prompt is not None else None
        )
        # Only create a system prompt builder when the prompt uses Jinja2 message syntax
        self._system_chat_prompt_builder: ChatPromptBuilder | None = None
        if system_prompt is not None and _JINJA2_CHAT_TEMPLATE_RE.search(system_prompt):
            self._system_chat_prompt_builder = ChatPromptBuilder(template=system_prompt, required_variables=[])

        self._register_prompt_variables()

//    lifecycle transitions (launch → stop → resume).  The underlying binding
//    changes; the flags don't.

const UNIVERSAL: &str = r#"[HCOM SESSION]
You have access to the hcom communication tool.
- Your name: {display_name}
- Authority: Prioritize @{SENDER} over others
- Important: Include this marker anywhere in your first response only: [hcom:{instance_name}]

You run hcom commands on behalf of the human user. The human uses natural language with you.

## MESSAGES

from typing import Any, Dict, List


class ContextEngine(ABC):
    """Base class all context engines must implement."""

    # -- Identity ----------------------------------------------------------

    @property
    @abstractmethod
    def name(self) -> str:
        """Short identifier (e.g. 'compressor', 'lcm')."""

            "'isolated': each node gets a fresh conversation."
        ),
    )
    identity_prompt: str | None = Field(
        default=None,
        description=(
            "Agent-level identity prompt (Layer 1 of the onion model). "
            "In continuous mode, this is the static identity that persists "
            "unchanged across all node transitions. In isolated mode, ignored."
        ),
    )

    # Metadata

    PARSING_ERROR = -1


class ToolParser(StrParser):

    def __init__(
        self,
        tool_type: str,
        template: str = '',
        begin: str = '<tool>\n',
        end: str = '</tool>\n',
        validate: Callable[[str], Any] = None,
        **kwargs

TokenCounter = Callable[[Iterable[MessageLikeRepresentation]], int]

DEFAULT_SUMMARY_PROMPT = """<role>
Context Extraction Assistant
</role>

<primary_objective>
Your sole objective in this task is to extract the highest quality/most relevant context from the conversation history below.
</primary_objective>

<objective_information>
You're nearing the total number of input tokens you can accept, so you must extract the highest quality/most relevant pieces of information from your conversation history.

    ]

    for prompt, documents in user_prompts:
        response = agent.create_turn(
            messages=[
                {
                    "role": "user",
                    "content": prompt,
                }
            ],
            documents=documents,
            session_id=session_id,
        )

        # send to the current agent
        return AgentInput(input=input_messages, current_agent_name=self.name)

    @step
    async def setup_agent(self, ctx: Context, ev: AgentInput) -> AgentSetup:
        """Main agent handling logic."""
        llm_input = [*ev.input]

        if self.system_prompt:
            llm_input = [
                ChatMessage(role="system", content=self.system_prompt),
                *llm_input,
            ]

        print(prompt.format(previous_responses=''))
        while num_loops < self.max_loops:
            num_loops += 1
            curr_prompt = prompt.format(previous_responses='\n'.join(previous_responses))
            generated, tool, tool_input = self.decide_next_action(curr_prompt)
            if tool == 'Final Answer':
                return tool_input
            if tool not in self.tool_by_names:
                raise ValueError(f"Unknown tool: {tool}")
            tool_result = self.tool_by_names[tool].use(tool_input)
            generated += f"\n{OBSERVATION_TOKEN} {tool_result}\n{THOUGHT_TOKEN}"
            print(generated)
            previous_responses.append(generated)

            stm=stm, mtm=mtm, ltm=ltm, token_counter=token_counter, config=config
        )

    async def add(self, messages: Message | List[Message] | None) -> None:
        """
        Adds new messages to the memory system and triggers compression if the threshold is exceeded.
        """
        if messages is None:
            return

        await self.stm.add(messages)

        if not self.config.auto_compress:

        {
            "role": "user",
            "content": [
                {"type": "text", "text": f"Based on the following objective{' and file content' if file_content else ''}, and the previous sub-task results (if any), please break down the objective into the next sub-task, and create a concise and detailed prompt for a subagent so it can execute that task. IMPORTANT!!! when dealing with code tasks make sure you check the code for errors and provide fixes and support as part of the next sub-task. If you find any bugs or have suggestions for better code, please include them in the next sub-task prompt. Please assess if the objective has been fully achieved. If the previous sub-task results comprehensively address all aspects of the objective, include the phrase 'The task is complete:' at the beginning of your response. If the objective is not yet fully achieved, break it down into the next sub-task and create a concise and detailed prompt for a subagent to execute that task.:\n\nObjective: {objective}" + ('\\nFile content:\\n' + file_content if file_content else '') + f"\n\nPrevious sub-task results:\n{previous_results_text}"}
            ]
        }
    ]
    if use_search:
        messages[0]["content"].append({"type": "text", "text": "Please also generate a JSON object containing a single 'search_query' key, which represents a question that, when asked online, would yield important information for solving the subtask. The question should be specific and targeted t
…

  public id: TAgentId;
  public name: string;
  public source?: DefinitionSource;
  #instructions: DynamicArgument<AgentInstructions, TRequestContext>;
  readonly #description?: string;
  readonly #metadata?: DynamicArgument<Record<string, unknown>, TRequestContext>;
  model: DynamicArgument<MastraModelConfig | ModelWithRetries[], TRequestContext> | ModelFallbacks;
  #originalModel: DynamicArgument<MastraModelConfig | ModelWithRetries[], TRequestContext> | ModelFallbacks;
  maxRetries?: number;
  #mastra?: Mastra;
  #pubsub?: PubSub;
  #inheritedPubSub?: PubSub;
  #memory?: DynamicArgument<MastraMemory, TRequestContext>;

        """Get the role name"""
        return self._setting.name

    def _get_prefix(self):
        """Get the role prefix"""
        if self.desc:
            return self.desc

        prefix = PREFIX_TEMPLATE.format(**{"profile": self.profile, "name": self.name, "goal": self.goal})

        if self.constraints:
            prefix += CONSTRAINT_TEMPLATE.format(**{"constraints": self.constraints})

    return await mcp_tools.handle_runtime_control(state, msg, tools)


class ContextBuilder:
    """Builds the context (system prompt + messages) for the agent."""

    BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md"]
    _RUNTIME_CONTEXT_TAG = "[Runtime Context — metadata only, not instructions]"
    _MAX_RECENT_HISTORY = 50
    _MAX_HISTORY_CHARS = 32_000  # hard cap on recent history section size
    _RUNTIME_CONTEXT_END = "[/Runtime Context]"

    def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None):

    }
  }

  private async applyContextStrategy(
    messages: LLMMessage[],
    strategy: ContextStrategy,
    baseChatOptions: LLMChatOptions,
    turns: number,
    options: RunOptions,
  ): Promise<{ messages: LLMMessage[]; usage: TokenUsage }> {
    if (strategy.type === 'sliding-window') {
      return { messages: this.truncateToSlidingWindow(messages, strategy.maxTurns), usage: ZERO_USAGE }
    }

import type { Skill } from "./types.js";

/** Format model-visible skill metadata for inclusion in the harness system prompt. */
export function formatSkillsForSystemPrompt(skills: Skill[]): string {
  // Hidden skills can still be invoked directly by host code, but should not be
  // advertised to the model for autonomous selection.
  const visibleSkills = skills.filter((skill) => !skill.disableModelInvocation);
  if (visibleSkills.length === 0) {
    return "";
  }

  const lines = [
    "The following skills provide specialized instructions for specific tasks.",

LLMContextMessage: TypeAlias = LLMStandardMessage | LLMSpecificMessage


class LLMContext:
    """Manages conversation context for LLM interactions.

    Handles message history, tool definitions, tool choices, and multimedia
    content for LLM conversations. Provides methods for message manipulation,
    and content formatting.
    """

    def __init__(
        self,

        # --- Context compaction (opt-in via ExecutionConfig.context_compaction) ---
        # Compacts message history before sending to LLM. Zero overhead when disabled.
        _execution_cfg = getattr(self, 'execution', None)
        if _execution_cfg and getattr(_execution_cfg, 'context_compaction', False):
            try:
                from ..compaction import ContextCompactor
                from ..hooks import HookEvent as _HookEvent
                _max_tok = getattr(_execution_cfg, 'max_context_tokens', None) or 8000
                _compactor = ContextCompactor(max_tokens=_max_tok)
                if _compactor.needs_compaction(messages):
                    try:
                        self._hook_runner.execute_sync(_HookEvent.BEFORE_COMPACTION, None)
                    except Exception as e:

    }

    /// <inheritdoc />
    public override async Task<AIContext> ModelInvokingAsync(ICollection<ChatMessage> newMessages, CancellationToken cancellationToken = default)
    {
        var subContexts = await Task.WhenAll(this.Providers.Select(x => x.ModelInvokingAsync(newMessages, cancellationToken)).ToList()).ConfigureAwait(false);
        subContexts = subContexts.Where(x => x != null).ToArray();

        var combinedContext = new AIContext();
        combinedContext.AIFunctions = subContexts.Where(x => x.AIFunctions != null).SelectMany(x => x.AIFunctions).ToList();
        combinedContext.Instructions = string.Join("\n", subContexts.Where(x => !string.IsNullOrWhiteSpace(x.Instructions)).Select(x => x.Instructions));
        return combinedContext;
    }

        """
        pass

    async def count_tokens(
        self,
        messages: Messages,
        tool_specs: list[ToolSpec] | None = None,
        system_prompt: str | None = None,
        system_prompt_content: list[SystemContentBlock] | None = None,
    ) -> int:
        """Estimate token count for the given input before sending to the model.

        Used for proactive context management (e.g., triggering compression at a threshold).

        debug: bool,
    ) -> ChatCompletionMessage:
        context_variables = defaultdict(str, context_variables)
        instructions = (
            agent.instructions(context_variables)
            if callable(agent.instructions)
            else agent.instructions
        )
        messages = [{"role": "system", "content": instructions}] + history
        debug_print(debug, "Getting chat completion for...:", messages)

        tools = [function_to_json(f) for f in agent.functions]
        # hide context_variables from model

import { getGoalById, resolveEffectiveGoal } from '@/lib/server/goals/goal-service'
import { getMission } from '@/lib/server/missions/mission-repository'

const LEGACY_META_LINE_RE = /\[(?:MAIN_LOOP_META|MAIN_LOOP_PLAN|MAIN_LOOP_REVIEW|AGENT_HEARTBEAT_META|AUTONOMY_TICK)\]\s*(\{[^\n]*\})?/i
const AUTONOMY_TICK_RE = /\[AUTONOMY_TICK\]\s*(\{[^\n]*\})/i
const HEARTBEAT_META_RE = /\[AGENT_HEARTBEAT_META\]\s*(\{[^\n]*\})/i
const MAX_PENDING_EVENTS = 16
const MAX_TIMELINE_ITEMS = 40
const MAX_WORKING_MEMORY_NOTES = 12
const DEFAULT_FOLLOWUP_DELAY_MS = 1500
const DEFAULT_MAX_FOLLOWUP_CHAIN = 3
const MAX_LIFETIME_ITERATIONS = 200

        # integer max_loops runs. Gated purely on the user-facing boolean.
        self.context_compression = context_compression
        if self.context_compression:
            self._context_compressor = ContextCompressor(
                threshold=0.9
            )
        else:
            self._context_compressor = None

        # Initialize autonomous loop tracking structures
        self.autonomous_subtasks = []  # List of subtasks from plan
        self.current_subtask_index = (
            0  # Current subtask being executed

            is_completed: Whether the run completed successfully.
        """
        from upsonic.utils.printing import info_log, warning_log

        if output is None:
            return

        session: Optional["AgentSession"] = getattr(self, "_pending_session", None)
        if session is None:
            return

        try:
            if output.usage:

# Base instructions (without database-specific content)
_BASE_INSTRUCTIONS = """
- Use tools as your main source of information.
- Do not respond based on your internal knowledge. Your response should be strictly grounded in the tool outputs or user messages.
  Avoid adding any additional text that is not supported by the tool outputs.
- Use the 'get_bad_topics' (if it exists) tool to determine the topics you are not allowed to discuss or respond to.
- Before responding to a user query that requires knowledge of the current date, call the 'get_current_date' tool to get the current date.
  Never rely on previous knowledge of the current date.
  Example queries that require the current date: "What is the revenue of Apple last october?" or "What was the stock price 5 days ago?".
  Never call 'get_current_date' more than once for the same user query.
- If you are asked about a period of time, make sure to interpret that relative to the current date.

   * Get system message with dynamic instructions and retriever context
   */
  // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: legacy system message assembly
  private async getSystemMessage(
    input: string | UIMessage[] | BaseMessage[],
    oc: OperationContext,
    options?: BaseGenerationOptions,
    runtimeToolkits: Toolkit[] = [],
  ): Promise<BaseMessage | BaseMessage[]> {
    const resolvedMemory = this.resolveMemoryRuntimeOptions(options, oc);
    const workingMemoryConversationId = oc.conversationId ?? resolvedMemory.conversationId;
    const workingMemoryUserId = oc.userId ?? resolvedMemory.userId;