ユーザープロンプト ──> モデルの応答

       ┌───────────────────────────────────────┐
       │                                       │
       ▼                                       │
[ THINK ] ───> (Decision) ───> [ ACT ] ───> [ OBSERVE ]
                               (Tool Call)   (Tool Result)

エラーのログを観測する
        │
        ▼
 仮説を立てる
        │
        ▼
 コマンドを実行する
        │
        ▼
 出力を確認する
        │
        ▼
    繰り返す

agentic-pipeline/
├── config.json       # 実行時設定
├── llm_client.py     # 低レベルのHTTPクライアント
├── memory.py         # コンテキスト/状態管理
├── agent.py          # エージェントのオーケストレーションエンジン
└── main.py           # 実行時の実行ループ

{
  "llm": {
    "provider": "openai",
    "model": "gpt-4o",
    "api_key": "sk-your-api-key",
    "temperature": 0.2,
    "max_tokens": 1024 
  }
}

ペイロードをシリアライズ ──> HTTPS POSTリクエストを送信 ──> JSON応答を受信 ──> 出力を解析

import json
import urllib.request
import urllib.error
from typing import Dict, Listclass LLMClient:
    def __init__(self, config: Dict):
        self.config = config["llm"]
        self.api_key = self.config["api_key"]

    def chat_completion(
        self,
        messages: List[Dict],
        temperature: float = None
    )-> str:
        payload = {
            "model": self.config["model"],
            "messages": messages,
            "temperature": temperature or self.config.get("temperature", 0.2),
            "max_tokens": self.config.get("max_tokens", 1024)
        }

        data = json.dumps(payload).encode("utf-8")

        req = urllib.request.Request(
            "[https://api.openai.com/v1/chat/completions](https://api.openai.com/v1/chat/completions)",
            data=data,
            method="POST"
        )

        req.add_header("Content-Type", "application/json")
        req.add_header("Authorization", f"Bearer {self.api_key}")

        try:
            with urllib.request.urlopen(req) as response:
                result = json.loads(response.read().decode())
                return result["choices"][0]["message"]["content"].strip()
        except urllib.error.HTTPError as e:
            error_body = e.read().decode()
            raise Exception(f"LLM API error: {e.code} - {error_body}")

from typing import List, Dict

class AgentMemory:
    def __init__(self, max_messages: int = 20):
        self. messages: List[Dict] = []
        self. max_messages = max_messages

    def add(self, role: str, content: str):
        self. messages. append({
            "role": role,
            "content": content
        })

        if len(self. messages) > self. max_messages:
            # システムプロンプトを保持する
            system_prompt = self. messages[0]

            # 会話ウィンドウをスライドさせる
            active_history = self. messages[1:]
            self. messages = (
                [system_prompt] + 
                active_history[-(self. max_messages - 1):]
            )

    def get_messages(self) -> List[Dict]:
        return self. messages. copy()

    def clear(self):
        self. messages. clear()

from llm_client import LLMClient
from memory import AgentMemory
from typing import Dict, Callable
import json

class Agent:
    def __init__(self, system_prompt: str, config_path: str = "config.json"):
        with open(config_path) as f:
            self. config = json.load(f)
        self. llm = LLMClient(self. config)
        self. memory = AgentMemory()
        self. system_prompt = system_prompt
        self. tools: Dict[str, dict] = {}

        self. memory.add("system", system_prompt)

from agent import Agent
import time

def web_search(query: str) -> str:
    print(f"インデックスを検索中: '{query}')
    time.sleep(1)
    if "agentic ai" in query.lower():
        return (
            "見つかりました: 現代のエージェンティックなシステムは、硬いチェーンから"
            "軽量な制御ループとモジュール化されたツールへ移行しています。"
        )
    return (
        "見つかりました: ゼロからエージェントを構築すると、フレームワークによって"
        "しばしば隠されている実装の詳細が明らかになります。"
    )if __name__ == "__main__":
    system_prompt = (
        "あなたは自律的なオペレーションアシスタントです。 "
        "手順を追って考えてください。 "
        "必要に応じてツールを使用してください。 "
        "タスクが完全に完了したら、「FINAL ANSWER」というフレーズを含めてください。"
    )

    agent = Agent(system_prompt)
    agent.register_tool(
        name="search",
        func=web_search,
        description="インデックスデータベースを照会します。入力スキーマ: {'query': str}"
    )

    task = "エージェント型AIにおけるトレンドを調査し、ゼロから構築することに価値がある理由を説明してください。"
    print(f" Objective: {task}")

    max_steps = 5
    for step in range(max_steps):
        print(f"
[Cycle {step + 1}]")
        prompt = task if step == 0 else "これまでの観察を分析し、続けてください。"

        response = agent.think(prompt)
        print(f"
 Agent:
{response}")

        tool_output = agent.act(response)
        if tool_output:
            print(f"
 Observation:
{tool_output}")

        if "final answer" in response.lower():
            print("
✅ 目的の達成。")
            break

plaintext
    実行ランタイムはこのブロックを解析し、ローカルのPython web_search 関数を実行します。
*   **Observe:** ツールの出力がメモリに追加されます。これにより、モデルは推論を続けるための追加の文脈を得ます。

### サイクル 2
モデルは、元の目的、これまでの観察、ツール出力を確認します。完全な応答を統合し、次を出力します:

```text
FINAL ANSWER